diff --git a/README.md b/README.md index ac48251..3b2874e 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ authentication ### Functional Areas * [Authentication](authentication) provides samples showing the various methods by which users can authenticate against the Visier platform. +* [Administration](administration) provides samples showing how to use Administration APIs for tenant configuration workflows (SDLC). * [Data Intake](data-intake) provides samples showing how to use APIs to send data to your Visier tenant. ### Before You Begin diff --git a/administration/README.md b/administration/README.md new file mode 100644 index 0000000..cceb01d --- /dev/null +++ b/administration/README.md @@ -0,0 +1,7 @@ +# Administration + +Samples that use Visier Administration APIs for tenant configuration workflows (SDLC). + +## Samples + +* [SDLC promotion sample](sdlc-promotion/python) shows a guarded Python workflow for promoting published configuration from a source tenant into a target tenant draft (SDLC), with publishing available only when explicitly requested. diff --git a/administration/sdlc-promotion/python/.gitignore b/administration/sdlc-promotion/python/.gitignore new file mode 100644 index 0000000..b8683bd --- /dev/null +++ b/administration/sdlc-promotion/python/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +*.pyc +config.yaml +.env diff --git a/administration/sdlc-promotion/python/README.md b/administration/sdlc-promotion/python/README.md new file mode 100644 index 0000000..bb42291 --- /dev/null +++ b/administration/sdlc-promotion/python/README.md @@ -0,0 +1,68 @@ +# Visier SDLC Promotion Sample + +Sample Python workflow for promoting Visier configuration from a **source** tenant to a **target** tenant using Visier's public Administration APIs and the official `visier-platform-sdk`. + +The guarded default is a preview: authenticate the source tenant, read published history, choose the export range, and show the target draft metadata that would be used. No ZIP export, target draft creation, import, or publish happens unless you explicitly request it. + +## Requirements + +- Python 3.9+ +- Access to source and target tenants with the required Administration API capabilities + +## Install + +```bash +python -m venv .venv +source .venv/bin/activate # Windows: .venv\Scripts\activate +pip install -r requirements.txt +``` + +Dependencies: `visier-platform-sdk`, `PyYAML`, and `python-dotenv`. + +## Configure + +1. Copy `config.yaml.example` to `config.yaml`. +2. Set `profiles.source` and `profiles.target` with each tenant's `base_url`, `api_key`, `username`, and `password`. +3. Prefer `${ENV_VAR}` placeholders in YAML and keep secrets in the environment or a `.env` file next to `config.yaml`. + +The loader only reads `.env` from the config file's directory to avoid mixing credentials from another working directory. It rejects non-HTTPS base URLs and rejects configs where `source` and `target` clearly point to the same tenant. + +## Promotion Modes + +| Command | Effect | +| --- | --- | +| `python promote.py` | Preview only. Authenticates source, lists source history, chooses export ids, and prints planned target draft metadata. | +| `python promote.py --apply` | Applies target changes. Exports the ZIP, creates a target draft, and imports commits. Does not publish. | +| `python promote.py --apply --publish` | Applies target changes and runs `commitAndPublish`. This changes the target tenant's published configuration. | + +`--publish` requires `--apply`. The old `--dry-run` flag is deprecated and now behaves like the default preview mode. + +## Export Scope + +| Flag | Meaning | +| --- | --- | +| `--mode latest` | Export only the most recent published version from the source. This is the default. | +| `--mode all` | Export full published history in one ZIP using a paginated source listing and an oldest-to-newest range. | +| `--config PATH` | Config YAML path. Defaults to `config.yaml` in the current working directory. | + +## Workflow Steps + +| Step | Tenant | What happens | +| --- | --- | --- | +| 0 | Source, and target only with `--apply` | Obtain a secure token through `visier-platform-sdk`. | +| 1 | Source | List published production versions with `GET /v1/admin/production-versions`. | +| 2 | Source | With `--apply`, export selected production versions as a ZIP. | +| 3 | Target | With `--apply`, create a new draft project. | +| 4 | Target | With `--apply`, upload the ZIP into the draft. | +| 5 | Target | With `--apply --publish`, commit and publish the draft. | + +## Output And Logging + +Normal CLI output is intentionally minimal: selected version ids, planned draft name/description, target draft id/name when created, ZIP size, and publish status. It does not print full API responses. + +Set `VISIER_SDLC_DEBUG=1` only while troubleshooting draft metadata. Debug logging still avoids request credentials and full response dumps. + +## API References + +- [Production versions](https://docs.visier.com/developer/apis/administration/production-versions/code-samples.htm) +- [Projects](https://docs.visier.com/developer/apis/administration/projects/fundamentals/code-samples.htm) diff --git a/administration/sdlc-promotion/python/config.yaml.example b/administration/sdlc-promotion/python/config.yaml.example new file mode 100644 index 0000000..b0d1eb8 --- /dev/null +++ b/administration/sdlc-promotion/python/config.yaml.example @@ -0,0 +1,28 @@ +# Copy to config.yaml and fill in real values. Never commit secrets. +# +# Profiles are named by *role* in the SDLC workflow (not your business environment names): +# source — tenant you export published work *from* +# target — tenant you import into and publish *to* +# +# Optional: use ${ENV_VAR} placeholders — values are read from the process +# environment or from a .env file in this config file's directory. +# The sample defaults to preview mode. Use --apply to create/import a target +# draft, and --apply --publish only when you are ready to publish target changes. + +profiles: + source: + # e.g. https://.api.visier.io (no trailing slash) + base_url: "https://your-source-vanity.api.visier.io" + api_key: "${VISIER_SOURCE_API_KEY}" + username: "${VISIER_SOURCE_USERNAME}" + password: "${VISIER_SOURCE_PASSWORD}" + # Optional — Visier header TargetTenantID (analytic tenant), if your setup requires it + # target_tenant_id: "WFF_j1r~c7o" + # Optional — if your tenant requires vanity on the token request + # vanity: "your-source-vanity" + + target: + base_url: "https://your-target-vanity.api.visier.io" + api_key: "${VISIER_TARGET_API_KEY}" + username: "${VISIER_TARGET_USERNAME}" + password: "${VISIER_TARGET_PASSWORD}" diff --git a/administration/sdlc-promotion/python/promote.py b/administration/sdlc-promotion/python/promote.py new file mode 100644 index 0000000..db70362 --- /dev/null +++ b/administration/sdlc-promotion/python/promote.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +""" +CLI: load ``config.yaml`` and run a guarded SDLC promotion sample. + +Usage (from this directory):: + + pip install -r requirements.txt + source .venv/bin/activate # if you use a virtual environment + python promote.py # preview only; no target changes + python promote.py --apply # create/import target draft; do not publish + python promote.py --apply --publish + +Troubleshooting draft project **name** vs API: set ``VISIER_SDLC_DEBUG=1`` and run again; +look for log lines tagged ``[SDLC troubleshooting]``. Unset the variable when done. +""" + +from __future__ import annotations + +import argparse +import logging +import sys + +from visier_sdlc import ( + NewDraftProject, + PromotionRequest, + VisierClient, + get_profile, + load_config, + run_sdlc_promotion, +) +from visier_sdlc.exceptions import VisierSDLCError + + +def _parse_args(argv: list[str] | None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Run Visier SDLC promotion: export from the source tenant, import on the target tenant, " + "and publish only when explicitly requested." + ), + ) + parser.add_argument( + "--mode", + choices=("latest", "all"), + default="latest", + help=( + "What to export from the source tenant's published history: " + "'latest' = only the most recent published version; " + "'all' = full version history in one export (paginated list, oldest→newest range)." + ), + ) + parser.add_argument( + "--config", + default="config.yaml", + metavar="PATH", + help="Path to config YAML (default: config.yaml).", + ) + parser.add_argument( + "--apply", + action="store_true", + help="Apply target changes: export the ZIP, create a target draft, and import commits. Does not publish.", + ) + parser.add_argument( + "--publish", + action="store_true", + help="Commit and publish the imported target draft. Requires --apply.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help=argparse.SUPPRESS, + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(argv) + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + if args.publish and not args.apply: + print("--publish requires --apply.", file=sys.stderr) + return 1 + if args.dry_run: + print( + "--dry-run is deprecated and now behaves like the default preview mode. " + "Use --apply to create/import a draft, or --apply --publish to publish.", + file=sys.stderr, + ) + + try: + cfg = load_config(args.config) + except FileNotFoundError: + print( + f"Config not found: {args.config!r}\n" + "Copy config.yaml.example to config.yaml and add credentials, " + "or pass --config /path/to/config.yaml", + file=sys.stderr, + ) + return 1 + except (ImportError, KeyError, ValueError) as e: + print(f"Configuration error: {e}", file=sys.stderr) + return 1 + + try: + source = VisierClient(get_profile(cfg, "source")) + target = VisierClient(get_profile(cfg, "target")) + except (ImportError, KeyError, ValueError) as e: + print(f"Setup error: {e}", file=sys.stderr) + return 1 + + full_history = args.mode == "all" + # name/description are set in the workflow from source vanity + base URL (and, for + # --mode latest, the exported version name and id) when auto_draft_naming is set. + request = PromotionRequest( + new_project=NewDraftProject( + name="", + description="", + release_version="1", + ticket_number="", + version_number=0, + ), + export_full_published_history=full_history, + auto_draft_naming=args.mode, + apply_to_target=args.apply, + publish=args.publish, + ) + logging.info( + "Mode: %s (full_published_history=%s, apply=%s, publish=%s)", + args.mode, + full_history, + args.apply, + args.publish, + ) + + try: + result = run_sdlc_promotion(source, target, request) + except VisierSDLCError as e: + print(f"Aborted: {e}", file=sys.stderr) + return 2 + + print("Exported version ids:", result.export_start_version_id, result.export_end_version_id) + print("Planned target draft name:", result.planned_project.name) + print("Planned target draft description:", result.planned_project.description) + if result.new_project is None: + print("Preview only: no ZIP export, target draft creation, import, or publish was run.") + return 0 + + print("Export ZIP size bytes:", result.export_zip_size_bytes) + print("Target draft id:", result.new_project.get("id")) + print("Target draft name:", result.new_project.get("name")) + if result.publish_result is not None: + print("Publish (Step 5): completed.") + else: + print("Publish (Step 5): skipped. Inspect the target draft before publishing.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/administration/sdlc-promotion/python/requirements.txt b/administration/sdlc-promotion/python/requirements.txt new file mode 100644 index 0000000..5c01da8 --- /dev/null +++ b/administration/sdlc-promotion/python/requirements.txt @@ -0,0 +1,6 @@ +# Official Visier SDK used by this sample +visier-platform-sdk>=22222222.99201.2630 +# Optional: load profiles from config.yaml +PyYAML>=6.0 +# Optional: load secrets from a local .env file (see config.py) +python-dotenv>=1.0.0 diff --git a/administration/sdlc-promotion/python/tests/test_config.py b/administration/sdlc-promotion/python/tests/test_config.py new file mode 100644 index 0000000..5956772 --- /dev/null +++ b/administration/sdlc-promotion/python/tests/test_config.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +import importlib.util +from pathlib import Path + +from visier_sdlc.config import load_config + + +@unittest.skipUnless(importlib.util.find_spec("yaml"), "PyYAML is not installed") +class ConfigTests(unittest.TestCase): + def test_load_config_rejects_same_source_and_target(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.yaml" + path.write_text( + """ +profiles: + source: + base_url: "https://same.api.visier.io" + api_key: "key" + username: "user" + password: "pass" + target: + base_url: "https://same.api.visier.io" + api_key: "key" + username: "user" + password: "pass" +""", + encoding="utf-8", + ) + + with self.assertRaisesRegex(ValueError, "same tenant"): + load_config(path) + + def test_load_config_requires_https(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.yaml" + path.write_text( + """ +profiles: + source: + base_url: "http://source.api.visier.io" + api_key: "key" + username: "user" + password: "pass" + target: + base_url: "https://target.api.visier.io" + api_key: "key" + username: "user" + password: "pass" +""", + encoding="utf-8", + ) + + with self.assertRaisesRegex(ValueError, "https"): + load_config(path) + + @unittest.skipUnless(importlib.util.find_spec("dotenv"), "python-dotenv is not installed") + def test_env_file_loaded_only_from_config_directory(self) -> None: + with tempfile.TemporaryDirectory() as cwd, tempfile.TemporaryDirectory() as cfg_dir: + cwd_path = Path(cwd) + cfg_path = Path(cfg_dir) + (cwd_path / ".env").write_text("VISIER_SOURCE_PASSWORD=wrong\n", encoding="utf-8") + (cfg_path / ".env").write_text("VISIER_SOURCE_PASSWORD=right\n", encoding="utf-8") + config_file = cfg_path / "config.yaml" + config_file.write_text( + """ +profiles: + source: + base_url: "https://source.api.visier.io" + api_key: "key" + username: "user" + password: "${VISIER_SOURCE_PASSWORD}" + target: + base_url: "https://target.api.visier.io" + api_key: "key" + username: "user" + password: "pass" +""", + encoding="utf-8", + ) + old_cwd = Path.cwd() + old_env = os.environ.pop("VISIER_SOURCE_PASSWORD", None) + try: + os.chdir(cwd_path) + cfg = load_config(config_file) + finally: + os.chdir(old_cwd) + if old_env is not None: + os.environ["VISIER_SOURCE_PASSWORD"] = old_env + else: + os.environ.pop("VISIER_SOURCE_PASSWORD", None) + + self.assertEqual(cfg["source"].password, "right") + + +if __name__ == "__main__": + unittest.main() diff --git a/administration/sdlc-promotion/python/tests/test_workflow.py b/administration/sdlc-promotion/python/tests/test_workflow.py new file mode 100644 index 0000000..607b60c --- /dev/null +++ b/administration/sdlc-promotion/python/tests/test_workflow.py @@ -0,0 +1,157 @@ +from __future__ import annotations + +import unittest + +from visier_sdlc.client import NewDraftProject +from visier_sdlc.config import EnvironmentProfile +from visier_sdlc.workflow import ( + PromotionRequest, + build_auto_draft_name_and_description, + pick_export_version_ids, + run_sdlc_promotion, +) + + +class FakeClient: + def __init__(self, profile: EnvironmentProfile) -> None: + self.profile = profile + self.calls: list[str] = [] + + def authenticate(self) -> str: + self.calls.append("authenticate") + return "token" + + def get_production_versions(self, *, limit: int = 400, start: int = 0) -> dict: + self.calls.append(f"get_production_versions:{limit}:{start}") + return { + "publishedVersions": [ + {"id": "newest", "name": "Newest project"}, + {"id": "oldest", "name": "Oldest project"}, + ] + } + + def get_all_published_production_versions(self, *, page_size: int = 400) -> dict: + self.calls.append(f"get_all_published_production_versions:{page_size}") + return self.get_production_versions() + + def export_production_versions_zip(self, start_version: str, end_version: str, excluded_versions=None) -> bytes: + self.calls.append(f"export:{start_version}:{end_version}") + return b"PKzip" + + def create_draft_project(self, spec: NewDraftProject) -> dict: + self.calls.append(f"create:{spec.name}") + return {"id": "draft-1", "name": spec.name} + + def import_commits(self, project_id: str, zip_bytes: bytes) -> dict: + self.calls.append(f"import:{project_id}:{len(zip_bytes)}") + return {"imported": True} + + def commit_and_publish(self, project_id: str) -> dict: + self.calls.append(f"publish:{project_id}") + return {"published": True} + + +def _profile(base_url: str) -> EnvironmentProfile: + return EnvironmentProfile( + base_url=base_url, + api_key="key", + username="user", + password="pass", + ) + + +class WorkflowTests(unittest.TestCase): + def test_preview_does_not_export_or_mutate_target(self) -> None: + source = FakeClient(_profile("https://source.api.visier.io")) + target = FakeClient(_profile("https://target.api.visier.io")) + + result = run_sdlc_promotion( + source, + target, + PromotionRequest( + new_project=NewDraftProject(name="preview"), + auto_draft_naming=None, + ), + ) + + self.assertEqual(result.export_start_version_id, "newest") + self.assertEqual(result.export_end_version_id, "newest") + self.assertEqual(result.planned_project.name, "preview") + self.assertIsNone(result.new_project) + self.assertIsNone(result.import_result) + self.assertIsNone(result.publish_result) + self.assertEqual(source.calls, ["authenticate", "get_production_versions:400:0"]) + self.assertEqual(target.calls, []) + + def test_apply_imports_but_does_not_publish(self) -> None: + source = FakeClient(_profile("https://source.api.visier.io")) + target = FakeClient(_profile("https://target.api.visier.io")) + + result = run_sdlc_promotion( + source, + target, + PromotionRequest( + new_project=NewDraftProject(name="apply"), + apply_to_target=True, + ), + ) + + self.assertEqual(result.new_project, {"id": "draft-1", "name": "apply"}) + self.assertEqual(result.import_result, {"imported": True}) + self.assertIsNone(result.publish_result) + self.assertIn("export:newest:newest", source.calls) + self.assertEqual(target.calls, ["authenticate", "create:apply", "import:draft-1:5"]) + + def test_publish_requires_apply_and_runs_step_five_when_enabled(self) -> None: + source = FakeClient(_profile("https://source.api.visier.io")) + target = FakeClient(_profile("https://target.api.visier.io")) + + with self.assertRaisesRegex(Exception, "publish=True requires apply_to_target=True"): + run_sdlc_promotion( + source, + target, + PromotionRequest(new_project=NewDraftProject(name="bad"), publish=True), + ) + + result = run_sdlc_promotion( + source, + target, + PromotionRequest( + new_project=NewDraftProject(name="publish"), + apply_to_target=True, + publish=True, + ), + ) + + self.assertEqual(result.publish_result, {"published": True}) + self.assertIn("publish:draft-1", target.calls) + + def test_pick_export_version_ids_full_history_uses_oldest_to_newest(self) -> None: + history = { + "publishedVersions": [ + {"id": "newest"}, + {"id": "middle"}, + {"id": "oldest"}, + ] + } + request = PromotionRequest( + new_project=NewDraftProject(name="x"), + export_full_published_history=True, + ) + + self.assertEqual(pick_export_version_ids(history, request), ("oldest", "newest")) + + def test_auto_draft_name_is_limited(self) -> None: + name, description = build_auto_draft_name_and_description( + "latest", + _profile("https://source-vanity.api.visier.io"), + {"publishedVersions": [{"id": "v1", "name": "A" * 100}]}, + "v1", + ) + + self.assertLessEqual(len(name), 50) + self.assertLessEqual(len(description), 150) + + +if __name__ == "__main__": + unittest.main() diff --git a/administration/sdlc-promotion/python/visier_sdlc/__init__.py b/administration/sdlc-promotion/python/visier_sdlc/__init__.py new file mode 100644 index 0000000..a1b5a55 --- /dev/null +++ b/administration/sdlc-promotion/python/visier_sdlc/__init__.py @@ -0,0 +1,45 @@ +""" +Lightweight helpers for promoting Visier configuration from a *source* tenant to a +*target* tenant using the public Administration APIs. +""" + +from visier_sdlc.client import NewDraftProject, VisierClient +from visier_sdlc.config import EnvironmentProfile, get_profile, load_config +from visier_sdlc.exceptions import VisierSDLCError +from visier_sdlc.workflow import ( + PromotionRequest, + PromotionResult, + build_auto_draft_name_and_description, + pick_export_version_ids, + run_dev_to_production_sdlc, + run_sdlc_promotion, + source_tenant_vanity_label, + step0_get_secure_token, + step1_list_production_history, + step2_download_production_export_zip, + step3_create_blank_project_on_target, + step4_import_zip_into_target_project, + step5_publish_target_project, +) + +__all__ = [ + "build_auto_draft_name_and_description", + "EnvironmentProfile", + "NewDraftProject", + "PromotionRequest", + "PromotionResult", + "pick_export_version_ids", + "VisierClient", + "VisierSDLCError", + "get_profile", + "load_config", + "source_tenant_vanity_label", + "run_dev_to_production_sdlc", + "run_sdlc_promotion", + "step0_get_secure_token", + "step1_list_production_history", + "step2_download_production_export_zip", + "step3_create_blank_project_on_target", + "step4_import_zip_into_target_project", + "step5_publish_target_project", +] diff --git a/administration/sdlc-promotion/python/visier_sdlc/client.py b/administration/sdlc-promotion/python/visier_sdlc/client.py new file mode 100644 index 0000000..b810392 --- /dev/null +++ b/administration/sdlc-promotion/python/visier_sdlc/client.py @@ -0,0 +1,277 @@ +""" +SDK-backed helpers for the Administration APIs used by the promotion sample. + +This module intentionally delegates HTTP, authentication, generated DTOs, and error +handling to ``visier-platform-sdk``. The orchestration code in this repository should +teach the promotion sequence, not become a second Python SDK. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any + +from visier_sdlc.config import EnvironmentProfile, visier_sdlc_troubleshoot_logging +from visier_sdlc.exceptions import VisierSDLCError + +logger = logging.getLogger(__name__) + +# Default timeout (seconds) for HTTP calls (exports can be large) +_DEFAULT_TIMEOUT = 120 + + +@dataclass +class NewDraftProject: + """Fields for ``POST /v1/admin/projects`` when creating a blank draft project.""" + + name: str + description: str = "" + release_version: str = "1" + ticket_number: str = "" + version_number: int = 0 + + +def _load_sdk() -> dict[str, Any]: + try: + from visier_platform_sdk import ( # type: ignore[import-not-found] + ApiClient, + Configuration, + ExportProductionVersionsAPIOperationParametersDTO, + ProductionVersionsApi, + ProductionVersionsAPIOperationRequestDTO, + ProjectDTO, + ProjectOperationRequestDTO, + ProjectsApi, + ) + from visier_platform_sdk.exceptions import ApiException # type: ignore[import-not-found] + except ImportError as e: # pragma: no cover - depends on caller environment + raise ImportError( + "Install visier-platform-sdk to run this sample: " + "pip install -r requirements.txt" + ) from e + return { + "ApiClient": ApiClient, + "Configuration": Configuration, + "ExportProductionVersionsAPIOperationParametersDTO": ExportProductionVersionsAPIOperationParametersDTO, + "ProductionVersionsApi": ProductionVersionsApi, + "ProductionVersionsAPIOperationRequestDTO": ProductionVersionsAPIOperationRequestDTO, + "ProjectDTO": ProjectDTO, + "ProjectOperationRequestDTO": ProjectOperationRequestDTO, + "ProjectsApi": ProjectsApi, + "ApiException": ApiException, + } + + +def _dto_to_dict(value: Any) -> dict[str, Any]: + """Return a plain dict from generated SDK DTOs or dict-like fake test objects.""" + if value is None: + return {} + if isinstance(value, dict): + return value + if hasattr(value, "to_dict"): + data = value.to_dict() + return data if isinstance(data, dict) else {} + if hasattr(value, "model_dump"): + data = value.model_dump(by_alias=True, exclude_none=True) + return data if isinstance(data, dict) else {} + public = {k: v for k, v in vars(value).items() if not k.startswith("_")} + return public + + +def _safe_api_error_message(exc: Exception) -> str: + """Surface status/reason and a short body snippet without dumping headers or tokens.""" + status = getattr(exc, "status", None) + reason = getattr(exc, "reason", None) + body = getattr(exc, "data", None) or getattr(exc, "body", None) + pieces: list[str] = [] + if status: + pieces.append(f"HTTP {status}") + if reason: + pieces.append(str(reason)) + if body: + body_text = str(body).strip() + if len(body_text) > 300: + body_text = body_text[:300] + "..." + pieces.append(body_text) + return ": ".join(pieces) if pieces else str(exc) + + +class VisierClient: + """ + API client for one Visier tenant. + + The generated ``visier-platform-sdk`` handles authentication and request signing. + This wrapper only keeps source/target profile metadata and provides small methods + that match the promotion steps. + """ + + def __init__(self, profile: EnvironmentProfile, *, timeout: int = _DEFAULT_TIMEOUT) -> None: + self._profile = profile + self._timeout = timeout + sdk = _load_sdk() + self._api_exception_type = sdk["ApiException"] + self._config = sdk["Configuration"]( + host=profile.base_url, + api_key=profile.api_key, + username=profile.username, + password=profile.password, + vanity=profile.vanity, + ) + self._api_client = sdk["ApiClient"](self._config) + self._production_versions_api = sdk["ProductionVersionsApi"](self._api_client) + self._projects_api = sdk["ProjectsApi"](self._api_client) + self._export_params_type = sdk["ExportProductionVersionsAPIOperationParametersDTO"] + self._production_versions_operation_type = sdk["ProductionVersionsAPIOperationRequestDTO"] + self._project_type = sdk["ProjectDTO"] + self._project_operation_type = sdk["ProjectOperationRequestDTO"] + + @property + def profile(self) -> EnvironmentProfile: + return self._profile + + def authenticate(self) -> str: + """ + Step 0 - force SDK authentication now so failures are reported before mutations. + """ + try: + self._config.refresh_config(self._config, True) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=0) from e + token = (self._config.asid_token or self._config.access_token or "").strip() + if not token: + raise VisierSDLCError("Authentication completed without a token.", step=0) + logger.info("Authenticated tenant (base_url=%s)", self._profile.base_url) + return token + + def get_production_versions( + self, + *, + limit: int = 400, + start: int = 0, + ) -> dict[str, Any]: + """Step 1 - list published production versions (newest first).""" + try: + response = self._production_versions_api.get_production_versions( + limit=limit, + start=start, + target_tenant_id=self._profile.target_tenant_id, + _request_timeout=self._timeout, + ) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=1) from e + return _dto_to_dict(response) + + def get_all_published_production_versions( + self, + *, + page_size: int = 400, + ) -> dict[str, Any]: + """ + List all published production versions by following pagination until a page + is empty or shorter than ``page_size``. + """ + merged: list[dict[str, Any]] = [] + start = 0 + while True: + page = self.get_production_versions(limit=page_size, start=start) + batch = page.get("publishedVersions") or page.get("published_versions") + if not isinstance(batch, list) or not batch: + break + merged.extend([_dto_to_dict(item) for item in batch]) + if len(batch) < page_size: + break + start += page_size + return {"publishedVersions": merged} + + def export_production_versions_zip( + self, + start_version: str, + end_version: str, + excluded_versions: list[str] | None = None, + ) -> bytes: + """Step 2 - export a range of production versions as a ZIP.""" + request = self._production_versions_operation_type( + operation="export", + export_parameters=self._export_params_type( + start_version=start_version, + end_version=end_version, + excluded_versions=excluded_versions, + ), + ) + try: + response = self._production_versions_api.post_production_versions_without_preload_content( + request, + target_tenant_id=self._profile.target_tenant_id, + _request_timeout=self._timeout, + _headers={"Accept": "application/zip, application/json"}, + ) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=2) from e + content_type = (getattr(response, "headers", {}).get("Content-Type") or "").lower() + content = getattr(response, "data", b"") or b"" + if isinstance(content, str): + content = content.encode("utf-8") + if "json" in content_type: + snippet = content.decode("utf-8", errors="replace")[:300] + raise VisierSDLCError( + "Expected a ZIP file but got JSON. Response may be an error: " + snippet, + step=2, + ) + if "zip" not in content_type and content[:2] != b"PK": + logger.warning("Unexpected Content-Type for export: %s", content_type or "(missing)") + return bytes(content) + + def create_draft_project(self, spec: NewDraftProject) -> dict[str, Any]: + """Step 3 - create a new blank draft project.""" + payload = self._project_type( + name=spec.name, + description=spec.description, + release_version=spec.release_version, + ticket_number=spec.ticket_number, + version_number=spec.version_number, + ) + if visier_sdlc_troubleshoot_logging(): + logger.info( + "[SDLC troubleshooting] Creating project name=%r description_len=%s", + spec.name, + len(spec.description or ""), + ) + try: + response = self._projects_api.create_project( + payload, + target_tenant_id=self._profile.target_tenant_id, + _request_timeout=self._timeout, + ) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=3) from e + data = _dto_to_dict(response) + logger.info("Created draft project id=%s name=%r", data.get("id"), data.get("name")) + return data + + def import_commits(self, project_id: str, zip_bytes: bytes) -> dict[str, Any]: + """Step 4 - import the ZIP from Step 2 into an existing draft project.""" + try: + response = self._projects_api.put_project_commits( + project_id, + zip_bytes, + target_tenant_id=self._profile.target_tenant_id, + _request_timeout=self._timeout, + ) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=4) from e + return _dto_to_dict(response) + + def commit_and_publish(self, project_id: str) -> dict[str, Any]: + """Step 5 - run Visier ``commitAndPublish`` for this tenant.""" + request = self._project_operation_type(operation="commitAndPublish") + try: + response = self._projects_api.run_project_operation( + project_id, + request, + target_tenant_id=self._profile.target_tenant_id, + _request_timeout=self._timeout, + ) + except self._api_exception_type as e: + raise VisierSDLCError(_safe_api_error_message(e), step=5) from e + return _dto_to_dict(response) diff --git a/administration/sdlc-promotion/python/visier_sdlc/config.py b/administration/sdlc-promotion/python/visier_sdlc/config.py new file mode 100644 index 0000000..dc00d48 --- /dev/null +++ b/administration/sdlc-promotion/python/visier_sdlc/config.py @@ -0,0 +1,169 @@ +""" +Load per-tenant connection settings (e.g. ``source`` and ``target`` profiles) without hardcoding secrets. + +- Supports YAML files with a ``profiles`` map (see ``config.yaml.example``). +- If ``python-dotenv`` is installed, ``.env`` is loaded automatically so you can + keep secrets in environment variables and reference them from YAML, or set + ``${VAR_NAME}`` placeholders in the file (see ``load_config``). +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping +from urllib.parse import urlparse + +# Optional: load .env before reading YAML so ${VAR} can resolve +try: + from dotenv import load_dotenv + + _DOTENV = True +except ImportError: # pragma: no cover - optional dependency + _DOTENV = False + + +_ENV_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}") + + +@dataclass(frozen=True) +class EnvironmentProfile: + """ + Connection details for one Visier tenant (URL + credentials; used for source or target). + + * ``base_url`` — API root, e.g. https://.api.visier.io + * ``api_key`` — sent as the ``apikey`` header (see Visier code samples) + * ``username`` / ``password`` — used to obtain the ASID token + * ``target_tenant_id`` — optional ``TargetTenantID`` header (partner / multi-tenant) + * ``vanity`` — if your tenant requires it, sent as ``vanityName`` on + ``POST /v1/admin/visierSecureToken`` (matches Visier's Python connector) + """ + + base_url: str + api_key: str + username: str + password: str + target_tenant_id: str | None = None + vanity: str | None = None + + @staticmethod + def from_dict(data: Mapping[str, Any], *, profile_name: str = "profile") -> "EnvironmentProfile": + missing = [k for k in ("base_url", "api_key", "username", "password") if not data.get(k)] + if missing: + hint = ( + " If you use ${VAR_NAME} in config.yaml, that variable must be set in the environment " + "or in a .env file in the same folder as config.yaml." + ) + raise ValueError( + f"Profile {profile_name!r} is missing or has empty values for: {', '.join(missing)}.{hint}" + ) + base_url = str(data["base_url"]).strip().rstrip("/") + _validate_base_url(base_url, profile_name=profile_name) + return EnvironmentProfile( + base_url=base_url, + api_key=str(data["api_key"]).strip(), + username=str(data["username"]).strip(), + password=str(data["password"]), + target_tenant_id=(str(data["target_tenant_id"]).strip() if data.get("target_tenant_id") else None), + vanity=(str(data["vanity"]).strip() if data.get("vanity") else None), + ) + + +def _expand_env_placeholders(obj: Any) -> Any: + """Replace ``${VAR}`` in strings, recursively in dicts/lists.""" + + if isinstance(obj, str): + def replacer(match: re.Match[str]) -> str: + return os.environ.get(match.group(1), "") + + return _ENV_PATTERN.sub(replacer, obj) + if isinstance(obj, dict): + return {k: _expand_env_placeholders(v) for k, v in obj.items()} + if isinstance(obj, list): + return [_expand_env_placeholders(v) for v in obj] + return obj + + +def _validate_base_url(base_url: str, *, profile_name: str) -> None: + parsed = urlparse(base_url) + if parsed.scheme != "https": + raise ValueError(f"Profile {profile_name!r} base_url must use https://") + if not parsed.netloc: + raise ValueError(f"Profile {profile_name!r} base_url must include a host") + if parsed.query or parsed.fragment: + raise ValueError(f"Profile {profile_name!r} base_url must not include query strings or fragments") + + +def load_config( + path: str | os.PathLike[str] | None = None, +) -> dict[str, EnvironmentProfile]: + """ + Read ``config.yaml`` and return a map ``profile_name -> EnvironmentProfile``. + + * If ``path`` is None, looks for ``config.yaml`` in the current working directory. + * If PyYAML is not installed, raises ``ImportError`` with a clear message. + * Calls ``_expand_env_placeholders`` on the parsed YAML. + """ + cfg_path = Path(path or os.environ.get("VISIER_SDLC_CONFIG", "config.yaml")) + if not cfg_path.is_file(): + raise FileNotFoundError(f"Config file not found: {cfg_path.resolve()}") + + # Load .env only from the config file's directory. Loading from cwd as well can + # accidentally mix credentials when the command is launched from another folder. + if _DOTENV: + load_dotenv(cfg_path.parent / ".env") + + try: + import yaml + except ImportError as e: # pragma: no cover + raise ImportError("Install pyyaml (see requirements.txt) to use config.yaml") from e + + raw = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + raw = _expand_env_placeholders(raw) + if not raw or "profiles" not in raw: + raise ValueError("config.yaml must contain a top-level 'profiles' mapping") + out: dict[str, EnvironmentProfile] = {} + for name, p in raw["profiles"].items(): + if not isinstance(p, dict): + raise ValueError(f"Profile {name!r} must be a mapping (got {type(p).__name__!r})") + out[str(name)] = EnvironmentProfile.from_dict(p, profile_name=str(name)) + validate_distinct_source_and_target(out) + return out + + +def tenant_fingerprint(profile: EnvironmentProfile) -> tuple[str, str | None]: + """Return a conservative identity tuple for same-tenant safety checks.""" + host = (urlparse(profile.base_url).hostname or profile.base_url).lower() + return (host, profile.target_tenant_id) + + +def validate_distinct_source_and_target(config: Mapping[str, EnvironmentProfile]) -> None: + """Reject configs where source and target clearly point at the same tenant.""" + source = config.get("source") + target = config.get("target") + if not source or not target: + return + if tenant_fingerprint(source) == tenant_fingerprint(target): + raise ValueError( + "profiles.source and profiles.target appear to point at the same tenant. " + "Use distinct source and target tenants before running this sample." + ) + + +def get_profile(config: Mapping[str, EnvironmentProfile], name: str) -> EnvironmentProfile: + """Return a profile by name or raise ``KeyError`` with a short hint.""" + try: + return config[name] + except KeyError as e: + available = ", ".join(sorted(config.keys())) or "(none)" + raise KeyError(f"Unknown profile {name!r}. Available: {available}") from e + + +# --- Optional troubleshooting (draft project name / API payload) --- +# Grep: SDLC troubleshooting | visier_sdlc_troubleshoot_logging | VISIER_SDLC_DEBUG +# Set ``VISIER_SDLC_DEBUG=1`` to enable. Remove the guarded log blocks in client.py / workflow.py +# when you no longer need them, or keep with the env var unset. +def visier_sdlc_troubleshoot_logging() -> bool: + return os.environ.get("VISIER_SDLC_DEBUG", "").strip().lower() in ("1", "true", "yes") diff --git a/administration/sdlc-promotion/python/visier_sdlc/exceptions.py b/administration/sdlc-promotion/python/visier_sdlc/exceptions.py new file mode 100644 index 0000000..747cd49 --- /dev/null +++ b/administration/sdlc-promotion/python/visier_sdlc/exceptions.py @@ -0,0 +1,20 @@ +""" +Custom errors for the SDLC workflow. + +Raising a dedicated exception type lets calling code catch failures without +treating them like generic Python errors, and the ``step`` field is used +to build clear messages such as "Step 3 failed: ...". +""" + +from __future__ import annotations + + +class VisierSDLCError(Exception): + """Raised when a Visier SDLC step fails or the API returns an error response.""" + + def __init__(self, message: str, *, step: int | None = None) -> None: + self.step = step + if step is not None: + super().__init__(f"Step {step} failed: {message}") + else: + super().__init__(message) diff --git a/administration/sdlc-promotion/python/visier_sdlc/workflow.py b/administration/sdlc-promotion/python/visier_sdlc/workflow.py new file mode 100644 index 0000000..56fbf4f --- /dev/null +++ b/administration/sdlc-promotion/python/visier_sdlc/workflow.py @@ -0,0 +1,514 @@ +""" +End-to-end SDLC helper: copy published work from the *source* tenant into a new draft +on the *target* tenant, then publish. + +Visier API paths still use names like ``production-versions`` (their terminology for +published configuration history). Each public step helper is numbered to match the brief. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, Literal, TypeVar +from urllib.parse import urlparse + +from visier_sdlc.client import NewDraftProject, VisierClient +from visier_sdlc.config import EnvironmentProfile, visier_sdlc_troubleshoot_logging +from visier_sdlc.exceptions import VisierSDLCError + +logger = logging.getLogger(__name__) + +# --- Auto-generated draft project display name + description (Visier length limits) ---------- +_MAX_DRAFT_DISPLAY_NAME_LEN = 50 +_MAX_DRAFT_DESC_LEN = 150 + + +def source_tenant_vanity_label(profile: EnvironmentProfile) -> str: + """ + Short label for the source tenant: explicit ``vanity`` in config, else the hostname + segment from a typical ``https://.api.visier.io`` URL, else the full host. + """ + if profile.vanity and str(profile.vanity).strip(): + return str(profile.vanity).strip() + parsed = urlparse(profile.base_url) + host = (parsed.hostname or "").strip().lower() + if not host: + return "source" + if host.endswith(".api.visier.io") and host != "api.visier.io": + return host[: -len(".api.visier.io")] + return host + + +def _draft_utc_date_str() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%d") + + +def _truncate_draft_display_name(text: str, max_len: int = _MAX_DRAFT_DISPLAY_NAME_LEN) -> str: + if len(text) <= max_len: + return text + if max_len < 2: + return "…"[:max_len] + return text[: max_len - 1] + "…" + + +def _truncate_draft_description(text: str, max_len: int = _MAX_DRAFT_DESC_LEN) -> str: + text = text.strip() + if len(text) <= max_len: + return text + if max_len < 2: + return "…"[:max_len] + return text[: max_len - 1] + "…" + + +def _base_url_for_description(base: str) -> str: + """Prefer scheme + host so long paths/query strings do not eat the 150-char budget.""" + parsed = urlparse(base) + if parsed.netloc and parsed.scheme: + return f"{parsed.scheme}://{parsed.netloc}" + if parsed.netloc: + return parsed.netloc + return base + + +def _published_row_for_id(history: dict[str, Any], version_id: str) -> dict[str, Any] | None: + """ + Return the ``publishedVersions`` entry whose ``id`` matches ``version_id``. + + Match is by stripped string equality on both sides. Returns ``None`` if there is no + such row (no fallback to another version). + """ + want = str(version_id).strip() + versions = history.get("publishedVersions") + if not isinstance(versions, list): + return None + for v in versions: + if not isinstance(v, dict): + continue + got = str(v.get("id", "")).strip() + if got and got == want: + return v + return None + + +def _build_all_mode_display_name(vanity: str) -> str: + """ + Project display name for ``--mode all`` (max :data:`_MAX_DRAFT_DISPLAY_NAME_LEN` chars). + """ + s = f"Import full version history from {vanity}" + return _truncate_draft_display_name(s, _MAX_DRAFT_DISPLAY_NAME_LEN) + + +def _build_latest_mode_display_name(vanity: str, pv_name: str) -> str: + """ + Project display name for ``--mode latest`` (max :data:`_MAX_DRAFT_DISPLAY_NAME_LEN` chars). + Format: ``Imported from {vanity} - {project name}``. Truncates the project name, then + the whole string, to stay within the limit. + """ + m = _MAX_DRAFT_DISPLAY_NAME_LEN + s = f"Imported from {vanity} - {pv_name}" + if len(s) <= m: + return s + prefix = f"Imported from {vanity} - " + if len(prefix) >= m: + return _truncate_draft_display_name(s, m) + room = m - len(prefix) + if room >= 2: + if len(pv_name) <= room: + return prefix + pv_name + return prefix + pv_name[: room - 1] + "…" + if room == 1: + return _truncate_draft_display_name(s, m) + return _truncate_draft_display_name(s, m) + + +def build_auto_draft_name_and_description( + mode: Literal["latest", "all"], + source_profile: EnvironmentProfile, + history: dict[str, Any], + exported_start_version_id: str, +) -> tuple[str, str]: + """ + Return ``(name, description)`` for the target draft project. + + * **latest** — title includes vanity, published version name, and id. + * **all** — title is the full-history import; description references vanity and base URL. + + Display names are at most 50 characters; descriptions at most 150 (Visier limits). + """ + vanity = source_tenant_vanity_label(source_profile) + base = source_profile.base_url + base_short = _base_url_for_description(base) + date = _draft_utc_date_str() + + if mode == "all": + name = _build_all_mode_display_name(vanity) + desc = _truncate_draft_description( + f"Full range from {vanity} on {date}. Source {base_short}. All versions (paginated export)." + ) + return name, desc + + row = _published_row_for_id(history, exported_start_version_id) + if not row: + raise ValueError( + "Cannot build latest draft name: no publishedVersions row with id matching " + f"{exported_start_version_id!r}. The exported version must appear in the same " + "Step 1 history used for auto naming (e.g. use a large enough list limit, " + "export_full_published_history, or ensure manual export ids are present on the " + "listed page)." + ) + pv_name = (row.get("name") or "Published version").strip() or "Published version" + pv_id = str(row.get("id") or exported_start_version_id) + + name = _build_latest_mode_display_name(vanity, pv_name) + desc = _truncate_draft_description( + f"Latest from {vanity} ({base_short}) on {date}. Version \u201c{pv_name}\u201d (id: {pv_id})." + ) + return name, desc + +T = TypeVar("T") + + +def _run_step(step: int, label: str, fn: Callable[[], T]) -> T: + """ + Run ``fn`` and re-wrap failures so logs and exceptions always mention ``step`` + ``label``. + + (Inner functions in :class:`VisierClient` already attach ``step`` for HTTP issues; this + catches anything else, such as key errors when parsing a response.) + """ + logger.info("--- Step %s — %s ---", step, label) + try: + return fn() + except VisierSDLCError: + raise + except Exception as e: # noqa: BLE001 - surface unexpected bugs with a clear step label + raise VisierSDLCError(f"{label}: {e}", step=step) from e + + +# --- Public step wrappers (explicit numbering for learning / scripting) --- + + +def step0_get_secure_token(client: VisierClient) -> str: + """Step 0: ``POST /v1/admin/visierSecureToken`` (see :meth:`VisierClient.authenticate`).""" + return _run_step(0, "Get secure token", client.authenticate) + + +def step1_list_production_history(client: VisierClient, *, limit: int = 400, start: int = 0) -> dict[str, Any]: + """ + Step 1: ``GET /v1/admin/production-versions`` — list publish history + (returns a ``publishedVersions`` list when successful). + """ + return _run_step( + 1, + "List published versions at source tenant", + lambda: client.get_production_versions(limit=limit, start=start), + ) + + +def step2_download_production_export_zip( + source: VisierClient, + start_version: str, + end_version: str, + excluded_versions: list[str] | None = None, +) -> bytes: + """Step 2: ``POST /v1/admin/production-versions`` with ``export`` operation (ZIP).""" + return _run_step( + 2, + "Download published-version export (ZIP) from source tenant", + lambda: source.export_production_versions_zip( + start_version, + end_version, + excluded_versions=excluded_versions, + ), + ) + + +def step3_create_blank_project_on_target( + target: VisierClient, + project: NewDraftProject, +) -> dict[str, Any]: + """Step 3: ``POST /v1/admin/projects`` — new draft on the *target* tenant.""" + return _run_step(3, "Create blank draft project in target tenant", lambda: target.create_draft_project(project)) + + +def step4_import_zip_into_target_project( + target: VisierClient, + project_id: str, + zip_bytes: bytes, +) -> dict[str, Any]: + """Step 4: ``PUT /v1/admin/projects/{id}/commits`` (ZIP from Step 2).""" + return _run_step( + 4, + "Import ZIP into target tenant draft project", + lambda: target.import_commits(project_id, zip_bytes), + ) + + +def step5_publish_target_project(target: VisierClient, project_id: str) -> dict[str, Any]: + """Step 5: ``POST /v1/admin/projects/{id}`` with ``commitAndPublish``.""" + return _run_step( + 5, + "Commit and publish draft in target tenant", + lambda: target.commit_and_publish(project_id), + ) + + +@dataclass +class PromotionResult: + """Summary returned to the caller after a successful end-to-end run.""" + + published_versions_list: dict[str, Any] + """Exact JSON from ``GET /v1/admin/production-versions`` (Step 1).""" + export_start_version_id: str + """Production version id passed to the export call (start of range).""" + export_end_version_id: str + """Production version id passed to the export call (end of range).""" + planned_project: NewDraftProject + """Draft project values that will be used if target changes are applied.""" + export_zip_size_bytes: int | None = None + """ZIP size when Step 2 ran; ``None`` in preview mode.""" + new_project: dict[str, Any] | None = None + """Step 3 response when target changes were applied; ``None`` in preview mode.""" + import_result: dict[str, Any] | None = None + """Step 4 response when target changes were applied; ``None`` in preview mode.""" + publish_result: dict[str, Any] | None = None + """Step 5 JSON when publish ran; ``None`` unless publishing was requested.""" + + +@dataclass +class PromotionRequest: + """All inputs required to run the workflow in one call.""" + + new_project: NewDraftProject + export_excluded_version_ids: list[str] | None = None + # --- Export version selection (ties Step 1 → Step 2 together) --- + # Visier returns ``publishedVersions`` newest-first. If you leave the manual ids unset, + # the workflow uses ``publishedVersions[export_published_version_index]`` for BOTH + # start and end (a single-version export), unless ``export_full_published_history`` is True. + export_start_version_id: str | None = None + export_end_version_id: str | None = None + export_published_version_index: int = 0 + # When True, Step 1 lists *all* pages of published versions, then export uses the **oldest** + # and **newest** id in that list as ``startVersion`` / ``endVersion`` (full range in one ZIP). + # Incompatible with manual ``export_start_version_id`` / ``export_end_version_id`` or index mode. + export_full_published_history: bool = False + # Single-page GET (when not using export_full_published_history). + source_production_versions_list_limit: int = 400 + source_production_versions_list_start: int = 0 + # Page size when fetching every page for ``export_full_published_history``. + source_production_versions_page_size: int = 400 + # When set to ``"latest"`` or ``"all"``, ``name`` / ``description`` on ``new_project`` are + # replaced using the source profile and Step 1 history (vanity + base URL in both cases). + auto_draft_naming: Literal["latest", "all"] | None = None + # Default is a preview: authenticate source, list versions, choose the export range, + # and compute the target draft metadata. No ZIP export and no target mutation happen + # unless ``apply_to_target`` is True. + apply_to_target: bool = False + # Publishing is production-impacting. Keep it separate from applying the import so + # callers can create and inspect a target draft without committing it. + publish: bool = False + + +def pick_export_version_ids( + history: dict[str, Any], + request: PromotionRequest, +) -> tuple[str, str]: + """ + Decide which production version UUIDs to send to ``POST …/production-versions`` (export). + + * If ``export_start_version_id`` and ``export_end_version_id`` are both set, those values + are used (manual range — can differ for multi-version exports). Not used with + ``export_full_published_history``. + * If ``export_full_published_history`` is True, ``startVersion``/``endVersion`` are + the **oldest** and **newest** version in ``history`` (list is newest-first from Visier). + * Otherwise, if both manual ids are ``None``, ids are taken from a single list row using + ``export_published_version_index`` (default ``0`` = latest only). + * Setting only one of the manual ids is not allowed (ambiguous). + """ + manual_start = request.export_start_version_id + manual_end = request.export_end_version_id + if request.export_full_published_history: + if manual_start is not None or manual_end is not None: + raise VisierSDLCError( + "Do not set export_start_version_id / export_end_version_id when using " + "export_full_published_history (or turn export_full_published_history off).", + step=1, + ) + versions = history.get("publishedVersions") + if not isinstance(versions, list) or not versions: + raise VisierSDLCError( + "No publishedVersions in the API response (or list is empty); nothing to export.", + step=1, + ) + if len(versions) == 1: + v = str(versions[0]["id"]) + if not v: + raise VisierSDLCError("publishedVersions[0] has no 'id' field.", step=1) + return v, v + # Newest-first: index 0 = latest publish, -1 = oldest in the merged list. + oldest = versions[-1] + newest = versions[0] + if not isinstance(oldest, dict) or not isinstance(newest, dict): + raise VisierSDLCError("publishedVersions entries are not objects; cannot read id.", step=1) + oid, nid = oldest.get("id"), newest.get("id") + if not oid or not nid: + raise VisierSDLCError("publishedVersions item missing 'id' (full-history export).", step=1) + return str(oid), str(nid) + + if manual_start is not None or manual_end is not None: + if manual_start is None or manual_end is None: + raise VisierSDLCError( + "Set both export_start_version_id and export_end_version_id for a manual range, " + "or leave both unset to auto-select from GET /v1/admin/production-versions.", + step=1, + ) + return manual_start.strip(), manual_end.strip() + + versions = history.get("publishedVersions") + if not isinstance(versions, list) or not versions: + raise VisierSDLCError( + "No publishedVersions in the API response (or list is empty); nothing to export.", + step=1, + ) + idx = request.export_published_version_index + if idx < 0 or idx >= len(versions): + raise VisierSDLCError( + f"export_published_version_index={idx} is out of range " + f"(API returned {len(versions)} production version(s)).", + step=1, + ) + entry = versions[idx] + if not isinstance(entry, dict): + raise VisierSDLCError(f"publishedVersions[{idx}] is not an object; cannot read id.", step=1) + vid = entry.get("id") + if not vid: + raise VisierSDLCError(f"publishedVersions[{idx}] has no 'id' field.", step=1) + return str(vid), str(vid) + + +def run_sdlc_promotion( + source: VisierClient, + target: VisierClient, + request: PromotionRequest, +) -> PromotionResult: + """ + Run a guarded SDLC promotion workflow. + + By default, this is a preview: authenticate the source tenant, list source published + history, choose the export ids, and compute target draft metadata. It does not export + the ZIP and does not mutate the target tenant. + + Set ``apply_to_target=True`` to export the ZIP, authenticate the target tenant, create + a target draft, and import the ZIP. Set ``publish=True`` as well to run Step 5 + ``commitAndPublish``. + + Export version ids for Step 2 are resolved from the Step 1 response unless you set + ``export_start_version_id`` / ``export_end_version_id``. Use + ``export_full_published_history=True`` to export every published version returned by a + paginated listing (oldest→newest span). + + Set ``auto_draft_naming`` to ``"latest"`` or ``"all"`` to fill the target draft + project name and description (vanity + base URL; for ``latest``, the exported + version’s name and id as well). + """ + if request.publish and not request.apply_to_target: + raise VisierSDLCError("publish=True requires apply_to_target=True.", step=None) + + step0_get_secure_token(source) + if request.export_full_published_history: + history = _run_step( + 1, + "List all published versions at source tenant (paginated)", + lambda: source.get_all_published_production_versions( + page_size=request.source_production_versions_page_size + ), + ) + else: + history = step1_list_production_history( + source, + limit=request.source_production_versions_list_limit, + start=request.source_production_versions_list_start, + ) + start_id, end_id = pick_export_version_ids(history, request) + logger.info("Selected published version range (start=%s, end=%s).", start_id, end_id) + if request.auto_draft_naming is not None: + try: + auto_name, auto_desc = build_auto_draft_name_and_description( + request.auto_draft_naming, + source.profile, + history, + start_id, + ) + except ValueError as e: + raise VisierSDLCError(f"Auto draft naming failed: {e}", step=1) from e + base = request.new_project + draft_project = NewDraftProject( + name=auto_name, + description=auto_desc, + release_version=base.release_version, + ticket_number=base.ticket_number, + version_number=base.version_number, + ) + else: + draft_project = request.new_project + if visier_sdlc_troubleshoot_logging(): + logger.info( + "[SDLC troubleshooting] Step 3 NewDraftProject: name=%r (len=%s) description_len=%s", + draft_project.name, + len(draft_project.name or ""), + len(draft_project.description or ""), + ) + if not request.apply_to_target: + logger.info("Preview only: no ZIP export, target draft creation, import, or publish was run.") + return PromotionResult( + published_versions_list=history, + export_start_version_id=start_id, + export_end_version_id=end_id, + planned_project=draft_project, + ) + + zip_bytes = step2_download_production_export_zip( + source, + start_id, + end_id, + request.export_excluded_version_ids, + ) + step0_get_secure_token(target) + created = step3_create_blank_project_on_target(target, draft_project) + project_id = created.get("id") + if not project_id: + raise VisierSDLCError("Create project response missing 'id' field; cannot import.", step=3) + imported = step4_import_zip_into_target_project(target, str(project_id), zip_bytes) + if not request.publish: + logger.info( + "No publish requested: stopping before Step 5. Draft project %s exists on target with import applied; " + "not publishing.", + project_id, + ) + return PromotionResult( + published_versions_list=history, + export_start_version_id=start_id, + export_end_version_id=end_id, + planned_project=draft_project, + export_zip_size_bytes=len(zip_bytes), + new_project=created, + import_result=imported, + publish_result=None, + ) + published = step5_publish_target_project(target, str(project_id)) + logger.info("SDLC promotion finished. Draft project %s published in target tenant.", project_id) + return PromotionResult( + published_versions_list=history, + export_start_version_id=start_id, + export_end_version_id=end_id, + planned_project=draft_project, + export_zip_size_bytes=len(zip_bytes), + new_project=created, + import_result=imported, + publish_result=published, + ) + + +# Backward compatibility (older name) +run_dev_to_production_sdlc = run_sdlc_promotion