diff --git a/README.md b/README.md index 34ee4bd..61b9ea3 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ Python 3.9 or newer is required. CI tests every minor from Python 3.9 through Py uv add sanka-sdk ``` -## Usage +## Hosted API ```python from sanka_sdk import SankaClient @@ -22,7 +22,11 @@ response = client.public_auth.whoami() print(response) ``` -## Local migration +`SankaClient` calls Sanka's hosted HTTP API and requires a token. Extension +management is part of the local migration runtime described below; it is not a +hosted API resource. + +## Local migration runtime The hosted API client and local migration adapter are separate surfaces: @@ -31,35 +35,107 @@ The hosted API client and local migration adapter are separate surfaces: | `from sanka_sdk import SankaClient` | Sanka's hosted HTTP API | API token | | `SankaMigrate` or `AsyncSankaMigrate` from `sanka_sdk.migrate` | A local `sanka-migrate` subprocess | None | -Install the migration runtime separately, then use the tokenless adapter: +Install the runtime separately. Installing `sanka-sdk` does not install or +authenticate `sanka-migrate`. ```bash uv tool install sanka-migrate ``` -### Synchronous +Use a runtime release that includes the extension marketplace commands and +the published default DRF extension dependency. + +### Configure an extension, scan, and plan + +Add the official marketplace and lock the extension before the first scan. +Marketplace snapshots are user-scoped; the extension lock belongs to the +project in `cwd`. ```python from sanka_sdk.migrate import SankaMigrate migrate = SankaMigrate(cwd="./django-app") +migrate.extensions.marketplaces.add( + "git@github.com:sankaHQ/extensions.git", + name="sanka", +) +migrate.extensions.add("sanka/drf-to-fastapi", marketplace="sanka") + scan = migrate.scan() plan = migrate.plan( to="fastapi", - generation="full", - strategy="native", - package_manager="uv", + extension_config={ + "generation": "minimal", + "output": "./fastapi-app", + "package_manager": "uv", + "strategy": "native", + }, + extension_environment=("DJANGO_SECRET_KEY",), ) applied = migrate.apply(plan_hash=plan.data["plan_hash"]) tested = migrate.test() verified = migrate.verify() ``` -### Asynchronous +`scan.data["recommendations"]` contains the selected extension, its target, +matching evidence, and install status. When the exact default package is +already installed and has not been disabled, `sanka-migrate` can lock it on +the first scan. Otherwise, if no matching extension is enabled, the command +stops with `SANKA_EXTENSION_REQUIRED`. The error details contain the +recommendations and exact `add_command`; the SDK does not bypass the runtime's +selection and trust checks. + +`extension_config` accepts JSON-compatible values and is serialized as stable, +sorted JSON. `extension_environment` accepts environment variable names, not +secret values. `sanka-migrate` forwards only those named values to the selected +extension. Both options are available on `scan()`, `plan()`, `apply()`, +`test()`, and `verify()`. + +### Manage extensions and marketplaces + +```python +extensions = migrate.extensions + +installed = extensions.list() +extensions.add("example/demo", marketplace="partner") +extensions.remove("example/demo") + +marketplaces = extensions.marketplaces +marketplaces.add( + "https://github.com/example/sanka-extensions.git", + name="partner", + trust=True, +) +marketplaces.list() +marketplaces.upgrade("partner") # Omit the name to upgrade all marketplaces. +marketplaces.remove("partner") +``` + +The Python methods map directly to these local commands: + +| Python method | `sanka-migrate` command | +|---|---| +| `extensions.list()` | `extension list --json` | +| `extensions.add(id, marketplace=...)` | `extension add ID --marketplace NAME --json` | +| `extensions.remove(id)` | `extension remove ID --json` | +| `extensions.marketplaces.list()` | `extension marketplace list --json` | +| `extensions.marketplaces.add(source, name=..., trust=True)` | `extension marketplace add SOURCE --name NAME --trust --json` | +| `extensions.marketplaces.upgrade(name)` | `extension marketplace upgrade NAME --json` | +| `extensions.marketplaces.remove(name)` | `extension marketplace remove NAME --json` | + +`trust=True` is an explicit operator decision. The SDK only passes `--trust`. +`sanka-migrate` owns source identity checks, immutable marketplace snapshots, +artifact verification, project locks, extension installation, upgrades, and +removal safety. An untrusted source fails with +`SANKA_MARKETPLACE_TRUST_REQUIRED`; the SDK does not bypass that check. + +### Async adapter Use `AsyncSankaMigrate` to run the same commands without blocking the event -loop. Cancelling an awaited command also terminates its local CLI process. +loop. Its lifecycle, extension, and marketplace methods have the same arguments +and results as the synchronous adapter. Cancelling an awaited command kills and +reaps its local CLI process. ```python import asyncio @@ -70,8 +146,9 @@ from sanka_sdk.migrate import AsyncSankaMigrate async def main() -> None: migrate = AsyncSankaMigrate(cwd="./django-app") - await migrate.scan() - plan = await migrate.plan(to="fastapi", generation="full") + scan = await migrate.scan() + await migrate.extensions.list() + plan = await migrate.plan(to="fastapi") await migrate.apply(plan_hash=plan.data["plan_hash"]) await migrate.test() await migrate.verify() @@ -90,17 +167,41 @@ Each method maps directly to the local runtime: | `test()` | `sanka-migrate test ... --json` | Prepare the generated target environment and run its tests | | `verify()` | `sanka-migrate verify ... --json` | Verify integrity and configured behavior | -Both adapters invoke an argument vector without a shell and never call Sanka's -hosted API. They forward only parameters you provide; defaults, validation, -framework detection, generated-target environments, and plan-hash safety remain -owned by `sanka-migrate`. Every call returns a typed `SankaMigrateResult` with -the `sanka-cli/v1` fields `data`, `artifacts`, `limitations`, and -`next_actions`. - -Failures raise `SankaMigrateError`. Its `command`, `exit_code`, `parsed_error`, -and `stderr` attributes distinguish a migration failure (exit `1`), invalid -usage (exit `2`), a missing executable, and an invalid protocol response. The -public classes and methods include docstrings for IDE hover and `help()`. +### Results, failures, and subprocess safety + +Both adapters execute an argv list without a shell and never call Sanka's +hosted API. Arguments such as marketplace URLs, paths, and configuration values +are not interpreted as shell commands. + +Every successful call returns a typed `SankaMigrateResult`. The validated +`sanka-cli/v1` fields are `schema_version`, `command`, `outcome`, +`migration_state`, `data`, `artifacts`, `limitations`, and `next_actions`. +`ScanData`, `ExtensionRecommendation`, `ExtensionEvidence`, and +`ExtensionFailure` describe the extension-specific data available to type +checkers and IDEs. + +```python +from sanka_sdk.migrate import SankaMigrateError + +try: + migrate.extensions.marketplaces.add("./third-party", name="third-party") +except SankaMigrateError as error: + print(error.command, error.exit_code) + print(error.parsed_error) # code, message, and optional details + print(error.result) # Complete validated failure envelope, when available. +``` + +Failures are fail-closed. The SDK rejects missing executables, malformed or +non-object JSON, a schema other than `sanka-cli/v1`, the wrong command, invalid +outcome/exit-code pairs, malformed `data.error`, and non-string artifact or +action lists. A valid CLI failure raises `SankaMigrateError` with its typed +result preserved. Exit `1` is a runtime failure, exit `2` is invalid usage, and +any other exit code is a protocol error. + +Defaults, framework detection, marketplace trust, immutable snapshots, +extension subprocess execution, generated-target environments, and plan-hash +safety remain in `sanka-migrate`. The SDK is a typed local adapter, not a second +migration runtime. See the [CLI execution model](https://github.com/sankaHQ/sanka/blob/main/docs/django-to-fastapi.md#cli-and-sdk-execution-model) and [Sanka developer documentation](https://sanka.com/docs/developers/). diff --git a/handwritten/sanka_sdk/migrate.py b/handwritten/sanka_sdk/migrate.py index 49cebe3..6c20e4d 100644 --- a/handwritten/sanka_sdk/migrate.py +++ b/handwritten/sanka_sdk/migrate.py @@ -10,19 +10,28 @@ import asyncio import json +import math import os +import re import subprocess from dataclasses import dataclass from typing import Any, Dict, Generic, List, Literal, Mapping, Optional, Sequence, Tuple, TypedDict, TypeVar, Union PathValue = Union[str, "os.PathLike[str]"] +JsonValue = Union[None, bool, int, float, str, List["JsonValue"], Dict[str, "JsonValue"]] +SankaMigrateCommand = Literal["scan", "plan", "apply", "test", "verify", "extension"] CLI_SCHEMA_VERSION = "sanka-cli/v1" __all__ = [ "ApplyData", "AsyncSankaMigrate", + "ExtensionEvidence", + "ExtensionFailure", + "ExtensionRecommendation", + "JsonValue", "PlanData", "SankaMigrate", + "SankaMigrateCommand", "SankaMigrateError", "SankaMigrateResult", "ScanData", @@ -33,10 +42,42 @@ TData = TypeVar("TData") +class ExtensionEvidence(TypedDict): + """Static project evidence that matched an extension recommendation.""" + + kind: str + value: str + path: str + + +class ExtensionRecommendation(TypedDict): + """One compatible extension recommended by ``sanka-migrate``.""" + + id: str + version: str + marketplace: str + targets: List[str] + evidence: List[ExtensionEvidence] + status: List[str] + add_command: str + + +class _ExtensionFailureRequired(TypedDict): + code: str + message: str + + +class ExtensionFailure(_ExtensionFailureRequired, total=False): + """Structured extension or marketplace failure returned by the CLI.""" + + details: Dict[str, JsonValue] + + class ScanData(TypedDict, total=False): """Core semantic scan fields; additional CLI fields remain available.""" scan_hash: str + recommendations: List[ExtensionRecommendation] class PlanData(TypedDict): @@ -79,7 +120,7 @@ class SankaMigrateResult(Generic[TData]): """ schema_version: str - command: str + command: SankaMigrateCommand outcome: str migration_state: str data: TData @@ -95,6 +136,7 @@ class SankaMigrateError(RuntimeError): command: Generic command that failed. exit_code: Process exit code, or ``None`` when the CLI could not start. parsed_error: Structured CLI error from ``data.error``, when available. + result: Complete valid failure envelope, when the CLI returned one. stderr: Diagnostic text written by the CLI. """ @@ -102,15 +144,17 @@ def __init__( self, message: str, *, - command: str, + command: SankaMigrateCommand, exit_code: Optional[int] = None, parsed_error: Optional[Dict[str, Any]] = None, + result: Optional[SankaMigrateResult[Dict[str, Any]]] = None, stderr: str = "", ) -> None: super().__init__(message) self.command = command self.exit_code = exit_code self.parsed_error = parsed_error + self.result = result self.stderr = stderr @@ -149,6 +193,8 @@ def scan( root: Optional[PathValue] = None, settings: Optional[str] = None, artifact_dir: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ScanData]: """Inspect a source application with ``sanka-migrate scan``. @@ -156,6 +202,8 @@ def scan( root: Source repository root. Omit it to use ``cwd``. settings: Explicit Django settings module; otherwise the CLI detects it. artifact_dir: Directory for the semantic scan artifact. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The scan result, discovered application data, risks, and artifact paths. @@ -170,7 +218,10 @@ def scan( ``scan = SankaMigrate(cwd="./app").scan()`` """ - return self._run("scan", _scan_args(root, settings, artifact_dir)) + return self._run( + "scan", + _scan_args(root, settings, artifact_dir, extension_config, extension_environment), + ) def plan( self, @@ -178,13 +229,15 @@ def plan( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, strategy: Optional[Literal["native", "compatibility"]] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, generation: Optional[Literal["full", "update", "minimal"]] = None, package_manager: Optional[Literal["uv", "pip"]] = None, orm: Optional[Literal["tortoise", "sqlalchemy", "psycopg"]] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[PlanData]: """Create a reviewable, hash-bound plan with ``sanka-migrate plan``. @@ -192,13 +245,15 @@ def plan( root: Source repository root. Omit it to use ``cwd``. file: Migration spec passed as ``--file``. state: Run-state SQLite file passed as ``--state``. - to: Target framework, currently ``"fastapi"`` for application migration. + to: Target advertised by an installed extension. strategy: Runtime strategy, currently ``"native"`` or ``"compatibility"``. artifact_dir: Directory containing scan and plan artifacts. output: Planned generated target directory. generation: Generation mode: ``"full"``, ``"update"``, or ``"minimal"``. package_manager: Generated environment manager: ``"uv"`` or ``"pip"``. orm: ORM selected when the scan detects database-backed routes. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The plan result. Use ``result.data["plan_hash"]`` for ``apply()``. @@ -227,6 +282,8 @@ def plan( generation, package_manager, orm, + extension_config, + extension_environment, ), ) @@ -237,7 +294,7 @@ def apply( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, force: bool = False, @@ -245,6 +302,8 @@ def apply( min_readiness: Optional[float] = None, gap_report_only: bool = False, bench_candidate: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ApplyData]: """Apply exactly one reviewed plan with ``sanka-migrate apply``. @@ -261,6 +320,8 @@ def apply( min_readiness: Minimum native readiness percentage from 0 through 100. gap_report_only: Write a gap report instead of generating an application. bench_candidate: Also write a Migration Bench candidate here. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The apply result and paths written from the reviewed plan. @@ -291,6 +352,8 @@ def apply( min_readiness, gap_report_only, bench_candidate, + extension_config, + extension_environment, ), ) @@ -300,9 +363,11 @@ def test( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[TestData]: """Run generated-target tests with ``sanka-migrate test``. @@ -313,6 +378,8 @@ def test( to: Target framework selector. artifact_dir: Directory containing the applied plan. output: Generated target directory. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Test verdict, target interpreter, dependencies, and generated test artifact. @@ -328,7 +395,19 @@ def test( ``tested = SankaMigrate(cwd="./source").test()`` """ - return self._run("test", _test_args(root, file, state, to, artifact_dir, output)) + return self._run( + "test", + _test_args( + root, + file, + state, + to, + artifact_dir, + output, + extension_config, + extension_environment, + ), + ) def verify( self, @@ -336,11 +415,13 @@ def verify( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, cases: Optional[PathValue] = None, no_http: bool = False, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[VerifyData]: """Verify the selected migration with ``sanka-migrate verify``. @@ -353,6 +434,8 @@ def verify( output: Generated target directory. cases: JSON file containing additional read-only HTTP verification cases. no_http: Skip HTTP probes when structural verification is sufficient. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Verification verdict, checked scope, artifacts, and limitations. @@ -370,10 +453,27 @@ def verify( return self._run( "verify", - _verify_args(root, file, state, to, artifact_dir, output, cases, no_http), + _verify_args( + root, + file, + state, + to, + artifact_dir, + output, + cases, + no_http, + extension_config, + extension_environment, + ), ) - def _run(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: + @property + def extensions(self) -> _SankaExtensions: + """Extension and marketplace management commands.""" + + return _SankaExtensions(self) + + def _run(self, command: SankaMigrateCommand, args: Sequence[str]) -> SankaMigrateResult[Any]: argv, environment = self._prepare(command, args) try: completed = subprocess.run( @@ -399,7 +499,9 @@ def _run(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: stderr=completed.stderr, ) - def _prepare(self, command: str, args: Sequence[str]) -> Tuple[List[str], Dict[str, str]]: + def _prepare( + self, command: SankaMigrateCommand, args: Sequence[str] + ) -> Tuple[List[str], Dict[str, str]]: if self.cwd is not None and not os.path.isdir(self.cwd): raise SankaMigrateError( "sanka-migrate working directory was not found: {}".format(self.cwd), @@ -428,6 +530,8 @@ async def scan( root: Optional[PathValue] = None, settings: Optional[str] = None, artifact_dir: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ScanData]: """Asynchronously inspect a source with ``sanka-migrate scan``. @@ -435,6 +539,8 @@ async def scan( root: Source repository root. Omit it to use ``cwd``. settings: Explicit Django settings module. artifact_dir: Directory for the semantic scan artifact. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The scan result, discovered application data, risks, and artifacts. @@ -449,7 +555,10 @@ async def scan( ``scan = await AsyncSankaMigrate(cwd="./app").scan()`` """ - return await self._run_async("scan", _scan_args(root, settings, artifact_dir)) + return await self._run_async( + "scan", + _scan_args(root, settings, artifact_dir, extension_config, extension_environment), + ) async def plan( self, @@ -457,13 +566,15 @@ async def plan( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, strategy: Optional[Literal["native", "compatibility"]] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, generation: Optional[Literal["full", "update", "minimal"]] = None, package_manager: Optional[Literal["uv", "pip"]] = None, orm: Optional[Literal["tortoise", "sqlalchemy", "psycopg"]] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[PlanData]: """Asynchronously create a hash-bound plan with ``sanka-migrate plan``. @@ -471,13 +582,15 @@ async def plan( root: Source repository root. Omit it to use ``cwd``. file: Migration spec passed as ``--file``. state: Run-state SQLite file passed as ``--state``. - to: Target framework, currently ``"fastapi"``. + to: Target advertised by an installed extension. strategy: ``"native"`` or ``"compatibility"``. artifact_dir: Directory containing scan and plan artifacts. output: Planned generated target directory. generation: ``"full"``, ``"update"``, or ``"minimal"``. package_manager: ``"uv"`` or ``"pip"``. orm: ORM selected for database-backed routes. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The plan result whose ``data["plan_hash"]`` is required by ``apply``. @@ -505,6 +618,8 @@ async def plan( generation, package_manager, orm, + extension_config, + extension_environment, ), ) @@ -515,7 +630,7 @@ async def apply( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, force: bool = False, @@ -523,6 +638,8 @@ async def apply( min_readiness: Optional[float] = None, gap_report_only: bool = False, bench_candidate: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ApplyData]: """Asynchronously apply one reviewed plan with ``sanka-migrate apply``. @@ -539,6 +656,8 @@ async def apply( min_readiness: Minimum native readiness percentage. gap_report_only: Write a gap report instead of an application. bench_candidate: Also write a Migration Bench candidate here. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The apply result and paths written from the reviewed plan. @@ -569,6 +688,8 @@ async def apply( min_readiness, gap_report_only, bench_candidate, + extension_config, + extension_environment, ), ) @@ -578,9 +699,11 @@ async def test( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[TestData]: """Asynchronously run generated tests with ``sanka-migrate test``. @@ -591,6 +714,8 @@ async def test( to: Target framework selector. artifact_dir: Directory containing the applied plan. output: Generated target directory. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Test verdict, target interpreter, dependencies, and test artifact. @@ -605,7 +730,19 @@ async def test( ``tested = await AsyncSankaMigrate(cwd="./source").test()`` """ - return await self._run_async("test", _test_args(root, file, state, to, artifact_dir, output)) + return await self._run_async( + "test", + _test_args( + root, + file, + state, + to, + artifact_dir, + output, + extension_config, + extension_environment, + ), + ) async def verify( self, @@ -613,11 +750,13 @@ async def verify( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, cases: Optional[PathValue] = None, no_http: bool = False, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[VerifyData]: """Asynchronously verify with ``sanka-migrate verify``. @@ -630,6 +769,8 @@ async def verify( output: Generated target directory. cases: JSON file with additional read-only HTTP cases. no_http: Skip HTTP probes when structural checks are sufficient. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Verification verdict, checked scope, artifacts, and limitations. @@ -646,10 +787,29 @@ async def verify( return await self._run_async( "verify", - _verify_args(root, file, state, to, artifact_dir, output, cases, no_http), + _verify_args( + root, + file, + state, + to, + artifact_dir, + output, + cases, + no_http, + extension_config, + extension_environment, + ), ) - async def _run_async(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: + @property + def extensions(self) -> _AsyncSankaExtensions: + """Asynchronous extension and marketplace management commands.""" + + return _AsyncSankaExtensions(self) + + async def _run_async( + self, command: SankaMigrateCommand, args: Sequence[str] + ) -> SankaMigrateResult[Any]: argv, environment = self._prepare(command, args) try: process = await asyncio.create_subprocess_exec( @@ -689,11 +849,145 @@ async def _run_async(self, command: str, args: Sequence[str]) -> SankaMigrateRes ) -def _scan_args(root: Optional[PathValue], settings: Optional[str], artifact_dir: Optional[PathValue]) -> List[str]: +class _SankaExtensionMarketplaces: + def __init__(self, migrate: SankaMigrate) -> None: + self._migrate = migrate + + def add( + self, + source: PathValue, + *, + name: Optional[str] = None, + trust: bool = False, + ) -> SankaMigrateResult[Dict[str, Any]]: + """Add an immutable marketplace snapshot, explicitly trusting it when requested.""" + + return self._migrate._run("extension", _marketplace_add_args(source, name, trust)) + + def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """List configured marketplace snapshots.""" + + return self._migrate._run("extension", ["extension", "marketplace", "list"]) + + def upgrade(self, name: Optional[str] = None) -> SankaMigrateResult[Dict[str, Any]]: + """Refresh one marketplace, or all marketplaces when ``name`` is omitted.""" + + args = ["extension", "marketplace", "upgrade"] + if name is not None: + args.append(name) + return self._migrate._run("extension", args) + + def remove(self, name: str) -> SankaMigrateResult[Dict[str, Any]]: + """Remove an unused marketplace snapshot.""" + + return self._migrate._run("extension", ["extension", "marketplace", "remove", name]) + + +class _SankaExtensions: + def __init__(self, migrate: SankaMigrate) -> None: + self._migrate = migrate + self.marketplaces = _SankaExtensionMarketplaces(migrate) + + def add( + self, extension_id: str, *, marketplace: Optional[str] = None + ) -> SankaMigrateResult[Dict[str, Any]]: + """Install and lock an extension, optionally selecting its marketplace.""" + + args = ["extension", "add", extension_id] + _option(args, "--marketplace", marketplace) + return self._migrate._run("extension", args) + + def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """List available and installed extensions.""" + + return self._migrate._run("extension", ["extension", "list"]) + + def remove(self, extension_id: str) -> SankaMigrateResult[Dict[str, Any]]: + """Unpin or disable an extension in the current project.""" + + return self._migrate._run("extension", ["extension", "remove", extension_id]) + + +class _AsyncSankaExtensionMarketplaces: + def __init__(self, migrate: AsyncSankaMigrate) -> None: + self._migrate = migrate + + async def add( + self, + source: PathValue, + *, + name: Optional[str] = None, + trust: bool = False, + ) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously add an immutable marketplace snapshot.""" + + return await self._migrate._run_async("extension", _marketplace_add_args(source, name, trust)) + + async def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously list configured marketplace snapshots.""" + + return await self._migrate._run_async("extension", ["extension", "marketplace", "list"]) + + async def upgrade(self, name: Optional[str] = None) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously refresh one marketplace, or all when omitted.""" + + args = ["extension", "marketplace", "upgrade"] + if name is not None: + args.append(name) + return await self._migrate._run_async("extension", args) + + async def remove(self, name: str) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously remove an unused marketplace snapshot.""" + + return await self._migrate._run_async( + "extension", ["extension", "marketplace", "remove", name] + ) + + +class _AsyncSankaExtensions: + def __init__(self, migrate: AsyncSankaMigrate) -> None: + self._migrate = migrate + self.marketplaces = _AsyncSankaExtensionMarketplaces(migrate) + + async def add( + self, extension_id: str, *, marketplace: Optional[str] = None + ) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously install and lock an extension.""" + + args = ["extension", "add", extension_id] + _option(args, "--marketplace", marketplace) + return await self._migrate._run_async("extension", args) + + async def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously list available and installed extensions.""" + + return await self._migrate._run_async("extension", ["extension", "list"]) + + async def remove(self, extension_id: str) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously unpin or disable an extension.""" + + return await self._migrate._run_async("extension", ["extension", "remove", extension_id]) + + +def _marketplace_add_args(source: PathValue, name: Optional[str], trust: bool) -> List[str]: + args = ["extension", "marketplace", "add", os.fspath(source)] + _option(args, "--name", name) + _flag(args, "--trust", trust) + return args + + +def _scan_args( + root: Optional[PathValue], + settings: Optional[str], + artifact_dir: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], +) -> List[str]: args = ["scan"] _positional(args, root) _option(args, "--settings", settings) _option(args, "--artifact-dir", artifact_dir) + _extension_options(args, extension_config, extension_environment) return args @@ -708,6 +1002,8 @@ def _plan_args( generation: Optional[str], package_manager: Optional[str], orm: Optional[str], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["plan"] _positional(args, root) @@ -723,6 +1019,7 @@ def _plan_args( ("--orm", orm), ): _option(args, flag, value) + _extension_options(args, extension_config, extension_environment) return args @@ -739,6 +1036,8 @@ def _apply_args( min_readiness: Optional[float], gap_report_only: bool, bench_candidate: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: if not plan_hash.strip(): raise ValueError("plan_hash must not be empty") @@ -757,6 +1056,7 @@ def _apply_args( _option(args, "--min-readiness", min_readiness) _flag(args, "--gap-report-only", gap_report_only) _option(args, "--bench-candidate", bench_candidate) + _extension_options(args, extension_config, extension_environment) return args @@ -767,6 +1067,8 @@ def _test_args( to: Optional[str], artifact_dir: Optional[PathValue], output: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["test"] _positional(args, root) @@ -778,6 +1080,7 @@ def _test_args( ("--output", output), ): _option(args, flag, value) + _extension_options(args, extension_config, extension_environment) return args @@ -790,6 +1093,8 @@ def _verify_args( output: Optional[PathValue], cases: Optional[PathValue], no_http: bool, + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["verify"] _positional(args, root) @@ -803,9 +1108,61 @@ def _verify_args( ): _option(args, flag, value) _flag(args, "--no-http", no_http) + _extension_options(args, extension_config, extension_environment) return args +def _extension_options( + args: List[str], + configuration: Optional[Mapping[str, JsonValue]], + environment: Sequence[str], +) -> None: + if configuration is not None: + normalized = dict(configuration) + _validate_json_value(normalized) + args.extend( + ( + "--extension-config", + json.dumps(normalized, ensure_ascii=False, separators=(",", ":"), sort_keys=True), + ) + ) + if isinstance(environment, (str, bytes)): + raise ValueError("extension_environment must be a sequence of environment variable names") + for name in environment: + if not isinstance(name, str) or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None: + raise ValueError("extension_environment must contain valid environment variable names") + args.extend(("--extension-env", name)) + + +def _validate_json_value(value: Any, active: Optional[set[int]] = None) -> None: + if value is None or isinstance(value, (bool, int, str)): + return + if isinstance(value, float): + if math.isfinite(value): + return + raise ValueError("extension_config must contain only JSON-compatible values") + elif isinstance(value, list): + items = value + elif isinstance(value, dict): + if all(isinstance(key, str) for key in value): + items = value.values() + else: + raise ValueError("extension_config must contain only JSON-compatible values") + else: + raise ValueError("extension_config must contain only JSON-compatible values") + + active = set() if active is None else active + identity = id(value) + if identity in active: + raise ValueError("extension_config must contain only JSON-compatible values") + active.add(identity) + try: + for item in items: + _validate_json_value(item, active) + finally: + active.remove(identity) + + def _positional(args: List[str], value: Optional[PathValue]) -> None: if value is not None: args.append(os.fspath(value)) @@ -821,7 +1178,9 @@ def _flag(args: List[str], flag: str, enabled: bool) -> None: args.append(flag) -def _finish_result(stdout: str, *, command: str, exit_code: int, stderr: str) -> SankaMigrateResult[Any]: +def _finish_result( + stdout: str, *, command: SankaMigrateCommand, exit_code: int, stderr: str +) -> SankaMigrateResult[Any]: result = _decode_result( stdout, command=command, @@ -841,12 +1200,13 @@ def _finish_result(stdout: str, *, command: str, exit_code: int, stderr: str) -> command=command, exit_code=exit_code, parsed_error=error_data, + result=result, stderr=stderr, ) return result -def _missing_executable(command: str) -> SankaMigrateError: +def _missing_executable(command: SankaMigrateCommand) -> SankaMigrateError: return SankaMigrateError( "sanka-migrate executable was not found; install it with " "`uv tool install sanka-migrate` or pass executable=...", @@ -857,7 +1217,7 @@ def _missing_executable(command: str) -> SankaMigrateError: def _decode_result( stdout: str, *, - command: str, + command: SankaMigrateCommand, exit_code: int, stderr: str, ) -> SankaMigrateResult[Dict[str, Any]]: @@ -895,12 +1255,44 @@ def _decode_result( stderr=stderr, ) + outcome = payload.get("outcome") + if outcome not in ("success", "error"): + raise _invalid_field(command, exit_code, stderr, "outcome", "'success' or 'error'") + if exit_code not in (0, 1, 2): + raise SankaMigrateError( + "invalid sanka-migrate exit code {}; expected 0, 1, or 2".format(exit_code), + command=command, + exit_code=exit_code, + stderr=stderr, + ) + if (outcome == "success") != (exit_code == 0): + raise SankaMigrateError( + "sanka-migrate outcome {!r} is inconsistent with exit code {}".format(outcome, exit_code), + command=command, + exit_code=exit_code, + stderr=stderr, + ) + data = payload.get("data") artifacts = payload.get("artifacts") limitations = payload.get("limitations") next_actions = payload.get("next_actions") if not isinstance(data, dict): raise _invalid_field(command, exit_code, stderr, "data", "an object") + error_data = data.get("error") + if outcome == "success": + if "error" in data: + raise _invalid_field(command, exit_code, stderr, "data.error", "absent on success") + else: + if not isinstance(error_data, dict): + raise _invalid_field(command, exit_code, stderr, "data.error", "an object") + for name in ("code", "message"): + if not isinstance(error_data.get(name), str): + raise _invalid_field( + command, exit_code, stderr, "data.error." + name, "a string" + ) + if "details" in error_data and not isinstance(error_data["details"], dict): + raise _invalid_field(command, exit_code, stderr, "data.error.details", "an object") for name, value in ( ("artifacts", artifacts), ("limitations", limitations), @@ -908,15 +1300,13 @@ def _decode_result( ): if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise _invalid_field(command, exit_code, stderr, name, "a string array") - if not isinstance(payload.get("outcome"), str): - raise _invalid_field(command, exit_code, stderr, "outcome", "a string") if not isinstance(payload.get("migration_state"), str): raise _invalid_field(command, exit_code, stderr, "migration_state", "a string") return SankaMigrateResult( schema_version=schema_version, command=payload_command, - outcome=payload["outcome"], + outcome=outcome, migration_state=payload["migration_state"], data=data, artifacts=artifacts, @@ -926,7 +1316,7 @@ def _decode_result( def _invalid_field( - command: str, + command: SankaMigrateCommand, exit_code: int, stderr: str, name: str, diff --git a/pyproject.toml b/pyproject.toml index 39354bf..bf905bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "sanka-sdk" -version = "0.2.5" +version = "0.2.6" description = "Python SDK for the Sanka API." readme = "README.md" requires-python = ">=3.9" diff --git a/src/sanka_sdk/migrate.py b/src/sanka_sdk/migrate.py index 49cebe3..6c20e4d 100644 --- a/src/sanka_sdk/migrate.py +++ b/src/sanka_sdk/migrate.py @@ -10,19 +10,28 @@ import asyncio import json +import math import os +import re import subprocess from dataclasses import dataclass from typing import Any, Dict, Generic, List, Literal, Mapping, Optional, Sequence, Tuple, TypedDict, TypeVar, Union PathValue = Union[str, "os.PathLike[str]"] +JsonValue = Union[None, bool, int, float, str, List["JsonValue"], Dict[str, "JsonValue"]] +SankaMigrateCommand = Literal["scan", "plan", "apply", "test", "verify", "extension"] CLI_SCHEMA_VERSION = "sanka-cli/v1" __all__ = [ "ApplyData", "AsyncSankaMigrate", + "ExtensionEvidence", + "ExtensionFailure", + "ExtensionRecommendation", + "JsonValue", "PlanData", "SankaMigrate", + "SankaMigrateCommand", "SankaMigrateError", "SankaMigrateResult", "ScanData", @@ -33,10 +42,42 @@ TData = TypeVar("TData") +class ExtensionEvidence(TypedDict): + """Static project evidence that matched an extension recommendation.""" + + kind: str + value: str + path: str + + +class ExtensionRecommendation(TypedDict): + """One compatible extension recommended by ``sanka-migrate``.""" + + id: str + version: str + marketplace: str + targets: List[str] + evidence: List[ExtensionEvidence] + status: List[str] + add_command: str + + +class _ExtensionFailureRequired(TypedDict): + code: str + message: str + + +class ExtensionFailure(_ExtensionFailureRequired, total=False): + """Structured extension or marketplace failure returned by the CLI.""" + + details: Dict[str, JsonValue] + + class ScanData(TypedDict, total=False): """Core semantic scan fields; additional CLI fields remain available.""" scan_hash: str + recommendations: List[ExtensionRecommendation] class PlanData(TypedDict): @@ -79,7 +120,7 @@ class SankaMigrateResult(Generic[TData]): """ schema_version: str - command: str + command: SankaMigrateCommand outcome: str migration_state: str data: TData @@ -95,6 +136,7 @@ class SankaMigrateError(RuntimeError): command: Generic command that failed. exit_code: Process exit code, or ``None`` when the CLI could not start. parsed_error: Structured CLI error from ``data.error``, when available. + result: Complete valid failure envelope, when the CLI returned one. stderr: Diagnostic text written by the CLI. """ @@ -102,15 +144,17 @@ def __init__( self, message: str, *, - command: str, + command: SankaMigrateCommand, exit_code: Optional[int] = None, parsed_error: Optional[Dict[str, Any]] = None, + result: Optional[SankaMigrateResult[Dict[str, Any]]] = None, stderr: str = "", ) -> None: super().__init__(message) self.command = command self.exit_code = exit_code self.parsed_error = parsed_error + self.result = result self.stderr = stderr @@ -149,6 +193,8 @@ def scan( root: Optional[PathValue] = None, settings: Optional[str] = None, artifact_dir: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ScanData]: """Inspect a source application with ``sanka-migrate scan``. @@ -156,6 +202,8 @@ def scan( root: Source repository root. Omit it to use ``cwd``. settings: Explicit Django settings module; otherwise the CLI detects it. artifact_dir: Directory for the semantic scan artifact. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The scan result, discovered application data, risks, and artifact paths. @@ -170,7 +218,10 @@ def scan( ``scan = SankaMigrate(cwd="./app").scan()`` """ - return self._run("scan", _scan_args(root, settings, artifact_dir)) + return self._run( + "scan", + _scan_args(root, settings, artifact_dir, extension_config, extension_environment), + ) def plan( self, @@ -178,13 +229,15 @@ def plan( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, strategy: Optional[Literal["native", "compatibility"]] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, generation: Optional[Literal["full", "update", "minimal"]] = None, package_manager: Optional[Literal["uv", "pip"]] = None, orm: Optional[Literal["tortoise", "sqlalchemy", "psycopg"]] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[PlanData]: """Create a reviewable, hash-bound plan with ``sanka-migrate plan``. @@ -192,13 +245,15 @@ def plan( root: Source repository root. Omit it to use ``cwd``. file: Migration spec passed as ``--file``. state: Run-state SQLite file passed as ``--state``. - to: Target framework, currently ``"fastapi"`` for application migration. + to: Target advertised by an installed extension. strategy: Runtime strategy, currently ``"native"`` or ``"compatibility"``. artifact_dir: Directory containing scan and plan artifacts. output: Planned generated target directory. generation: Generation mode: ``"full"``, ``"update"``, or ``"minimal"``. package_manager: Generated environment manager: ``"uv"`` or ``"pip"``. orm: ORM selected when the scan detects database-backed routes. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The plan result. Use ``result.data["plan_hash"]`` for ``apply()``. @@ -227,6 +282,8 @@ def plan( generation, package_manager, orm, + extension_config, + extension_environment, ), ) @@ -237,7 +294,7 @@ def apply( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, force: bool = False, @@ -245,6 +302,8 @@ def apply( min_readiness: Optional[float] = None, gap_report_only: bool = False, bench_candidate: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ApplyData]: """Apply exactly one reviewed plan with ``sanka-migrate apply``. @@ -261,6 +320,8 @@ def apply( min_readiness: Minimum native readiness percentage from 0 through 100. gap_report_only: Write a gap report instead of generating an application. bench_candidate: Also write a Migration Bench candidate here. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The apply result and paths written from the reviewed plan. @@ -291,6 +352,8 @@ def apply( min_readiness, gap_report_only, bench_candidate, + extension_config, + extension_environment, ), ) @@ -300,9 +363,11 @@ def test( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[TestData]: """Run generated-target tests with ``sanka-migrate test``. @@ -313,6 +378,8 @@ def test( to: Target framework selector. artifact_dir: Directory containing the applied plan. output: Generated target directory. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Test verdict, target interpreter, dependencies, and generated test artifact. @@ -328,7 +395,19 @@ def test( ``tested = SankaMigrate(cwd="./source").test()`` """ - return self._run("test", _test_args(root, file, state, to, artifact_dir, output)) + return self._run( + "test", + _test_args( + root, + file, + state, + to, + artifact_dir, + output, + extension_config, + extension_environment, + ), + ) def verify( self, @@ -336,11 +415,13 @@ def verify( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, cases: Optional[PathValue] = None, no_http: bool = False, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[VerifyData]: """Verify the selected migration with ``sanka-migrate verify``. @@ -353,6 +434,8 @@ def verify( output: Generated target directory. cases: JSON file containing additional read-only HTTP verification cases. no_http: Skip HTTP probes when structural verification is sufficient. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Verification verdict, checked scope, artifacts, and limitations. @@ -370,10 +453,27 @@ def verify( return self._run( "verify", - _verify_args(root, file, state, to, artifact_dir, output, cases, no_http), + _verify_args( + root, + file, + state, + to, + artifact_dir, + output, + cases, + no_http, + extension_config, + extension_environment, + ), ) - def _run(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: + @property + def extensions(self) -> _SankaExtensions: + """Extension and marketplace management commands.""" + + return _SankaExtensions(self) + + def _run(self, command: SankaMigrateCommand, args: Sequence[str]) -> SankaMigrateResult[Any]: argv, environment = self._prepare(command, args) try: completed = subprocess.run( @@ -399,7 +499,9 @@ def _run(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: stderr=completed.stderr, ) - def _prepare(self, command: str, args: Sequence[str]) -> Tuple[List[str], Dict[str, str]]: + def _prepare( + self, command: SankaMigrateCommand, args: Sequence[str] + ) -> Tuple[List[str], Dict[str, str]]: if self.cwd is not None and not os.path.isdir(self.cwd): raise SankaMigrateError( "sanka-migrate working directory was not found: {}".format(self.cwd), @@ -428,6 +530,8 @@ async def scan( root: Optional[PathValue] = None, settings: Optional[str] = None, artifact_dir: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ScanData]: """Asynchronously inspect a source with ``sanka-migrate scan``. @@ -435,6 +539,8 @@ async def scan( root: Source repository root. Omit it to use ``cwd``. settings: Explicit Django settings module. artifact_dir: Directory for the semantic scan artifact. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The scan result, discovered application data, risks, and artifacts. @@ -449,7 +555,10 @@ async def scan( ``scan = await AsyncSankaMigrate(cwd="./app").scan()`` """ - return await self._run_async("scan", _scan_args(root, settings, artifact_dir)) + return await self._run_async( + "scan", + _scan_args(root, settings, artifact_dir, extension_config, extension_environment), + ) async def plan( self, @@ -457,13 +566,15 @@ async def plan( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, strategy: Optional[Literal["native", "compatibility"]] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, generation: Optional[Literal["full", "update", "minimal"]] = None, package_manager: Optional[Literal["uv", "pip"]] = None, orm: Optional[Literal["tortoise", "sqlalchemy", "psycopg"]] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[PlanData]: """Asynchronously create a hash-bound plan with ``sanka-migrate plan``. @@ -471,13 +582,15 @@ async def plan( root: Source repository root. Omit it to use ``cwd``. file: Migration spec passed as ``--file``. state: Run-state SQLite file passed as ``--state``. - to: Target framework, currently ``"fastapi"``. + to: Target advertised by an installed extension. strategy: ``"native"`` or ``"compatibility"``. artifact_dir: Directory containing scan and plan artifacts. output: Planned generated target directory. generation: ``"full"``, ``"update"``, or ``"minimal"``. package_manager: ``"uv"`` or ``"pip"``. orm: ORM selected for database-backed routes. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The plan result whose ``data["plan_hash"]`` is required by ``apply``. @@ -505,6 +618,8 @@ async def plan( generation, package_manager, orm, + extension_config, + extension_environment, ), ) @@ -515,7 +630,7 @@ async def apply( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, force: bool = False, @@ -523,6 +638,8 @@ async def apply( min_readiness: Optional[float] = None, gap_report_only: bool = False, bench_candidate: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[ApplyData]: """Asynchronously apply one reviewed plan with ``sanka-migrate apply``. @@ -539,6 +656,8 @@ async def apply( min_readiness: Minimum native readiness percentage. gap_report_only: Write a gap report instead of an application. bench_candidate: Also write a Migration Bench candidate here. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: The apply result and paths written from the reviewed plan. @@ -569,6 +688,8 @@ async def apply( min_readiness, gap_report_only, bench_candidate, + extension_config, + extension_environment, ), ) @@ -578,9 +699,11 @@ async def test( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[TestData]: """Asynchronously run generated tests with ``sanka-migrate test``. @@ -591,6 +714,8 @@ async def test( to: Target framework selector. artifact_dir: Directory containing the applied plan. output: Generated target directory. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Test verdict, target interpreter, dependencies, and test artifact. @@ -605,7 +730,19 @@ async def test( ``tested = await AsyncSankaMigrate(cwd="./source").test()`` """ - return await self._run_async("test", _test_args(root, file, state, to, artifact_dir, output)) + return await self._run_async( + "test", + _test_args( + root, + file, + state, + to, + artifact_dir, + output, + extension_config, + extension_environment, + ), + ) async def verify( self, @@ -613,11 +750,13 @@ async def verify( root: Optional[PathValue] = None, file: Optional[PathValue] = None, state: Optional[PathValue] = None, - to: Optional[Literal["fastapi"]] = None, + to: Optional[str] = None, artifact_dir: Optional[PathValue] = None, output: Optional[PathValue] = None, cases: Optional[PathValue] = None, no_http: bool = False, + extension_config: Optional[Mapping[str, JsonValue]] = None, + extension_environment: Sequence[str] = (), ) -> SankaMigrateResult[VerifyData]: """Asynchronously verify with ``sanka-migrate verify``. @@ -630,6 +769,8 @@ async def verify( output: Generated target directory. cases: JSON file with additional read-only HTTP cases. no_http: Skip HTTP probes when structural checks are sufficient. + extension_config: JSON-compatible settings for the selected extension. + extension_environment: Ambient environment variable names to forward. Returns: Verification verdict, checked scope, artifacts, and limitations. @@ -646,10 +787,29 @@ async def verify( return await self._run_async( "verify", - _verify_args(root, file, state, to, artifact_dir, output, cases, no_http), + _verify_args( + root, + file, + state, + to, + artifact_dir, + output, + cases, + no_http, + extension_config, + extension_environment, + ), ) - async def _run_async(self, command: str, args: Sequence[str]) -> SankaMigrateResult[Any]: + @property + def extensions(self) -> _AsyncSankaExtensions: + """Asynchronous extension and marketplace management commands.""" + + return _AsyncSankaExtensions(self) + + async def _run_async( + self, command: SankaMigrateCommand, args: Sequence[str] + ) -> SankaMigrateResult[Any]: argv, environment = self._prepare(command, args) try: process = await asyncio.create_subprocess_exec( @@ -689,11 +849,145 @@ async def _run_async(self, command: str, args: Sequence[str]) -> SankaMigrateRes ) -def _scan_args(root: Optional[PathValue], settings: Optional[str], artifact_dir: Optional[PathValue]) -> List[str]: +class _SankaExtensionMarketplaces: + def __init__(self, migrate: SankaMigrate) -> None: + self._migrate = migrate + + def add( + self, + source: PathValue, + *, + name: Optional[str] = None, + trust: bool = False, + ) -> SankaMigrateResult[Dict[str, Any]]: + """Add an immutable marketplace snapshot, explicitly trusting it when requested.""" + + return self._migrate._run("extension", _marketplace_add_args(source, name, trust)) + + def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """List configured marketplace snapshots.""" + + return self._migrate._run("extension", ["extension", "marketplace", "list"]) + + def upgrade(self, name: Optional[str] = None) -> SankaMigrateResult[Dict[str, Any]]: + """Refresh one marketplace, or all marketplaces when ``name`` is omitted.""" + + args = ["extension", "marketplace", "upgrade"] + if name is not None: + args.append(name) + return self._migrate._run("extension", args) + + def remove(self, name: str) -> SankaMigrateResult[Dict[str, Any]]: + """Remove an unused marketplace snapshot.""" + + return self._migrate._run("extension", ["extension", "marketplace", "remove", name]) + + +class _SankaExtensions: + def __init__(self, migrate: SankaMigrate) -> None: + self._migrate = migrate + self.marketplaces = _SankaExtensionMarketplaces(migrate) + + def add( + self, extension_id: str, *, marketplace: Optional[str] = None + ) -> SankaMigrateResult[Dict[str, Any]]: + """Install and lock an extension, optionally selecting its marketplace.""" + + args = ["extension", "add", extension_id] + _option(args, "--marketplace", marketplace) + return self._migrate._run("extension", args) + + def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """List available and installed extensions.""" + + return self._migrate._run("extension", ["extension", "list"]) + + def remove(self, extension_id: str) -> SankaMigrateResult[Dict[str, Any]]: + """Unpin or disable an extension in the current project.""" + + return self._migrate._run("extension", ["extension", "remove", extension_id]) + + +class _AsyncSankaExtensionMarketplaces: + def __init__(self, migrate: AsyncSankaMigrate) -> None: + self._migrate = migrate + + async def add( + self, + source: PathValue, + *, + name: Optional[str] = None, + trust: bool = False, + ) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously add an immutable marketplace snapshot.""" + + return await self._migrate._run_async("extension", _marketplace_add_args(source, name, trust)) + + async def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously list configured marketplace snapshots.""" + + return await self._migrate._run_async("extension", ["extension", "marketplace", "list"]) + + async def upgrade(self, name: Optional[str] = None) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously refresh one marketplace, or all when omitted.""" + + args = ["extension", "marketplace", "upgrade"] + if name is not None: + args.append(name) + return await self._migrate._run_async("extension", args) + + async def remove(self, name: str) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously remove an unused marketplace snapshot.""" + + return await self._migrate._run_async( + "extension", ["extension", "marketplace", "remove", name] + ) + + +class _AsyncSankaExtensions: + def __init__(self, migrate: AsyncSankaMigrate) -> None: + self._migrate = migrate + self.marketplaces = _AsyncSankaExtensionMarketplaces(migrate) + + async def add( + self, extension_id: str, *, marketplace: Optional[str] = None + ) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously install and lock an extension.""" + + args = ["extension", "add", extension_id] + _option(args, "--marketplace", marketplace) + return await self._migrate._run_async("extension", args) + + async def list(self) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously list available and installed extensions.""" + + return await self._migrate._run_async("extension", ["extension", "list"]) + + async def remove(self, extension_id: str) -> SankaMigrateResult[Dict[str, Any]]: + """Asynchronously unpin or disable an extension.""" + + return await self._migrate._run_async("extension", ["extension", "remove", extension_id]) + + +def _marketplace_add_args(source: PathValue, name: Optional[str], trust: bool) -> List[str]: + args = ["extension", "marketplace", "add", os.fspath(source)] + _option(args, "--name", name) + _flag(args, "--trust", trust) + return args + + +def _scan_args( + root: Optional[PathValue], + settings: Optional[str], + artifact_dir: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], +) -> List[str]: args = ["scan"] _positional(args, root) _option(args, "--settings", settings) _option(args, "--artifact-dir", artifact_dir) + _extension_options(args, extension_config, extension_environment) return args @@ -708,6 +1002,8 @@ def _plan_args( generation: Optional[str], package_manager: Optional[str], orm: Optional[str], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["plan"] _positional(args, root) @@ -723,6 +1019,7 @@ def _plan_args( ("--orm", orm), ): _option(args, flag, value) + _extension_options(args, extension_config, extension_environment) return args @@ -739,6 +1036,8 @@ def _apply_args( min_readiness: Optional[float], gap_report_only: bool, bench_candidate: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: if not plan_hash.strip(): raise ValueError("plan_hash must not be empty") @@ -757,6 +1056,7 @@ def _apply_args( _option(args, "--min-readiness", min_readiness) _flag(args, "--gap-report-only", gap_report_only) _option(args, "--bench-candidate", bench_candidate) + _extension_options(args, extension_config, extension_environment) return args @@ -767,6 +1067,8 @@ def _test_args( to: Optional[str], artifact_dir: Optional[PathValue], output: Optional[PathValue], + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["test"] _positional(args, root) @@ -778,6 +1080,7 @@ def _test_args( ("--output", output), ): _option(args, flag, value) + _extension_options(args, extension_config, extension_environment) return args @@ -790,6 +1093,8 @@ def _verify_args( output: Optional[PathValue], cases: Optional[PathValue], no_http: bool, + extension_config: Optional[Mapping[str, JsonValue]], + extension_environment: Sequence[str], ) -> List[str]: args = ["verify"] _positional(args, root) @@ -803,9 +1108,61 @@ def _verify_args( ): _option(args, flag, value) _flag(args, "--no-http", no_http) + _extension_options(args, extension_config, extension_environment) return args +def _extension_options( + args: List[str], + configuration: Optional[Mapping[str, JsonValue]], + environment: Sequence[str], +) -> None: + if configuration is not None: + normalized = dict(configuration) + _validate_json_value(normalized) + args.extend( + ( + "--extension-config", + json.dumps(normalized, ensure_ascii=False, separators=(",", ":"), sort_keys=True), + ) + ) + if isinstance(environment, (str, bytes)): + raise ValueError("extension_environment must be a sequence of environment variable names") + for name in environment: + if not isinstance(name, str) or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None: + raise ValueError("extension_environment must contain valid environment variable names") + args.extend(("--extension-env", name)) + + +def _validate_json_value(value: Any, active: Optional[set[int]] = None) -> None: + if value is None or isinstance(value, (bool, int, str)): + return + if isinstance(value, float): + if math.isfinite(value): + return + raise ValueError("extension_config must contain only JSON-compatible values") + elif isinstance(value, list): + items = value + elif isinstance(value, dict): + if all(isinstance(key, str) for key in value): + items = value.values() + else: + raise ValueError("extension_config must contain only JSON-compatible values") + else: + raise ValueError("extension_config must contain only JSON-compatible values") + + active = set() if active is None else active + identity = id(value) + if identity in active: + raise ValueError("extension_config must contain only JSON-compatible values") + active.add(identity) + try: + for item in items: + _validate_json_value(item, active) + finally: + active.remove(identity) + + def _positional(args: List[str], value: Optional[PathValue]) -> None: if value is not None: args.append(os.fspath(value)) @@ -821,7 +1178,9 @@ def _flag(args: List[str], flag: str, enabled: bool) -> None: args.append(flag) -def _finish_result(stdout: str, *, command: str, exit_code: int, stderr: str) -> SankaMigrateResult[Any]: +def _finish_result( + stdout: str, *, command: SankaMigrateCommand, exit_code: int, stderr: str +) -> SankaMigrateResult[Any]: result = _decode_result( stdout, command=command, @@ -841,12 +1200,13 @@ def _finish_result(stdout: str, *, command: str, exit_code: int, stderr: str) -> command=command, exit_code=exit_code, parsed_error=error_data, + result=result, stderr=stderr, ) return result -def _missing_executable(command: str) -> SankaMigrateError: +def _missing_executable(command: SankaMigrateCommand) -> SankaMigrateError: return SankaMigrateError( "sanka-migrate executable was not found; install it with " "`uv tool install sanka-migrate` or pass executable=...", @@ -857,7 +1217,7 @@ def _missing_executable(command: str) -> SankaMigrateError: def _decode_result( stdout: str, *, - command: str, + command: SankaMigrateCommand, exit_code: int, stderr: str, ) -> SankaMigrateResult[Dict[str, Any]]: @@ -895,12 +1255,44 @@ def _decode_result( stderr=stderr, ) + outcome = payload.get("outcome") + if outcome not in ("success", "error"): + raise _invalid_field(command, exit_code, stderr, "outcome", "'success' or 'error'") + if exit_code not in (0, 1, 2): + raise SankaMigrateError( + "invalid sanka-migrate exit code {}; expected 0, 1, or 2".format(exit_code), + command=command, + exit_code=exit_code, + stderr=stderr, + ) + if (outcome == "success") != (exit_code == 0): + raise SankaMigrateError( + "sanka-migrate outcome {!r} is inconsistent with exit code {}".format(outcome, exit_code), + command=command, + exit_code=exit_code, + stderr=stderr, + ) + data = payload.get("data") artifacts = payload.get("artifacts") limitations = payload.get("limitations") next_actions = payload.get("next_actions") if not isinstance(data, dict): raise _invalid_field(command, exit_code, stderr, "data", "an object") + error_data = data.get("error") + if outcome == "success": + if "error" in data: + raise _invalid_field(command, exit_code, stderr, "data.error", "absent on success") + else: + if not isinstance(error_data, dict): + raise _invalid_field(command, exit_code, stderr, "data.error", "an object") + for name in ("code", "message"): + if not isinstance(error_data.get(name), str): + raise _invalid_field( + command, exit_code, stderr, "data.error." + name, "a string" + ) + if "details" in error_data and not isinstance(error_data["details"], dict): + raise _invalid_field(command, exit_code, stderr, "data.error.details", "an object") for name, value in ( ("artifacts", artifacts), ("limitations", limitations), @@ -908,15 +1300,13 @@ def _decode_result( ): if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise _invalid_field(command, exit_code, stderr, name, "a string array") - if not isinstance(payload.get("outcome"), str): - raise _invalid_field(command, exit_code, stderr, "outcome", "a string") if not isinstance(payload.get("migration_state"), str): raise _invalid_field(command, exit_code, stderr, "migration_state", "a string") return SankaMigrateResult( schema_version=schema_version, command=payload_command, - outcome=payload["outcome"], + outcome=outcome, migration_state=payload["migration_state"], data=data, artifacts=artifacts, @@ -926,7 +1316,7 @@ def _decode_result( def _invalid_field( - command: str, + command: SankaMigrateCommand, exit_code: int, stderr: str, name: str, diff --git a/tests/test_migrate.py b/tests/test_migrate.py index 5fa7a87..9c25f0e 100644 --- a/tests/test_migrate.py +++ b/tests/test_migrate.py @@ -7,6 +7,7 @@ import textwrap import unittest from pathlib import Path +from typing import get_args import sanka_sdk.migrate as migrate_module from sanka_sdk.migrate import AsyncSankaMigrate, SankaMigrate, SankaMigrateError @@ -21,6 +22,7 @@ command = sys.argv[1] mode = os.environ.get("FAKE_SANKA_MODE", "success") +is_error = mode in ("error", "trust-error") pid_file = os.environ.get("FAKE_SANKA_PID_FILE") if pid_file: with open(pid_file, "w", encoding="utf-8") as file: @@ -36,13 +38,44 @@ payload = { "schema_version": "wrong/v1" if mode == "wrong-schema" else "sanka-cli/v1", "command": "wrong" if mode == "wrong-command" else command, - "outcome": "error" if mode == "error" else "success", - "migration_state": "failed" if mode == "error" else "complete", + "outcome": os.environ.get("FAKE_SANKA_OUTCOME", "error" if is_error else "success"), + "migration_state": "failed" if is_error else "complete", "data": { "argv": sys.argv[1:], **( - {"error": {"code": "SANKA_USAGE", "message": "bad option"}} - if mode == "error" + { + "error": { + "code": "SANKA_MARKETPLACE_TRUST_REQUIRED", + "message": "explicit trust is required", + "details": {"identity": "local:/third-party"}, + } + } + if mode == "trust-error" + else {"error": {"code": "SANKA_USAGE", "message": "bad option"}} + if is_error + else {} + ), + **( + { + "recommendations": [ + { + "id": "sanka/drf-to-fastapi", + "version": "0.1.0a1", + "marketplace": "official", + "targets": ["fastapi"], + "evidence": [ + { + "kind": "declared_dependency", + "value": "djangorestframework", + "path": "requirements.txt", + } + ], + "status": ["available"], + "add_command": "sanka-migrate extension add sanka/drf-to-fastapi", + } + ] + } + if mode == "recommendations" else {} ), }, @@ -50,8 +83,13 @@ "limitations": [], "next_actions": [], } +error_case = os.environ.get("FAKE_SANKA_ERROR_CASE") +if error_case == "missing": + payload["data"].pop("error", None) +elif error_case: + payload["data"]["error"] = json.loads(error_case) print(json.dumps(payload)) -raise SystemExit(int(os.environ.get("FAKE_SANKA_EXIT", "2" if mode == "error" else "0"))) +raise SystemExit(int(os.environ.get("FAKE_SANKA_EXIT", "2" if is_error else "0"))) """ @@ -244,6 +282,164 @@ def test_arguments_are_not_interpreted_by_a_shell(self) -> None: self.assertIn(root, self.argv(result)) self.assertFalse(marker.exists()) + def test_extension_configuration_accepts_arbitrary_targets_and_stable_unicode_json(self) -> None: + result = self.migrate.plan( + to="vendor/flask-v2", + extension_config={ + "z": {"日本語": ["値", True, None]}, + "a": 1, + }, + extension_environment=("DJANGO_SECRET_KEY", "API_TOKEN"), + ) + + self.assertEqual( + self.argv(result), + [ + "plan", + "--to", + "vendor/flask-v2", + "--extension-config", + '{"a":1,"z":{"日本語":["値",true,null]}}', + "--extension-env", + "DJANGO_SECRET_KEY", + "--extension-env", + "API_TOKEN", + "--json", + ], + ) + + def test_extension_configuration_is_recursively_validated_before_spawn(self) -> None: + invalid_values = ( + {"nested": object()}, + {"nested": [{"bad": {"not-json"}}]}, + {1: "non-string key"}, + {"non-finite": float("nan")}, + ) + + for value in invalid_values: + with self.subTest(value=value): + with self.assertRaisesRegex(ValueError, "JSON-compatible"): + self.migrate.plan(extension_config=value) # type: ignore[arg-type] + + def test_extension_configuration_rejects_cycles_but_allows_shared_containers(self) -> None: + shared = ["value"] + self.assertEqual( + self.argv(self.migrate.plan(extension_config={"first": shared, "second": shared})), + [ + "plan", + "--extension-config", + '{"first":["value"],"second":["value"]}', + "--json", + ], + ) + + cyclic_list: list[object] = [] + cyclic_list.append(cyclic_list) + cyclic_dict: dict[str, object] = {} + cyclic_dict["self"] = cyclic_dict + for value in (cyclic_list, cyclic_dict): + with self.subTest(value=type(value).__name__): + with self.assertRaisesRegex(ValueError, "JSON-compatible"): + self.migrate.plan(extension_config={"cycle": value}) # type: ignore[dict-item] + + def test_extension_environment_rejects_invalid_containers_and_names_before_spawn(self) -> None: + migrate = SankaMigrate(cwd=self.root, executable=self.root / "missing") + for environment in ( + "API_TOKEN", + b"API_TOKEN", + ("API-TOKEN",), + ("9API_TOKEN",), + ("API_TÖKEN",), + ): + with self.subTest(environment=environment): + with self.assertRaisesRegex(ValueError, "extension_environment"): + migrate.plan(extension_environment=environment) # type: ignore[arg-type] + + def test_sync_extension_management_has_full_grouped_parity_and_no_shell(self) -> None: + marker = self.root / "unexpected-extension" + source = "$(touch {})".format(marker) + results = [ + self.migrate.extensions.add("example/demo", marketplace="third-party"), + self.migrate.extensions.list(), + self.migrate.extensions.remove("example/demo"), + self.migrate.extensions.marketplaces.add(source, name="third-party", trust=True), + self.migrate.extensions.marketplaces.list(), + self.migrate.extensions.marketplaces.upgrade("third-party"), + self.migrate.extensions.marketplaces.remove("third-party"), + ] + + self.assertEqual( + [self.argv(result) for result in results], + [ + ["extension", "add", "example/demo", "--marketplace", "third-party", "--json"], + ["extension", "list", "--json"], + ["extension", "remove", "example/demo", "--json"], + [ + "extension", + "marketplace", + "add", + source, + "--name", + "third-party", + "--trust", + "--json", + ], + ["extension", "marketplace", "list", "--json"], + ["extension", "marketplace", "upgrade", "third-party", "--json"], + ["extension", "marketplace", "remove", "third-party", "--json"], + ], + ) + self.assertFalse(marker.exists()) + + def test_recommendations_expose_typed_evidence(self) -> None: + self.assertIn("SankaMigrateCommand", migrate_module.__all__) + for name, keys in ( + ("ExtensionEvidence", {"kind", "path", "value"}), + ( + "ExtensionRecommendation", + {"add_command", "evidence", "id", "marketplace", "status", "targets", "version"}, + ), + ("ExtensionFailure", {"code", "message"}), + ): + definition = getattr(migrate_module, name) + self.assertEqual(set(definition.__required_keys__), keys) + + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={"FAKE_SANKA_MODE": "recommendations"}, + ) + recommendation = migrate.scan().data["recommendations"][0] + self.assertEqual(recommendation["targets"], ["fastapi"]) + self.assertEqual( + recommendation["evidence"], + [ + { + "kind": "declared_dependency", + "value": "djangorestframework", + "path": "requirements.txt", + } + ], + ) + self.assertIn("extension", get_args(migrate_module.SankaMigrateCommand)) + + def test_third_party_trust_failure_keeps_the_complete_result(self) -> None: + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={"FAKE_SANKA_MODE": "trust-error"}, + ) + + with self.assertRaises(SankaMigrateError) as raised: + migrate.extensions.marketplaces.add("/third-party", name="third-party") + + self.assertEqual(raised.exception.parsed_error["code"], "SANKA_MARKETPLACE_TRUST_REQUIRED") + self.assertEqual(raised.exception.result.command, "extension") + self.assertEqual( + raised.exception.result.data["error"]["details"], + {"identity": "local:/third-party"}, + ) + def test_structured_errors_keep_exit_and_cli_details(self) -> None: migrate = SankaMigrate( cwd=self.root, @@ -274,6 +470,101 @@ def test_protocol_rejects_malformed_schema_and_command(self) -> None: with self.assertRaisesRegex(SankaMigrateError, message): migrate.scan() + def test_protocol_rejects_invalid_outcome_and_exit_pairs_without_a_result(self) -> None: + invalid_pairs = ( + ("bogus", "0"), + ("bogus", "1"), + ("bogus", "2"), + ("bogus", "99"), + ("success", "1"), + ("success", "2"), + ("success", "99"), + ("error", "0"), + ("error", "99"), + ) + for outcome, exit_code in invalid_pairs: + with self.subTest(outcome=outcome, exit_code=exit_code): + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={ + "FAKE_SANKA_MODE": "error" if outcome == "error" else "success", + "FAKE_SANKA_OUTCOME": outcome, + "FAKE_SANKA_EXIT": exit_code, + }, + ) + with self.assertRaises(SankaMigrateError) as raised: + migrate.scan() + self.assertIsNone(raised.exception.result) + + def test_protocol_preserves_valid_failure_results_for_supported_exit_codes(self) -> None: + for exit_code in ("1", "2"): + for mode, expected in ( + ("error", {"code": "SANKA_USAGE", "message": "bad option"}), + ( + "trust-error", + { + "code": "SANKA_MARKETPLACE_TRUST_REQUIRED", + "message": "explicit trust is required", + "details": {"identity": "local:/third-party"}, + }, + ), + ): + with self.subTest(exit_code=exit_code, mode=mode): + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={"FAKE_SANKA_MODE": mode, "FAKE_SANKA_EXIT": exit_code}, + ) + with self.assertRaises(SankaMigrateError) as raised: + migrate.scan() + self.assertEqual(raised.exception.parsed_error, expected) + self.assertEqual(raised.exception.result.data["error"], expected) + self.assertEqual(raised.exception.result.outcome, "error") + self.assertEqual(raised.exception.exit_code, int(exit_code)) + + def test_protocol_rejects_malformed_failure_payloads_without_a_result(self) -> None: + malformed_errors = ( + ("missing", "missing"), + ("string", '"failure"'), + ("empty", "{}"), + ("numeric-code", '{"code":7,"message":"bad option"}'), + ("numeric-message", '{"code":"SANKA_USAGE","message":7}'), + ( + "non-object-details", + '{"code":"SANKA_USAGE","message":"bad option","details":[]}', + ), + ) + for exit_code in ("1", "2"): + for case, error_payload in malformed_errors: + with self.subTest(exit_code=exit_code, case=case): + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={ + "FAKE_SANKA_MODE": "error", + "FAKE_SANKA_EXIT": exit_code, + "FAKE_SANKA_ERROR_CASE": error_payload, + }, + ) + with self.assertRaises(SankaMigrateError) as raised: + migrate.scan() + self.assertIsNone(raised.exception.result) + + def test_protocol_rejects_an_error_payload_on_success_without_a_result(self) -> None: + migrate = SankaMigrate( + cwd=self.root, + executable=self.executable, + env={ + "FAKE_SANKA_ERROR_CASE": '{"code":"SANKA_FAILED","message":"not failed"}' + }, + ) + + with self.assertRaises(SankaMigrateError) as raised: + migrate.scan() + + self.assertIsNone(raised.exception.result) + def test_apply_requires_a_reviewed_plan_hash(self) -> None: with self.assertRaisesRegex(ValueError, "plan_hash"): self.migrate.apply(plan_hash="") @@ -285,7 +576,8 @@ def test_missing_executable_has_an_install_hint(self) -> None: def test_every_public_symbol_and_method_has_hover_documentation(self) -> None: for name in migrate_module.__all__: - self.assertTrue(inspect.getdoc(getattr(migrate_module, name)), name) + if name not in ("JsonValue", "SankaMigrateCommand"): + self.assertTrue(inspect.getdoc(getattr(migrate_module, name)), name) for client in (SankaMigrate, AsyncSankaMigrate): for name in ("scan", "plan", "apply", "test", "verify"): documentation = inspect.getdoc(getattr(client, name)) @@ -340,6 +632,62 @@ async def test_every_async_lifecycle_method_uses_the_shared_cli_contract(self) - for result in results: self.assertEqual(self.argv(result)[-1], "--json") + async def test_async_extension_configuration_and_management_match_sync(self) -> None: + results = [ + await self.migrate.plan( + to="vendor/flask-v2", + extension_config={"日本語": {"b": 2, "a": 1}}, + extension_environment=("DJANGO_SECRET_KEY",), + ), + await self.migrate.extensions.add("example/demo", marketplace="third-party"), + await self.migrate.extensions.list(), + await self.migrate.extensions.remove("example/demo"), + await self.migrate.extensions.marketplaces.add( + "https://example.invalid/extensions.git", + name="third-party", + trust=True, + ), + await self.migrate.extensions.marketplaces.list(), + await self.migrate.extensions.marketplaces.upgrade("third-party"), + await self.migrate.extensions.marketplaces.remove("third-party"), + ] + + self.assertEqual( + [self.argv(result) for result in results], + [ + [ + "plan", + "--to", + "vendor/flask-v2", + "--extension-config", + '{"日本語":{"a":1,"b":2}}', + "--extension-env", + "DJANGO_SECRET_KEY", + "--json", + ], + ["extension", "add", "example/demo", "--marketplace", "third-party", "--json"], + ["extension", "list", "--json"], + ["extension", "remove", "example/demo", "--json"], + [ + "extension", + "marketplace", + "add", + "https://example.invalid/extensions.git", + "--name", + "third-party", + "--trust", + "--json", + ], + ["extension", "marketplace", "list", "--json"], + ["extension", "marketplace", "upgrade", "third-party", "--json"], + ["extension", "marketplace", "remove", "third-party", "--json"], + ], + ) + + async def test_async_extension_environment_uses_shared_validation(self) -> None: + with self.assertRaisesRegex(ValueError, "extension_environment"): + await self.migrate.plan(extension_environment="API_TOKEN") # type: ignore[arg-type] + async def test_async_execution_does_not_block_the_event_loop(self) -> None: migrate = AsyncSankaMigrate( cwd=self.root, diff --git a/uv.lock b/uv.lock index ed49b20..5f39718 100644 --- a/uv.lock +++ b/uv.lock @@ -289,7 +289,7 @@ wheels = [ [[package]] name = "sanka-sdk" -version = "0.2.5" +version = "0.2.6" source = { editable = "." } dependencies = [ { name = "httpx" },