diff --git a/README.md b/README.md index b3e66d4..e6d3cc4 100644 --- a/README.md +++ b/README.md @@ -208,7 +208,7 @@ with HonuaAdminClient("https://your-honua-server.com", api_key="honua-api-key") | Sync + async | `HonuaClient` / `AsyncHonuaClient` in lockstep (sync clients generated from the async source of truth) | | Automatic retry | 429/502/503 with exponential backoff and `Retry-After` support; configurable via `max_retries`, `retry_methods` | | Typed errors | `HonuaAuthError`, `HonuaRateLimitError`, `HonuaHttpError`, `HonuaTimeoutError`, `HonuaTransportError` — see [common errors](docs/quickstart.md#common-errors) | -| CLI | `honua` (services / layers / style apply / sanitized `doctor` diagnostics) and `honua-migrate` (offline ArcPy script scan / translate / run, plus `.pyt` / `.atbx` toolbox and GP-service classification) | +| CLI | `honua` (services / layers / style apply / sanitized `doctor` diagnostics) and `honua-migrate` (offline ArcPy script scan / translate / run, plus `.pyt` / `.atbx` toolbox and GP-service classification, with optional server-attested toolbox translation verdicts via `--server`) | | Quality gates | mypy `strict` workspace-wide, 94% coverage gate, public-API [compatibility snapshot](docs/compatibility.md), per-capability [SDK coverage snapshot](docs/sdk-coverage.md), live-server [conformance lane](.github/workflows/conformance.yml) against shared [geospatial-grpc](https://github.com/honua-io/geospatial-grpc) fixtures | ## Documentation diff --git a/compatibility/public-api.json b/compatibility/public-api.json index b0ecec7..4e52530 100644 --- a/compatibility/public-api.json +++ b/compatibility/public-api.json @@ -82,9 +82,20 @@ "ServiceSettingsResponse", "ServiceSummary", "StyleEncoding", + "TOOLBOX_TRANSLATION_ARTIFACT_VERSION", + "TOOLBOX_TRANSLATION_MANIFEST_KIND", + "TOOLBOX_TRANSLATION_REPORT_KIND", "TableDiscoveryResponse", "TableInfo", "TimeInfoResponse", + "ToolboxParameterBinding", + "ToolboxParameterMapping", + "ToolboxToolDescriptor", + "ToolboxToolTranslation", + "ToolboxTranslationIssue", + "ToolboxTranslationManifest", + "ToolboxTranslationReport", + "ToolboxTranslationSummary", "UpdateSecureConnectionRequest", "__version__", "evaluate_admin_compatibility", @@ -960,6 +971,11 @@ "kind": "method", "signature": "(self, *, timeout: 'float | httpx.Timeout | None' = None, extra_headers: 'Mapping[str, str] | None' = None, idempotency_key: 'str | None' = None) -> 'EncryptionValidationResult'" }, + "validate_toolbox_translation": { + "async": true, + "kind": "method", + "signature": "(self, manifest: 'ToolboxTranslationManifest', *, timeout: 'float | httpx.Timeout | None' = None, extra_headers: 'Mapping[str, str] | None' = None, idempotency_key: 'str | None' = None) -> 'ToolboxTranslationReport'" + }, "with_options": { "kind": "method", "signature": "(self, *, timeout: 'float | None' = None, max_retries: 'int | None' = None, base_url: 'str | None' = None) -> 'AsyncHonuaAdminClient'" @@ -1294,6 +1310,10 @@ "kind": "method", "signature": "(self, *, timeout: 'float | httpx.Timeout | None' = None, extra_headers: 'Mapping[str, str] | None' = None, idempotency_key: 'str | None' = None) -> 'EncryptionValidationResult'" }, + "validate_toolbox_translation": { + "kind": "method", + "signature": "(self, manifest: 'ToolboxTranslationManifest', *, timeout: 'float | httpx.Timeout | None' = None, extra_headers: 'Mapping[str, str] | None' = None, idempotency_key: 'str | None' = None) -> 'ToolboxTranslationReport'" + }, "with_options": { "kind": "method", "signature": "(self, *, timeout: 'float | None' = None, max_retries: 'int | None' = None, base_url: 'str | None' = None) -> 'HonuaAdminClient'" @@ -3494,6 +3514,21 @@ "type": "_LiteralGenericAlias", "value": "typing.Literal['mapbox-style', 'sld-1.0', 'sld-1.1']" }, + "TOOLBOX_TRANSLATION_ARTIFACT_VERSION": { + "kind": "constant", + "type": "str", + "value": "'1.0'" + }, + "TOOLBOX_TRANSLATION_MANIFEST_KIND": { + "kind": "constant", + "type": "str", + "value": "'honua.migration.toolbox-translation'" + }, + "TOOLBOX_TRANSLATION_REPORT_KIND": { + "kind": "constant", + "type": "str", + "value": "'honua.migration.toolbox-translation-report'" + }, "TableDiscoveryResponse": { "fields": [ { @@ -3580,6 +3615,304 @@ "qualname": "TimeInfoResponse", "signature": "(start_time_field: 'str | None', end_time_field: 'str | None', track_id_field: 'str | None') -> None" }, + "ToolboxParameterBinding": { + "fields": [ + { + "annotation": "str", + "name": "source_name" + }, + { + "annotation": "str", + "name": "target_parameter" + }, + { + "annotation": "str", + "name": "value_type" + }, + { + "annotation": "bool", + "default": "False", + "name": "required" + } + ], + "kind": "class", + "methods": { + "from_dict": { + "kind": "classmethod", + "signature": "(cls, data: 'dict[str, Any]') -> 'ToolboxParameterBinding'" + }, + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxParameterBinding", + "signature": "(source_name: 'str', target_parameter: 'str', value_type: 'str', required: 'bool' = False) -> None" + }, + "ToolboxParameterMapping": { + "fields": [ + { + "annotation": "str", + "name": "source_name" + }, + { + "annotation": "str", + "name": "target_parameter" + }, + { + "annotation": "str | None", + "default": "None", + "name": "source_data_type" + } + ], + "kind": "class", + "methods": { + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxParameterMapping", + "signature": "(source_name: 'str', target_parameter: 'str', source_data_type: 'str | None' = None) -> None" + }, + "ToolboxToolDescriptor": { + "fields": [ + { + "annotation": "str", + "name": "tool_name" + }, + { + "annotation": "str | None", + "default": "None", + "name": "display_name" + }, + { + "annotation": "str | None", + "default": "None", + "name": "target_process_id" + }, + { + "annotation": "list[ToolboxParameterMapping]", + "defaultFactory": "list", + "name": "parameter_mappings" + }, + { + "annotation": "list[str]", + "defaultFactory": "list", + "name": "unsupported_constructs" + } + ], + "kind": "class", + "methods": { + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxToolDescriptor", + "signature": "(tool_name: 'str', display_name: 'str | None' = None, target_process_id: 'str | None' = None, parameter_mappings: 'list[ToolboxParameterMapping]' = , unsupported_constructs: 'list[str]' = ) -> None" + }, + "ToolboxToolTranslation": { + "fields": [ + { + "annotation": "str", + "name": "tool_name" + }, + { + "annotation": "str", + "name": "classification" + }, + { + "annotation": "str | None", + "default": "None", + "name": "process_id" + }, + { + "annotation": "list[ToolboxParameterBinding]", + "defaultFactory": "list", + "name": "parameter_bindings" + }, + { + "annotation": "list[ToolboxTranslationIssue]", + "defaultFactory": "list", + "name": "issues" + } + ], + "kind": "class", + "methods": { + "from_dict": { + "kind": "classmethod", + "signature": "(cls, data: 'dict[str, Any]') -> 'ToolboxToolTranslation'" + }, + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxToolTranslation", + "signature": "(tool_name: 'str', classification: 'str', process_id: 'str | None' = None, parameter_bindings: 'list[ToolboxParameterBinding]' = , issues: 'list[ToolboxTranslationIssue]' = ) -> None" + }, + "ToolboxTranslationIssue": { + "fields": [ + { + "annotation": "str", + "name": "code" + }, + { + "annotation": "str", + "name": "message" + }, + { + "annotation": "str | None", + "default": "None", + "name": "parameter_name" + } + ], + "kind": "class", + "methods": { + "from_dict": { + "kind": "classmethod", + "signature": "(cls, data: 'dict[str, Any]') -> 'ToolboxTranslationIssue'" + }, + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxTranslationIssue", + "signature": "(code: 'str', message: 'str', parameter_name: 'str | None' = None) -> None" + }, + "ToolboxTranslationManifest": { + "fields": [ + { + "annotation": "str", + "name": "toolbox_name" + }, + { + "annotation": "str", + "name": "source_format" + }, + { + "annotation": "str | None", + "default": "None", + "name": "source_label" + }, + { + "annotation": "list[ToolboxToolDescriptor]", + "defaultFactory": "list", + "name": "tools" + }, + { + "annotation": "str", + "default": "'honua.migration.toolbox-translation'", + "name": "artifact_kind" + }, + { + "annotation": "str", + "default": "'1.0'", + "name": "artifact_version" + } + ], + "kind": "class", + "methods": { + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxTranslationManifest", + "signature": "(toolbox_name: 'str', source_format: 'str', source_label: 'str | None' = None, tools: 'list[ToolboxToolDescriptor]' = , artifact_kind: 'str' = 'honua.migration.toolbox-translation', artifact_version: 'str' = '1.0') -> None" + }, + "ToolboxTranslationReport": { + "fields": [ + { + "annotation": "str", + "name": "toolbox_name" + }, + { + "annotation": "str", + "name": "source_format" + }, + { + "annotation": "ToolboxTranslationSummary", + "defaultFactory": "honua_admin._models.ToolboxTranslationSummary", + "name": "summary" + }, + { + "annotation": "list[ToolboxToolTranslation]", + "defaultFactory": "list", + "name": "tools" + }, + { + "annotation": "str | None", + "default": "None", + "name": "artifact_kind" + }, + { + "annotation": "str | None", + "default": "None", + "name": "artifact_version" + } + ], + "kind": "class", + "methods": { + "from_dict": { + "kind": "classmethod", + "signature": "(cls, data: 'dict[str, Any]') -> 'ToolboxTranslationReport'" + }, + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxTranslationReport", + "signature": "(toolbox_name: 'str', source_format: 'str', summary: 'ToolboxTranslationSummary' = , tools: 'list[ToolboxToolTranslation]' = , artifact_kind: 'str | None' = None, artifact_version: 'str | None' = None) -> None" + }, + "ToolboxTranslationSummary": { + "fields": [ + { + "annotation": "int", + "default": "0", + "name": "tool_count" + }, + { + "annotation": "int", + "default": "0", + "name": "translated_count" + }, + { + "annotation": "int", + "default": "0", + "name": "partially_translated_count" + }, + { + "annotation": "int", + "default": "0", + "name": "unsupported_count" + } + ], + "kind": "class", + "methods": { + "from_dict": { + "kind": "classmethod", + "signature": "(cls, data: 'dict[str, Any]') -> 'ToolboxTranslationSummary'" + }, + "to_dict": { + "kind": "method", + "signature": "(self) -> 'dict[str, Any]'" + } + }, + "module": "honua_admin._models", + "qualname": "ToolboxTranslationSummary", + "signature": "(tool_count: 'int' = 0, translated_count: 'int' = 0, partially_translated_count: 'int' = 0, unsupported_count: 'int' = 0) -> None" + }, "UpdateSecureConnectionRequest": { "fields": [ { diff --git a/docs/honua-gp/codemod-translation-coverage.md b/docs/honua-gp/codemod-translation-coverage.md index 8501192..3febb19 100644 --- a/docs/honua-gp/codemod-translation-coverage.md +++ b/docs/honua-gp/codemod-translation-coverage.md @@ -13,8 +13,8 @@ licensed Esri software. | Input | Reader | CLI | Output | | --- | --- | --- | --- | | `.py` ArcPy script | `scan_arcpy_source` / `translate_arcpy_source` | `honua-migrate scan` / `translate` | `ArcPyScanReport` / `ArcPyMigrationPlan` | -| `.pyt` Python toolbox | `parse_pyt_file` | `honua-migrate pyt` | `PytToolbox` (per-tool `execute` body classified) | -| `.atbx` ModelBuilder toolbox | `parse_atbx_toolbox` | `honua-migrate atbx` | `ModelBuilderToolbox` (models + script-tool names) | +| `.pyt` Python toolbox | `parse_pyt_file` | `honua-migrate pyt` / `translate` | `PytToolbox` (per-tool `execute` body classified) | +| `.atbx` ModelBuilder toolbox | `parse_atbx_toolbox` | `honua-migrate atbx` / `translate` | `ModelBuilderToolbox` (models + script-tool names) | | ArcGIS REST GPServer task defs | `parse_gp_service_definition` / `parse_gp_task_definition` | `honua-migrate gpservice` | `GpService` / `GpTask` | All four share the same registry-driven classification, so a `Buffer` maps to @@ -30,7 +30,12 @@ step, or a GPServer task. external `.py`) are surfaced by name for the caller to scan via the `.py` path. The proprietary binary **`.tbx`** format is **not** parsed (it is not clean-room readable) -- `parse_binary_toolbox`/`parse_atbx_toolbox` raise a - clear redirect. Export `.tbx` to `.atbx` or `.pyt` first. + clear redirect carrying the concrete ArcGIS Pro export steps + (`BINARY_TOOLBOX_EXPORT_GUIDANCE`). This is a standing policy decision, not an + unimplemented stub: Honua never reverse-engineers a proprietary Esri + container, and the migration path is always export-to-open-format -- the same + rule that governs `.loc`/`.lox` locator files. Export `.tbx` to `.atbx` or + `.pyt` first. * **GP-service** definitions are public ArcGIS REST API JSON (`.../GPServer?f=json` and per-task `.../GPServer/?f=json`). @@ -49,6 +54,74 @@ Coverage percentage in the parity-evidence report is gated on *job-executability*, not on how many tools the codemod can parse -- so growing the registry never inflates the runnable-coverage number. +## Server-attested verdicts + +The statuses above are the SDK's **own** view of the Honua process catalog, and +that view can drift from the server that would actually run the job. For a +toolbox, `honua-migrate` can have the server settle it instead: + +```bash +honua-migrate translate roads.pyt --server https://honua.example --api-key "$HONUA_ADMIN_API_KEY" +honua-migrate pyt roads.pyt --server https://honua.example --attestation attest.json +honua-migrate atbx models.atbx --server https://honua.example --require-attested +``` + +Each command builds a translation manifest and posts it to +`POST /api/v1/admin/import/toolbox/translation/validate` (honua-server#2145), +which validates every proposed mapping against the canonical process catalog and +returns a per-tool `translated` / `partially-translated` / `unsupported` +classification with the specific reasons a tool cannot be fully translated. + +The emitted report carries an `attestation` block whose `verdictSource` is +either `server-attested` or `local-only`: + +* **The server wins.** Where the server contradicts the SDK, its verdict is the + effective one; the local verdict is kept beside it and the disagreement is + listed under `disagreements`. A disagreement means the SDK's registry has + drifted from the catalog. +* **A local verdict is never labelled attested.** Running without `--server`, an + unreachable server, refused credentials, or a malformed response all produce a + complete `local-only` report with an explicit `fallbackReason`. There is no + partial attestation -- one failed batch un-attests the whole toolbox. Pass + `--require-attested` to make a local-only verdict a non-zero exit instead. +* **A response only counts as attestation if it is unambiguously one, and is + about the artifact that was submitted.** The report must carry the expected + `artifactKind` and a readable `artifactVersion` (never defaulted in + client-side), its `toolboxName`/`sourceFormat` must match what was sent (so a + stale or misrouted report cannot attest the wrong toolbox just because tool + names happen to collide), and every tool's `classification` must be one of the + three declared values. A 200 that misses either bar -- an error + envelope, a proxy page, a newer server's vocabulary -- degrades to + `local-only`, because an accepted-but-unrecognized classification would show as + a tool's effective verdict while no summary counter tallied it. +* **Every discovered toolbox tool is submitted**, including the ones the reader + could only learn the *name* of: + * `.atbx` script tools -- their logic lives in an external `.py` the reader + does not follow (`script_tool_names`); + * `.atbx` models whose definition yielded no recognizable step + (`unresolved_tool_names`); + * `.pyt` tools listed in `self.tools` whose class is imported rather than + defined in the file (`declared_tool_names` minus the materialised tools). + + Those go in with no proposed target -- nothing was read, so nothing is claimed + -- and come back `unsupported`. An attestation can therefore never cover a + strict subset of the toolbox while claiming to cover all of it. Scan the + referenced script/module with the arcpy `.py` path to classify them properly. + +Attestation is toolbox-scoped, because the endpoint's manifest declares a +toolbox `sourceFormat` (`pyt` / `atbx` / `tbx`). `translate` on a bare arcpy +`.py` script therefore refuses `--server` rather than inventing a format. + +Auth reuses the existing admin credential path (`--api-key`, or +`$HONUA_ADMIN_API_KEY`), since the endpoint is in the admin import group. +Attestation needs the `honua-admin` package installed; without it, the toolbox +still translates and the report is simply marked `local-only`. + +Library entry points: `honua_sdk.migration.build_pyt_translation_manifest` / +`build_atbx_translation_manifest` build the manifest, and +`attest_translation(manifest, validator=...)` merges a server verdict over it. +The validator is injected, so the merge logic itself stays offline and pure. + ## Registered tools Job-executable (`translatable`) targets, gated by `EXECUTABLE_PROCESS_IDS`: @@ -83,7 +156,8 @@ single source of truth; this table is a human-readable projection of it. ## Deferred -* Binary `.tbx` parsing (proprietary, not clean-room readable). +* Binary `.tbx` parsing -- **will not be implemented** (proprietary container; + export-to-open-format is the migration path). * Resolving an `.atbx` script tool's referenced `.py` body automatically (today its name is surfaced; point the `.py` scanner at it). * Compiled .NET / ArcObjects custom tools (separate track). diff --git a/packages/honua-admin/honua_admin/__init__.py b/packages/honua-admin/honua_admin/__init__.py index 2df7c6c..af8e657 100644 --- a/packages/honua-admin/honua_admin/__init__.py +++ b/packages/honua-admin/honua_admin/__init__.py @@ -33,6 +33,9 @@ MINIMUM_SUPPORTED_CONTROL_PLANE_BASE_PATH, MINIMUM_SUPPORTED_SERVER_RELEASE_CHANNEL, MINIMUM_SUPPORTED_SERVER_VERSION, + TOOLBOX_TRANSLATION_ARTIFACT_VERSION, + TOOLBOX_TRANSLATION_MANIFEST_KIND, + TOOLBOX_TRANSLATION_REPORT_KIND, AccessPolicyResponse, AdminCapabilitiesResponse, AdminCompatibilityBaseline, @@ -100,6 +103,14 @@ TableDiscoveryResponse, TableInfo, TimeInfoResponse, + ToolboxParameterBinding, + ToolboxParameterMapping, + ToolboxToolDescriptor, + ToolboxToolTranslation, + ToolboxTranslationIssue, + ToolboxTranslationManifest, + ToolboxTranslationReport, + ToolboxTranslationSummary, UpdateSecureConnectionRequest, evaluate_admin_compatibility, ) @@ -109,6 +120,9 @@ "MINIMUM_SUPPORTED_CONTROL_PLANE_BASE_PATH", "MINIMUM_SUPPORTED_SERVER_RELEASE_CHANNEL", "MINIMUM_SUPPORTED_SERVER_VERSION", + "TOOLBOX_TRANSLATION_ARTIFACT_VERSION", + "TOOLBOX_TRANSLATION_MANIFEST_KIND", + "TOOLBOX_TRANSLATION_REPORT_KIND", "AccessPolicyResponse", "AdminCapabilitiesResponse", "AdminCompatibilityBaseline", @@ -188,6 +202,14 @@ "TableDiscoveryResponse", "TableInfo", "TimeInfoResponse", + "ToolboxParameterBinding", + "ToolboxParameterMapping", + "ToolboxToolDescriptor", + "ToolboxToolTranslation", + "ToolboxTranslationIssue", + "ToolboxTranslationManifest", + "ToolboxTranslationReport", + "ToolboxTranslationSummary", "UpdateSecureConnectionRequest", "__version__", "evaluate_admin_compatibility", diff --git a/packages/honua-admin/honua_admin/_async_client.py b/packages/honua-admin/honua_admin/_async_client.py index cc5b08c..5a56750 100644 --- a/packages/honua-admin/honua_admin/_async_client.py +++ b/packages/honua-admin/honua_admin/_async_client.py @@ -62,6 +62,8 @@ ServiceSummary, StyleEncoding, TableDiscoveryResponse, + ToolboxTranslationManifest, + ToolboxTranslationReport, UpdateSecureConnectionRequest, evaluate_admin_compatibility, style_encoding_media_type, @@ -1093,6 +1095,56 @@ async def scan_migration_source( ) return MigrationSourceInventoryArtifact.from_dict(data) + async def validate_toolbox_translation( + self, + manifest: ToolboxTranslationManifest, + *, + timeout: float | httpx.Timeout | None = None, + extra_headers: Mapping[str, str] | None = None, + idempotency_key: str | None = None, + ) -> ToolboxTranslationReport: + """POST /api/v1/admin/import/toolbox/translation/validate + + Validate an SDK-translated ArcGIS toolbox manifest against the server's + canonical geoprocessing process catalog, and return the + server-authoritative per-tool classification. + + The server owns the round-trip proof: it never parses toolbox sources + and never emulates ``arcpy``. The SDK proposes a mapping; the catalog + decides whether that mapping is ``translated``, + ``partially-translated``, or ``unsupported``, and returns the specific + reasons a tool cannot be fully translated. + + A verdict derived from this call may be presented as *server-attested*. + A verdict the SDK computed on its own may not — see + :func:`honua_sdk.migration.attest_translation`, which builds an + attestation report around this method and degrades explicitly (never + silently) when the server cannot be reached. + + Args: + manifest: The translated toolbox manifest to validate. + + Returns: + The server's :class:`ToolboxTranslationReport`. + + Per-request options (``timeout`` / ``extra_headers`` / + ``idempotency_key``) are forwarded to :meth:`_request`. + + Raises: + HonuaHttpError: The server rejected the manifest (for example an + unsupported ``sourceFormat``) or refused the credentials. + HonuaTransportError: The request failed at the transport layer. + """ + data = await self._request_json( + "POST", + "/api/v1/admin/import/toolbox/translation/validate", + json_body=manifest.to_dict(), + headers=self._idempotency_headers(idempotency_key), + timeout=timeout, + extra_headers=extra_headers, + ) + return ToolboxTranslationReport.from_dict(data) + # ====================================================================== # Connections # ====================================================================== diff --git a/packages/honua-admin/honua_admin/_client.py b/packages/honua-admin/honua_admin/_client.py index be69c61..0c391ed 100644 --- a/packages/honua-admin/honua_admin/_client.py +++ b/packages/honua-admin/honua_admin/_client.py @@ -64,6 +64,8 @@ ServiceSummary, StyleEncoding, TableDiscoveryResponse, + ToolboxTranslationManifest, + ToolboxTranslationReport, UpdateSecureConnectionRequest, evaluate_admin_compatibility, style_encoding_media_type, @@ -1094,6 +1096,56 @@ def scan_migration_source( ) return MigrationSourceInventoryArtifact.from_dict(data) + def validate_toolbox_translation( + self, + manifest: ToolboxTranslationManifest, + *, + timeout: float | httpx.Timeout | None = None, + extra_headers: Mapping[str, str] | None = None, + idempotency_key: str | None = None, + ) -> ToolboxTranslationReport: + """POST /api/v1/admin/import/toolbox/translation/validate + + Validate an SDK-translated ArcGIS toolbox manifest against the server's + canonical geoprocessing process catalog, and return the + server-authoritative per-tool classification. + + The server owns the round-trip proof: it never parses toolbox sources + and never emulates ``arcpy``. The SDK proposes a mapping; the catalog + decides whether that mapping is ``translated``, + ``partially-translated``, or ``unsupported``, and returns the specific + reasons a tool cannot be fully translated. + + A verdict derived from this call may be presented as *server-attested*. + A verdict the SDK computed on its own may not — see + :func:`honua_sdk.migration.attest_translation`, which builds an + attestation report around this method and degrades explicitly (never + silently) when the server cannot be reached. + + Args: + manifest: The translated toolbox manifest to validate. + + Returns: + The server's :class:`ToolboxTranslationReport`. + + Per-request options (``timeout`` / ``extra_headers`` / + ``idempotency_key``) are forwarded to :meth:`_request`. + + Raises: + HonuaHttpError: The server rejected the manifest (for example an + unsupported ``sourceFormat``) or refused the credentials. + HonuaTransportError: The request failed at the transport layer. + """ + data = self._request_json( + "POST", + "/api/v1/admin/import/toolbox/translation/validate", + json_body=manifest.to_dict(), + headers=self._idempotency_headers(idempotency_key), + timeout=timeout, + extra_headers=extra_headers, + ) + return ToolboxTranslationReport.from_dict(data) + # ====================================================================== # Connections # ====================================================================== diff --git a/packages/honua-admin/honua_admin/_models.py b/packages/honua-admin/honua_admin/_models.py index b4db570..5b030b4 100644 --- a/packages/honua-admin/honua_admin/_models.py +++ b/packages/honua-admin/honua_admin/_models.py @@ -1622,11 +1622,207 @@ def to_dict(self) -> dict[str, Any]: } +#: ``artifactKind`` the server requires on an inbound toolbox translation manifest. +TOOLBOX_TRANSLATION_MANIFEST_KIND = "honua.migration.toolbox-translation" + +#: ``artifactKind`` the server stamps on the validation report it returns. +TOOLBOX_TRANSLATION_REPORT_KIND = "honua.migration.toolbox-translation-report" + +#: Manifest schema version this client speaks. The server rejects an +#: unrecognised version rather than reinterpreting the payload under v1. +TOOLBOX_TRANSLATION_ARTIFACT_VERSION = "1.0" + + +@dataclass(frozen=True, slots=True) +class ToolboxParameterBinding: + """Canonical parameter signature the server round-tripped for one mapping.""" + + source_name: str + target_parameter: str + value_type: str + required: bool = False + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ToolboxParameterBinding: + return cls(**_extract_fields(cls, _snake_keys(data))) + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass(frozen=True, slots=True) +class ToolboxTranslationIssue: + """One explicit unsupported-report entry the server attached to a tool. + + ``code`` is a stable machine-readable identifier (for example + ``no-native-executor``, ``unknown-process``, ``missing-required-parameter``); + treat unknown codes as opaque rather than switching on the message text. + """ + + code: str + message: str + parameter_name: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ToolboxTranslationIssue: + return cls(**_extract_fields(cls, _snake_keys(data))) + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass(frozen=True, slots=True) +class ToolboxToolTranslation: + """Server verdict for a single toolbox tool. + + ``classification`` is one of ``translated`` / ``partially-translated`` / + ``unsupported``. + """ + + tool_name: str + classification: str + process_id: str | None = None + parameter_bindings: list[ToolboxParameterBinding] = field(default_factory=list) + issues: list[ToolboxTranslationIssue] = field(default_factory=list) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ToolboxToolTranslation: + d = _snake_keys(data) + d["parameter_bindings"] = _model_list(ToolboxParameterBinding, d.get("parameter_bindings", [])) + d["issues"] = _model_list(ToolboxTranslationIssue, d.get("issues", [])) + return cls(**_extract_fields(cls, d)) + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass(frozen=True, slots=True) +class ToolboxTranslationSummary: + """Per-classification tool counts for a translation report.""" + + tool_count: int = 0 + translated_count: int = 0 + partially_translated_count: int = 0 + unsupported_count: int = 0 + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ToolboxTranslationSummary: + return cls(**_extract_fields(cls, _snake_keys(data))) + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass(frozen=True, slots=True) +class ToolboxTranslationReport: + """Server-authoritative validation report for a translated toolbox. + + Returned by + :meth:`~honua_admin.HonuaAdminClient.validate_toolbox_translation`. The + canonical process catalog — not the SDK's local view of it — decides every + ``classification`` here, which is what makes a report built from this + *server-attested* rather than a local assertion. + + ``artifact_kind`` / ``artifact_version`` are deliberately **not** defaulted + to the expected identity: they stay ``None`` when the response omitted them. + The genuine endpoint always stamps both, so their absence is evidence that + the payload is not a translation report — and stamping a plausible identity + here would erase exactly the evidence a caller needs to refuse the response. + :func:`honua_sdk.migration.attest_translation` requires both fields before it + will call a verdict attested, and this round-trips their absence to it. + """ + + toolbox_name: str + source_format: str + summary: ToolboxTranslationSummary = field(default_factory=ToolboxTranslationSummary) + tools: list[ToolboxToolTranslation] = field(default_factory=list) + artifact_kind: str | None = None + artifact_version: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> ToolboxTranslationReport: + d = _snake_keys(data) + summary = d.get("summary") + d["summary"] = ( + ToolboxTranslationSummary.from_dict(summary) + if isinstance(summary, dict) + else ToolboxTranslationSummary() + ) + d["tools"] = _model_list(ToolboxToolTranslation, d.get("tools", [])) + return cls(**_extract_fields(cls, d)) + + def to_dict(self) -> dict[str, Any]: + # An omitted identity stays omitted, so a consumer of this dict sees the + # response the server actually sent. + d = _dataclass_to_camel_dict(self) + identity: dict[str, Any] = { + key: d.pop(key) for key in ("artifactKind", "artifactVersion") if key in d + } + return {**identity, **d} + + # --------------------------------------------------------------------------- # Request models (mutable) # --------------------------------------------------------------------------- +@dataclass +class ToolboxParameterMapping: + """Proposed mapping of one source toolbox parameter onto a process parameter.""" + + source_name: str + target_parameter: str + source_data_type: str | None = None + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass +class ToolboxToolDescriptor: + """One toolbox tool as translated by the SDK scanner. + + ``target_process_id`` is the canonical Honua process the scanner proposes as + the native executor (for example ``geometry.buffer``). Leave it ``None`` when + no native mapping was found — the server then reports the tool unsupported + rather than stubbing it as executable. + """ + + tool_name: str + display_name: str | None = None + target_process_id: str | None = None + parameter_mappings: list[ToolboxParameterMapping] = field(default_factory=list) + unsupported_constructs: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return _dataclass_to_camel_dict(self) + + +@dataclass +class ToolboxTranslationManifest: + """SDK-emitted translation manifest for one ArcGIS toolbox. + + ``source_format`` is the source toolbox format: ``pyt``, ``atbx``, or + ``tbx``. Redact local paths and credentials out of ``source_label`` before + submitting — the server echoes it into operator-visible output. + """ + + toolbox_name: str + source_format: str + source_label: str | None = None + tools: list[ToolboxToolDescriptor] = field(default_factory=list) + artifact_kind: str = TOOLBOX_TRANSLATION_MANIFEST_KIND + artifact_version: str = TOOLBOX_TRANSLATION_ARTIFACT_VERSION + + def to_dict(self) -> dict[str, Any]: + d = _dataclass_to_camel_dict(self) + return { + "artifactKind": d.pop("artifactKind"), + "artifactVersion": d.pop("artifactVersion"), + **d, + } + + @dataclass class MigrationInventoryScanRequest: source_kind: str diff --git a/packages/honua-sdk/honua_sdk/migration/__init__.py b/packages/honua-sdk/honua_sdk/migration/__init__.py index c43c574..402dbe4 100644 --- a/packages/honua-sdk/honua_sdk/migration/__init__.py +++ b/packages/honua-sdk/honua_sdk/migration/__init__.py @@ -6,6 +6,15 @@ :data:`~honua_sdk.migration.arcpy.EXECUTABLE_PROCESS_IDS` (a server-runnable built-in process); anything else is emitted as ``"manual-review"``. +A toolbox verdict can additionally be **server-attested**: build a manifest with +:func:`~honua_sdk.migration.build_pyt_translation_manifest` / +:func:`~honua_sdk.migration.build_atbx_translation_manifest` and pass it through +:func:`~honua_sdk.migration.attest_translation`, which has the server's canonical +process catalog classify every tool. Where the server and the SDK disagree the +server wins and the disagreement is reported; where no server is reachable the +report degrades to an explicitly marked ``local-only`` verdict. A local verdict +is never presented as attested. See :mod:`honua_sdk.migration.attestation`. + The codemod deliberately has **no** notion of "custom code" execution or backend selection: it never emits a custom-code (operator-supplied-code) geoprocessing job, and it cannot request local/on-host execution. Per honua-server ADR-0063, @@ -23,6 +32,7 @@ JOB_STATUS_FAILED, JOB_STATUS_RUNNING, JOB_STATUS_SUCCESSFUL, + ArcPyArgumentBinding, ArcPyCall, ArcPyJobError, ArcPyJobTimeoutError, @@ -34,11 +44,39 @@ UnsupportedArcPyCallError, build_parity_evidence, build_parity_evidence_for_source, + resolve_argument_bindings, scan_arcpy_file, scan_arcpy_source, translate_arcpy_report, translate_arcpy_source, ) +from .attestation import ( + AGREEMENT_AGREED, + AGREEMENT_DISAGREED, + AGREEMENT_NOT_ATTESTED, + CLASSIFICATION_PARTIALLY_TRANSLATED, + CLASSIFICATION_TRANSLATED, + CLASSIFICATION_UNSUPPORTED, + CLASSIFICATIONS, + LOCAL_ONLY, + MAX_MANIFEST_TOOLS, + SERVER_ATTESTED, + SOURCE_FORMAT_ATBX, + SOURCE_FORMAT_PYT, + SOURCE_FORMAT_TBX, + SUPPORTED_REPORT_VERSIONS, + AttestedToolVerdict, + AttestedTranslationReport, + TranslationAttestationError, + TranslationManifest, + TranslationParameterMapping, + TranslationToolProposal, + TranslationValidator, + attest_translation, + build_atbx_translation_manifest, + build_pyt_translation_manifest, + source_format_for_path, +) from .modelbuilder import ( GpService, GpTask, @@ -56,6 +94,7 @@ parse_model_definition, ) from .pyt import ( + BINARY_TOOLBOX_EXPORT_GUIDANCE, PytParameter, PytTool, PytToolbox, @@ -67,12 +106,28 @@ ) __all__ = [ + "AGREEMENT_AGREED", + "AGREEMENT_DISAGREED", + "AGREEMENT_NOT_ATTESTED", + "BINARY_TOOLBOX_EXPORT_GUIDANCE", + "CLASSIFICATIONS", + "CLASSIFICATION_PARTIALLY_TRANSLATED", + "CLASSIFICATION_TRANSLATED", + "CLASSIFICATION_UNSUPPORTED", "EXECUTABLE_PROCESS_IDS", "JOB_STATUS_ACCEPTED", "JOB_STATUS_DISMISSED", "JOB_STATUS_FAILED", "JOB_STATUS_RUNNING", "JOB_STATUS_SUCCESSFUL", + "LOCAL_ONLY", + "MAX_MANIFEST_TOOLS", + "SERVER_ATTESTED", + "SOURCE_FORMAT_ATBX", + "SOURCE_FORMAT_PYT", + "SOURCE_FORMAT_TBX", + "SUPPORTED_REPORT_VERSIONS", + "ArcPyArgumentBinding", "ArcPyCall", "ArcPyJobError", "ArcPyJobTimeoutError", @@ -81,6 +136,8 @@ "ArcPyProcessRunner", "ArcPyProcessTranslation", "ArcPyScanReport", + "AttestedToolVerdict", + "AttestedTranslationReport", "GpService", "GpTask", "GpTaskParameter", @@ -90,15 +147,23 @@ "PytParameter", "PytTool", "PytToolbox", + "TranslationAttestationError", + "TranslationManifest", + "TranslationParameterMapping", + "TranslationToolProposal", + "TranslationValidator", "UnsupportedArcPyCallError", "UnsupportedModelFormatError", "UnsupportedToolboxError", + "attest_translation", "build_atbx_parity_evidence", + "build_atbx_translation_manifest", "build_gp_service_parity_evidence", "build_model_parity_evidence", "build_parity_evidence", "build_parity_evidence_for_source", "build_pyt_parity_evidence", + "build_pyt_translation_manifest", "parse_atbx_toolbox", "parse_binary_toolbox", "parse_gp_service_definition", @@ -106,8 +171,10 @@ "parse_model_definition", "parse_pyt_file", "parse_pyt_source", + "resolve_argument_bindings", "scan_arcpy_file", "scan_arcpy_source", + "source_format_for_path", "translate_arcpy_report", "translate_arcpy_source", ] diff --git a/packages/honua-sdk/honua_sdk/migration/_cli.py b/packages/honua-sdk/honua_sdk/migration/_cli.py index 70115a8..61e5fb1 100644 --- a/packages/honua-sdk/honua_sdk/migration/_cli.py +++ b/packages/honua-sdk/honua_sdk/migration/_cli.py @@ -5,6 +5,7 @@ honua-migrate scan path/to/script.py python -m honua_sdk.migration scan path/to/script.py honua-migrate translate path/to/script.py --evidence out.json + honua-migrate translate path/to/toolbox.pyt --server https://example.test honua-migrate run path/to/script.py --server https://example.test honua-migrate pyt path/to/toolbox.pyt honua-migrate atbx path/to/toolbox.atbx --evidence out.json @@ -13,14 +14,29 @@ The ``scan``, ``translate``, ``pyt``, ``atbx``, and ``gpservice`` commands work offline (no ArcGIS or network). The ``run`` command executes the translatable steps through ``HonuaClient.ogc_processes().execute(...)`` against ``--server``. + +**Server attestation.** ``translate``/``pyt``/``atbx`` classify a toolbox from +the SDK's own view of the Honua process catalog, which can drift from the server +that would run the job. Pass ``--server`` (plus admin credentials) to have the +server's canonical catalog classify every tool instead; the emitted report then +carries ``verdictSource: "server-attested"``. Without ``--server`` -- or when the +server cannot be reached -- the report still emits, marked +``verdictSource: "local-only"`` with the reason. A local verdict is never +presented as attested. See ``honua_sdk.migration.attestation``. + +Attestation needs the ``honua-admin`` package (the endpoint is an admin one and +reuses the admin credential path); everything else here runs with ``honua-sdk`` +alone. """ from __future__ import annotations import argparse import json +import os import sys -from collections.abc import Sequence +from collections.abc import Iterator, Sequence +from contextlib import contextmanager from pathlib import Path from typing import Any @@ -30,6 +46,17 @@ scan_arcpy_file, translate_arcpy_report, ) +from .attestation import ( + SOURCE_FORMAT_ATBX, + SOURCE_FORMAT_PYT, + AttestedTranslationReport, + TranslationManifest, + TranslationValidator, + attest_translation, + build_atbx_translation_manifest, + build_pyt_translation_manifest, + source_format_for_path, +) from .modelbuilder import ( UnsupportedModelFormatError, build_atbx_parity_evidence, @@ -48,6 +75,13 @@ # proprietary binary .tbx remains an explicit stub in the ``pyt`` command. _BINARY_TOOLBOX_SUFFIXES = {".tbx"} +#: Environment fallback for ``--api-key`` so a credential never has to appear in +#: shell history or a CI command line. +ADMIN_API_KEY_ENV = "HONUA_ADMIN_API_KEY" + +#: Exit code for ``--require-attested`` when the verdict stayed local-only. +EXIT_NOT_ATTESTED = 4 + def _emit(obj: object, *, out: Path | None) -> None: text = json.dumps(obj, indent=2, sort_keys=False) @@ -57,6 +91,129 @@ def _emit(obj: object, *, out: Path | None) -> None: out.write_text(text + "\n", encoding="utf-8") +# --------------------------------------------------------------------------- +# Server attestation +# --------------------------------------------------------------------------- + + +@contextmanager +def _server_validator(args: argparse.Namespace) -> Iterator[TranslationValidator | None]: + """Yield a validator bound to ``--server``, or ``None`` when offline. + + The endpoint lives in the admin import group, so this reuses the existing + admin credential path (``HonuaAdminClient(api_key=...)``) rather than + introducing a second auth mechanism. ``honua_admin`` is imported lazily and + is an optional dependency: every other command in this CLI runs with + ``honua-sdk`` alone. + """ + + server = getattr(args, "server", None) + if not server: + yield None + return + + # Lazy + optional: honua-admin is not a honua-sdk dependency, so every other + # command in this CLI keeps working without it installed. + from honua_admin import HonuaAdminClient + + api_key = getattr(args, "api_key", None) or os.environ.get(ADMIN_API_KEY_ENV) + with HonuaAdminClient(server, api_key=api_key, timeout=args.attest_timeout) as client: + + def validate(manifest: TranslationManifest) -> dict[str, Any]: + return client.validate_toolbox_translation(_admin_manifest(manifest)).to_dict() + + yield validate + + +def _admin_manifest(manifest: TranslationManifest) -> Any: + """Map the codemod's manifest onto the admin client's request model.""" + + from honua_admin import ( + ToolboxParameterMapping, + ToolboxToolDescriptor, + ToolboxTranslationManifest, + ) + + return ToolboxTranslationManifest( + toolbox_name=manifest.toolbox_name, + source_format=manifest.source_format, + source_label=manifest.source_label, + tools=[ + ToolboxToolDescriptor( + tool_name=tool.tool_name, + display_name=tool.display_name, + target_process_id=tool.target_process_id, + parameter_mappings=[ + ToolboxParameterMapping( + source_name=mapping.source_name, + target_parameter=mapping.target_parameter, + source_data_type=mapping.source_data_type, + ) + for mapping in tool.parameter_mappings + ], + unsupported_constructs=list(tool.unsupported_constructs), + ) + for tool in manifest.tools + ], + ) + + +def _attest(manifest: TranslationManifest, args: argparse.Namespace) -> AttestedTranslationReport: + """Attest a manifest, degrading to a marked local-only verdict on failure.""" + + server = getattr(args, "server", None) + try: + with _server_validator(args) as validator: + return attest_translation(manifest, validator=validator, server=server) + except Exception as exc: + # Reaching a server is optional by design, so even a failure to *build* + # the client (honua-admin missing, malformed base URL) leaves a usable + # local-only report rather than aborting the migration run. + return attest_translation(manifest, validator=_failing_validator(exc), server=server) + + +def _failing_validator(exc: BaseException) -> TranslationValidator: + def validate(manifest: TranslationManifest) -> dict[str, Any]: + raise exc + + return validate + + +def _report_attestation(report: AttestedTranslationReport, args: argparse.Namespace) -> int: + """Attach the attestation to the output, print a summary, and pick an exit code.""" + + if args.attestation is not None: + _emit(report.to_dict(), out=args.attestation) + + summary = report.to_dict()["summary"] + if report.attested: + print( + f"attestation: server-attested by {report.server} -- " + f"{summary['translatedCount']} translated, " + f"{summary['partiallyTranslatedCount']} partial, " + f"{summary['unsupportedCount']} unsupported, " + f"{summary['disagreementCount']} disagreeing with the local verdict", + file=sys.stderr, + ) + for disagreement in report.disagreements: + print( + f" disagreement: {disagreement.tool_name} " + f"local={disagreement.local_classification} " + f"server={disagreement.server_classification} (server wins)", + file=sys.stderr, + ) + return 0 + + print(f"attestation: local-only (NOT server-attested) -- {report.fallback_reason}", file=sys.stderr) + if args.require_attested: + print( + "--require-attested was set, so a local-only verdict is a failure.", + file=sys.stderr, + ) + return EXIT_NOT_ATTESTED + return 0 + + def _cmd_scan(args: argparse.Namespace) -> int: report = scan_arcpy_file(args.path) _emit(report.to_dict(), out=args.output) @@ -67,6 +224,22 @@ def _cmd_scan(args: argparse.Namespace) -> int: def _cmd_translate(args: argparse.Namespace) -> int: + # The validation endpoint's contract is toolbox-scoped, so `translate` routes + # a toolbox container to the toolbox lane and keeps the arcpy-script lane for + # a plain .py script. + if source_format_for_path(args.path) is not None: + return _translate_toolbox(args) + + if args.server: + print( + f"--server cannot attest {args.path}: the server validation endpoint takes a " + "toolbox manifest (sourceFormat pyt/atbx/tbx), and a bare arcpy .py script is a " + "script rather than a toolbox. Run it without --server for the local plan, or " + "point translate at a .pyt / .atbx toolbox to get a server-attested verdict.", + file=sys.stderr, + ) + return 2 + report = scan_arcpy_file(args.path) if report.syntax_error is not None: print(f"syntax error: {report.syntax_error}", file=sys.stderr) @@ -89,6 +262,49 @@ def _cmd_translate(args: argparse.Namespace) -> int: return 0 +def _translate_toolbox(args: argparse.Namespace) -> int: + """Translate a ``.pyt`` / ``.atbx`` toolbox and (optionally) attest it.""" + + source_format = source_format_for_path(args.path) + if Path(args.path).suffix.lower() in _BINARY_TOOLBOX_SUFFIXES: + try: + parse_binary_toolbox(args.path) + except UnsupportedToolboxError as exc: + print(str(exc), file=sys.stderr) + return 3 + return 3 # pragma: no cover -- parse_binary_toolbox always raises for .tbx + + if source_format == SOURCE_FORMAT_ATBX: + try: + atbx = parse_atbx_toolbox(args.path) + except UnsupportedModelFormatError as exc: + print(str(exc), file=sys.stderr) + return 3 + document = atbx.to_dict() + manifest = build_atbx_translation_manifest(atbx) + evidence = build_atbx_parity_evidence(atbx) + parse_error = atbx.parse_error + else: + toolbox = parse_pyt_file(args.path) + document = toolbox.to_dict() + manifest = build_pyt_translation_manifest(toolbox) + evidence = build_pyt_parity_evidence(toolbox) + parse_error = toolbox.syntax_error + + if args.evidence is not None: + _emit(evidence, out=args.evidence) + + attestation = _attest(manifest, args) + document["translationManifest"] = manifest.to_dict() + document["attestation"] = attestation.to_dict() + _emit(document, out=args.output) + + if parse_error is not None: + print(f"parse error: {parse_error}", file=sys.stderr) + return 2 + return _report_attestation(attestation, args) + + def _cmd_run(args: argparse.Namespace) -> int: # Imported lazily so scan/translate/pyt work without httpx/network deps wired. from honua_sdk import HonuaClient @@ -145,11 +361,16 @@ def _cmd_pyt(args: argparse.Namespace) -> int: toolbox = parse_pyt_file(args.path) if args.evidence is not None: _emit(build_pyt_parity_evidence(toolbox), out=args.evidence) - _emit(toolbox.to_dict(), out=args.output) + + document = toolbox.to_dict() + attestation = _attest(build_pyt_translation_manifest(toolbox, source_format=SOURCE_FORMAT_PYT), args) + document["attestation"] = attestation.to_dict() + _emit(document, out=args.output) + if toolbox.syntax_error is not None: print(f"syntax error: {toolbox.syntax_error}", file=sys.stderr) return 2 - return 0 + return _report_attestation(attestation, args) def _cmd_atbx(args: argparse.Namespace) -> int: @@ -160,11 +381,16 @@ def _cmd_atbx(args: argparse.Namespace) -> int: return 3 if args.evidence is not None: _emit(build_atbx_parity_evidence(toolbox), out=args.evidence) - _emit(toolbox.to_dict(), out=args.output) + + document = toolbox.to_dict() + attestation = _attest(build_atbx_translation_manifest(toolbox, source_format=SOURCE_FORMAT_ATBX), args) + document["attestation"] = attestation.to_dict() + _emit(document, out=args.output) + if toolbox.parse_error is not None: print(f"parse error: {toolbox.parse_error}", file=sys.stderr) return 2 - return 0 + return _report_attestation(attestation, args) def _cmd_gpservice(args: argparse.Namespace) -> int: @@ -180,6 +406,53 @@ def _cmd_gpservice(args: argparse.Namespace) -> int: return 0 +def _add_attestation_arguments(parser: argparse.ArgumentParser) -> None: + """Add the shared ``--server`` attestation options to a toolbox command. + + Without ``--server`` the command stays fully offline and emits a report + marked ``local-only``; a local verdict is never labelled attested. + """ + + group = parser.add_argument_group( + "server attestation", + "Have the server's canonical process catalog classify each tool, instead of " + "relying on the SDK's local view of it. Requires the honua-admin package.", + ) + group.add_argument( + "--server", + default=None, + help=( + "Honua server base URL to validate the translated toolbox against. " + "Omit to stay offline (the report is then marked local-only)." + ), + ) + group.add_argument( + "--api-key", + default=None, + help=f"Admin API key for --server (default: ${ADMIN_API_KEY_ENV}).", + ) + group.add_argument( + "--attest-timeout", + type=float, + default=30.0, + help="Request timeout in seconds for the validation call (default: 30).", + ) + group.add_argument( + "--attestation", + type=Path, + default=None, + help="Also write the attestation report JSON to this path.", + ) + group.add_argument( + "--require-attested", + action="store_true", + help=( + f"Exit {EXIT_NOT_ATTESTED} when the verdict could not be server-attested, " + "instead of accepting the local-only fallback." + ), + ) + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="honua-migrate", description=__doc__.splitlines()[0]) sub = parser.add_subparsers(dest="command", required=True) @@ -191,9 +464,16 @@ def build_parser() -> argparse.ArgumentParser: translate = sub.add_parser( "translate", - help="Emit OGC Processes payloads + a parity-evidence coverage report (offline).", + help=( + "Translate an arcpy .py script or a .pyt/.atbx toolbox; emits OGC Processes " + "payloads + parity evidence, and a server-attested verdict with --server." + ), + ) + translate.add_argument( + "path", + type=Path, + help="Path to an arcpy .py script, a .pyt Python toolbox, or a .atbx ModelBuilder toolbox.", ) - translate.add_argument("path", type=Path, help="Path to an arcpy .py script.") translate.add_argument( "--output", type=Path, @@ -206,6 +486,7 @@ def build_parser() -> argparse.ArgumentParser: default=None, help="Write the parity-evidence JSON report to this path.", ) + _add_attestation_arguments(translate) translate.set_defaults(func=_cmd_translate) run = sub.add_parser("run", help="Execute translatable steps via ArcPyProcessRunner against --server.") @@ -219,6 +500,7 @@ def build_parser() -> argparse.ArgumentParser: pyt.add_argument("path", type=Path, help="Path to a .pyt Python toolbox (or .tbx to see the binary-format stub).") pyt.add_argument("--output", type=Path, default=None, help="Write the toolbox JSON here (default: stdout).") pyt.add_argument("--evidence", type=Path, default=None, help="Write the aggregated parity-evidence JSON here.") + _add_attestation_arguments(pyt) pyt.set_defaults(func=_cmd_pyt) atbx = sub.add_parser( @@ -228,6 +510,7 @@ def build_parser() -> argparse.ArgumentParser: atbx.add_argument("path", type=Path, help="Path to a .atbx ModelBuilder toolbox.") atbx.add_argument("--output", type=Path, default=None, help="Write the toolbox JSON here (default: stdout).") atbx.add_argument("--evidence", type=Path, default=None, help="Write the aggregated parity-evidence JSON here.") + _add_attestation_arguments(atbx) atbx.set_defaults(func=_cmd_atbx) gpservice = sub.add_parser( diff --git a/packages/honua-sdk/honua_sdk/migration/arcpy.py b/packages/honua-sdk/honua_sdk/migration/arcpy.py index d324d1a..80cf8c4 100644 --- a/packages/honua-sdk/honua_sdk/migration/arcpy.py +++ b/packages/honua-sdk/honua_sdk/migration/arcpy.py @@ -362,6 +362,24 @@ def __init__(self, job_id: str | None, status: str | None, polls: int) -> None: ) +@dataclass(frozen=True) +class ArcPyArgumentBinding: + """One supplied ArcPy argument paired with the canonical parameter it feeds. + + ``declared`` is ``False`` for an argument the registered tool signature does + not know about -- a positional beyond the signature, or an unrecognized + keyword. Those still carry their value into the translated payload (nothing + is silently dropped) but the translator makes no claim that the target name + is a real canonical parameter, so a report must not present them as mapped. + """ + + source_name: str + target_parameter: str + kind: InputKind + value: JsonValue + declared: bool + + @dataclass(frozen=True) class _ArgSpec: arcpy_name: str @@ -1768,35 +1786,112 @@ def _lookup_spec(family: str, tool: str) -> _ToolSpec | None: return spec -def _translate_call(call: ArcPyCall, *, process_id_map: Mapping[str, str]) -> ArcPyProcessTranslation: +def resolve_argument_bindings(call: ArcPyCall) -> tuple[ArcPyArgumentBinding, ...]: + """Resolve one call's arguments to canonical process parameter names. + + This is the same resolution :func:`translate_arcpy_report` performs when it + builds an OGC Processes payload, exposed so a caller can see *which* source + argument produced *which* canonical parameter -- a pairing the flattened + payload dict no longer carries. :mod:`honua_sdk.migration.attestation` uses + it to build the per-tool parameter mappings the server validation endpoint + checks against the canonical process catalog. + + Returns an empty tuple when the call has no registered Honua mapping. + """ + spec = _lookup_spec(call.family, call.tool) if spec is None: - raise UnsupportedArcPyCallError(f"ArcPy call {call.qualified_name!r} is not supported by the translator.") + return () + return tuple(_resolve_bindings(call, spec)) - inputs: JsonObject = {} - outputs: JsonObject = {} - consumed_keywords: set[str] = set() + +def _resolve_bindings(call: ArcPyCall, spec: _ToolSpec) -> list[ArcPyArgumentBinding]: + """Pair every supplied call argument with the canonical parameter it feeds.""" + + bindings: list[ArcPyArgumentBinding] = [] for index, value in enumerate(call.args): if index >= len(spec.args): - inputs[f"arg_{index + 1}"] = value + # Positional beyond the registered signature: the translator keeps the + # value under a synthetic name, but no canonical parameter is known. + bindings.append( + ArcPyArgumentBinding( + source_name=f"arg_{index + 1}", + target_parameter=f"arg_{index + 1}", + kind="parameter", + value=value, + declared=False, + ) + ) continue - _assign_process_value(spec.args[index], value, inputs=inputs, outputs=outputs) + arg_spec = spec.args[index] + bindings.append( + ArcPyArgumentBinding( + source_name=arg_spec.arcpy_name, + target_parameter=arg_spec.process_name, + kind=arg_spec.kind, + value=value, + declared=True, + ) + ) spec_by_keyword = {_normalize_keyword(arg.arcpy_name): arg for arg in spec.args} for raw_name, value in call.kwargs.items(): normalized = _normalize_keyword(raw_name) process_name = spec.aliases.get(normalized) - arg_spec = spec_by_keyword.get(normalized) + keyword_spec = spec_by_keyword.get(normalized) if process_name is not None: - kind = _kind_for_process_name(process_name, spec) - _assign_process_value(_ArgSpec(raw_name, process_name, kind), value, inputs=inputs, outputs=outputs) - consumed_keywords.add(raw_name) - elif arg_spec is not None: - _assign_process_value(arg_spec, value, inputs=inputs, outputs=outputs) - consumed_keywords.add(raw_name) + bindings.append( + ArcPyArgumentBinding( + source_name=raw_name, + target_parameter=process_name, + kind=_kind_for_process_name(process_name, spec), + value=value, + declared=True, + ) + ) + elif keyword_spec is not None: + bindings.append( + ArcPyArgumentBinding( + source_name=raw_name, + target_parameter=keyword_spec.process_name, + kind=keyword_spec.kind, + value=value, + declared=True, + ) + ) else: - inputs[_camel_to_snake(raw_name)] = value + # Unrecognized keyword: passed through under a snake_cased name so the + # value is not silently dropped, but the translator does not claim it + # is a canonical parameter. + bindings.append( + ArcPyArgumentBinding( + source_name=raw_name, + target_parameter=_camel_to_snake(raw_name), + kind="parameter", + value=value, + declared=False, + ) + ) + + return bindings + + +def _translate_call(call: ArcPyCall, *, process_id_map: Mapping[str, str]) -> ArcPyProcessTranslation: + spec = _lookup_spec(call.family, call.tool) + if spec is None: + raise UnsupportedArcPyCallError(f"ArcPy call {call.qualified_name!r} is not supported by the translator.") + + inputs: JsonObject = {} + outputs: JsonObject = {} + consumed_keywords: set[str] = set() + + positional_count = len(call.args) + for index, binding in enumerate(_resolve_bindings(call, spec)): + target = outputs if binding.kind == "output" else inputs + target[binding.target_parameter] = binding.value + if index >= positional_count and binding.declared: + consumed_keywords.add(binding.source_name) metadata: JsonObject = { "source": "arcpy", @@ -1835,11 +1930,6 @@ def _translate_call(call: ArcPyCall, *, process_id_map: Mapping[str, str]) -> Ar ) -def _assign_process_value(arg_spec: _ArgSpec, value: JsonValue, *, inputs: JsonObject, outputs: JsonObject) -> None: - target = outputs if arg_spec.kind == "output" else inputs - target[arg_spec.process_name] = value - - def _kind_for_process_name(process_name: str, spec: _ToolSpec) -> InputKind: for arg in spec.args: if arg.process_name == process_name: diff --git a/packages/honua-sdk/honua_sdk/migration/attestation.py b/packages/honua-sdk/honua_sdk/migration/attestation.py new file mode 100644 index 0000000..9320786 --- /dev/null +++ b/packages/honua-sdk/honua_sdk/migration/attestation.py @@ -0,0 +1,846 @@ +"""Server-attested toolbox translation reports. + +The migration codemod can classify an ArcGIS toolbox tool on its own, from the +SDK's built-in view of which Honua processes exist and what they accept. That +local view can drift from the server that would actually run the job: the SDK +can call a tool translated when the server's submit validator would reject it, +or flag one unsupported when the server would accept it. A migration report is +used to decide whether a migration is viable, so a locally-derived verdict that +has silently drifted is worse than no verdict. + +This module closes that gap. It builds the manifest the server's +``POST /api/v1/admin/import/toolbox/translation/validate`` endpoint expects, +submits it through a caller-supplied validator, and merges the server's per-tool +classification over the local one. Two rules govern the result: + +* **The server wins.** Where the two verdicts disagree, the server's stands and + the disagreement is reported rather than quietly overwritten -- a disagreement + is the signal that the SDK's catalog view has drifted. +* **A local verdict is never dressed up as attested.** Offline operation still + works, but the report is stamped ``local-only`` with an explicit reason. Any + failure -- unreachable server, refused credentials, a malformed or incomplete + response -- degrades the *whole* report to ``local-only``. There is no partial + attestation. + +The validator is injected rather than constructed here so this module stays +pure, offline, and free of a dependency on the admin client. The +``honua-migrate`` CLI supplies one backed by +:meth:`honua_admin.HonuaAdminClient.validate_toolbox_translation`, which is the +existing admin credential path -- the endpoint lives in the admin import group. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterable, Iterator, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .arcpy import ArcPyCall, JsonObject, resolve_argument_bindings +from .modelbuilder import ModelBuilderToolbox +from .pyt import PytToolbox + +#: Verdict source for a report whose classifications came from the server's +#: canonical process catalog. +SERVER_ATTESTED = "server-attested" + +#: Verdict source for a report that carries only the SDK's local classification. +#: A ``local-only`` report is a useful offline artifact; it is not attestation. +LOCAL_ONLY = "local-only" + +#: Per-tool classifications, matching the server's vocabulary exactly. +CLASSIFICATION_TRANSLATED = "translated" +CLASSIFICATION_PARTIALLY_TRANSLATED = "partially-translated" +CLASSIFICATION_UNSUPPORTED = "unsupported" + +#: The complete classification vocabulary this report format understands. +#: +#: The summary counts tools by exactly these three values, so a fourth value -- +#: from a newer server, a rewriting proxy, or a malformed response -- would +#: appear as a tool's effective classification while no counter included it. +#: That is an internally inconsistent artifact, so an unrecognized +#: classification degrades the report to ``local-only`` instead of being +#: attested. Widening the vocabulary is a deliberate change here, not something +#: a response gets to do at runtime. +CLASSIFICATIONS: frozenset[str] = frozenset( + { + CLASSIFICATION_TRANSLATED, + CLASSIFICATION_PARTIALLY_TRANSLATED, + CLASSIFICATION_UNSUPPORTED, + } +) + +#: Agreement between the local and server verdict for one tool. +AGREEMENT_AGREED = "agreed" +AGREEMENT_DISAGREED = "disagreed" +AGREEMENT_NOT_ATTESTED = "not-attested" + +#: Artifact identity the server requires on an inbound manifest. A payload that +#: identifies as anything else is rejected rather than reinterpreted as v1. +MANIFEST_ARTIFACT_KIND = "honua.migration.toolbox-translation" +MANIFEST_ARTIFACT_VERSION = "1.0" + +#: Artifact identity the server stamps on the report it returns. +#: +#: This is *required* on an attested response, not optional. The genuine +#: endpoint always emits both fields, so their absence means the payload came +#: from something else -- a proxy, an error envelope, a different API -- and +#: cannot back an attestation claim. Client-side parsing must never fill them +#: in: a locally-manufactured identity would make a malformed response +#: indistinguishable from a real v1 report. +REPORT_ARTIFACT_KIND = "honua.migration.toolbox-translation-report" + +#: Report schema versions this client knows how to read. +SUPPORTED_REPORT_VERSIONS: frozenset[str] = frozenset({"1.0"}) + +#: Schema id for the merged attestation artifact this module emits. +ATTESTATION_SCHEMA = "honua.migration.toolbox-translation-attestation/v1" + +#: Source toolbox formats the server accepts. +SOURCE_FORMAT_PYT = "pyt" +SOURCE_FORMAT_ATBX = "atbx" +SOURCE_FORMAT_TBX = "tbx" + +#: Tools the server accepts in one manifest. Larger toolboxes are submitted as +#: several manifests and the reports merged, rather than truncated. +MAX_MANIFEST_TOOLS = 200 + +#: Signature of a validator: a manifest in, the server's report payload out. +#: Raising is the documented way to signal that attestation is unavailable -- +#: an unreachable server, refused credentials, or a rejected manifest all +#: surface that way and all degrade the report to ``local-only``. +TranslationValidator = Callable[["TranslationManifest"], JsonObject] + +_SUFFIX_SOURCE_FORMATS = { + ".pyt": SOURCE_FORMAT_PYT, + ".atbx": SOURCE_FORMAT_ATBX, + ".tbx": SOURCE_FORMAT_TBX, +} + + +class TranslationAttestationError(RuntimeError): + """Raised when a server response cannot be trusted as an attestation. + + Carried as a ``fallbackReason`` rather than propagated: a report that cannot + be attested degrades to ``local-only`` instead of failing the migration run. + """ + + +@dataclass(frozen=True) +class TranslationParameterMapping: + """One proposed source-parameter to canonical-parameter mapping.""" + + source_name: str + target_parameter: str + source_data_type: str | None = None + + def to_dict(self) -> JsonObject: + result: JsonObject = { + "sourceName": self.source_name, + "targetParameter": self.target_parameter, + } + if self.source_data_type is not None: + result["sourceDataType"] = self.source_data_type + return result + + +@dataclass(frozen=True) +class TranslationToolProposal: + """One toolbox tool the scanner proposes to map onto a native process. + + ``local_classification`` is what the SDK concluded on its own. It is kept + beside the server's answer rather than replaced by it, so a reader can see + both and a drifting SDK catalog view is visible instead of invisible. + """ + + tool_name: str + local_classification: str + display_name: str | None = None + target_process_id: str | None = None + parameter_mappings: tuple[TranslationParameterMapping, ...] = () + unsupported_constructs: tuple[str, ...] = () + + def to_descriptor(self) -> JsonObject: + """Render the wire descriptor the validation endpoint expects.""" + + return { + "toolName": self.tool_name, + "displayName": self.display_name, + "targetProcessId": self.target_process_id, + "parameterMappings": [mapping.to_dict() for mapping in self.parameter_mappings], + "unsupportedConstructs": list(self.unsupported_constructs), + } + + +@dataclass(frozen=True) +class TranslationManifest: + """A translated toolbox, ready to submit for server validation.""" + + toolbox_name: str + source_format: str + tools: tuple[TranslationToolProposal, ...] + source_label: str | None = None + + def to_dict(self) -> JsonObject: + """Render the manifest payload for the validation endpoint.""" + + result: JsonObject = { + "artifactKind": MANIFEST_ARTIFACT_KIND, + "artifactVersion": MANIFEST_ARTIFACT_VERSION, + "toolboxName": self.toolbox_name, + "sourceFormat": self.source_format, + } + if self.source_label is not None: + result["sourceLabel"] = self.source_label + result["tools"] = [tool.to_descriptor() for tool in self.tools] + return result + + def batches(self, size: int = MAX_MANIFEST_TOOLS) -> tuple["TranslationManifest", ...]: + """Split into manifests the server will accept, preserving tool order. + + The endpoint caps a manifest at :data:`MAX_MANIFEST_TOOLS` tools. A + larger toolbox is submitted as several manifests whose reports are then + merged, so a big toolbox degrades to more requests rather than to a + rejected manifest and a silent loss of attestation. + """ + + if size < 1: + raise ValueError("size must be at least 1.") + if len(self.tools) <= size: + return (self,) + return tuple( + TranslationManifest( + toolbox_name=self.toolbox_name, + source_format=self.source_format, + tools=tuple(self.tools[start : start + size]), + source_label=self.source_label, + ) + for start in range(0, len(self.tools), size) + ) + + +@dataclass(frozen=True) +class AttestedToolVerdict: + """The merged verdict for one tool. + + ``classification`` is the *effective* verdict: the server's when the report + is attested, the SDK's local one otherwise. ``local_classification`` and + ``server_classification`` are both retained so the merge is auditable. + """ + + tool_name: str + classification: str + local_classification: str + agreement: str + server_classification: str | None = None + process_id: str | None = None + parameter_bindings: tuple[JsonObject, ...] = () + issues: tuple[JsonObject, ...] = () + + @property + def disagreed(self) -> bool: + """Whether the server contradicted the SDK's local classification.""" + + return self.agreement == AGREEMENT_DISAGREED + + def to_dict(self) -> JsonObject: + return { + "toolName": self.tool_name, + "classification": self.classification, + "localClassification": self.local_classification, + "serverClassification": self.server_classification, + "agreement": self.agreement, + "processId": self.process_id, + "parameterBindings": [dict(binding) for binding in self.parameter_bindings], + "issues": [dict(issue) for issue in self.issues], + } + + +@dataclass(frozen=True) +class AttestedTranslationReport: + """A translation report plus an explicit statement of who attested it.""" + + manifest: TranslationManifest + verdict_source: str + tools: tuple[AttestedToolVerdict, ...] + fallback_reason: str | None = None + server: str | None = None + + @property + def attested(self) -> bool: + """Whether the classifications came from the server's process catalog.""" + + return self.verdict_source == SERVER_ATTESTED + + @property + def disagreements(self) -> tuple[AttestedToolVerdict, ...]: + """Tools where the server contradicted the SDK's local verdict.""" + + return tuple(verdict for verdict in self.tools if verdict.disagreed) + + def count(self, classification: str) -> int: + """Number of tools carrying *classification* as their effective verdict.""" + + return sum(1 for verdict in self.tools if verdict.classification == classification) + + def to_dict(self) -> JsonObject: + """Render the attestation artifact.""" + + result: JsonObject = { + "schema": ATTESTATION_SCHEMA, + "verdictSource": self.verdict_source, + "attested": self.attested, + "server": self.server, + "fallbackReason": self.fallback_reason, + "toolboxName": self.manifest.toolbox_name, + "sourceFormat": self.manifest.source_format, + "sourceLabel": self.manifest.source_label, + "summary": { + "toolCount": len(self.tools), + "translatedCount": self.count(CLASSIFICATION_TRANSLATED), + "partiallyTranslatedCount": self.count(CLASSIFICATION_PARTIALLY_TRANSLATED), + "unsupportedCount": self.count(CLASSIFICATION_UNSUPPORTED), + "disagreementCount": len(self.disagreements), + }, + "tools": [verdict.to_dict() for verdict in self.tools], + "disagreements": [ + { + "toolName": verdict.tool_name, + "local": verdict.local_classification, + "server": verdict.server_classification, + } + for verdict in self.disagreements + ], + } + return result + + +# --------------------------------------------------------------------------- +# Attestation +# --------------------------------------------------------------------------- + + +def attest_translation( + manifest: TranslationManifest, + *, + validator: TranslationValidator | None = None, + server: str | None = None, +) -> AttestedTranslationReport: + """Merge a server validation over a manifest's local classifications. + + Args: + manifest: The translated toolbox to have validated. + validator: Callable that submits a manifest and returns the server's + report payload. ``None`` -- the default -- produces a + ``local-only`` report without any network access. A validator that + raises also produces a ``local-only`` report, carrying the failure + as :attr:`AttestedTranslationReport.fallback_reason`. + server: Operator-visible label for the validating server, recorded on + the report. Pass a base URL, never a credential. + + Returns: + The merged :class:`AttestedTranslationReport`. It is + ``server-attested`` only when every submitted tool came back classified; + any failure at all degrades the whole report to ``local-only``. + """ + + if validator is None: + return _local_only( + manifest, + "No server validator was configured, so the verdict is the SDK's local " + "view of the process catalog and has not been attested by a server.", + server=server, + ) + + try: + classifications = _collect_server_classifications(manifest, validator) + # Deliberately broad: any failure at all -- transport, HTTP status, a + # malformed report -- means the verdict is not attested, and none of them + # is allowed to abort an otherwise-usable offline migration run. + except Exception as exc: + return _local_only(manifest, _describe_failure(exc), server=server) + + return AttestedTranslationReport( + manifest=manifest, + verdict_source=SERVER_ATTESTED, + tools=tuple( + _merge_verdict(tool, classifications[tool.tool_name]) for tool in manifest.tools + ), + server=server, + ) + + +def _collect_server_classifications( + manifest: TranslationManifest, + validator: TranslationValidator, +) -> dict[str, JsonObject]: + """Submit every batch and return the per-tool server verdicts. + + Raises: + TranslationAttestationError: The response is not a usable report, or it + does not classify every submitted tool. Both cases mean the report + cannot be presented as attested. + """ + + classifications: dict[str, JsonObject] = {} + for batch in manifest.batches(): + classifications.update(_parse_report(validator(batch), batch)) + + missing = [tool.tool_name for tool in manifest.tools if tool.tool_name not in classifications] + if missing: + raise TranslationAttestationError( + "The server report did not classify every submitted tool " + f"(missing: {', '.join(missing)})." + ) + return classifications + + +def _parse_report(payload: Any, batch: TranslationManifest) -> dict[str, JsonObject]: + """Extract per-tool verdicts from one server report payload.""" + + if not isinstance(payload, dict): + raise TranslationAttestationError( + f"The server returned a {type(payload).__name__} where a translation report object was expected." + ) + + # Artifact identity is REQUIRED, not merely consistent-if-present. The + # genuine endpoint always stamps both fields, so a payload without them is + # not a translation report and must not back an attestation claim. + artifact_kind = payload.get("artifactKind") + if artifact_kind != REPORT_ARTIFACT_KIND: + raise TranslationAttestationError( + f"The server returned artifactKind {artifact_kind!r}, not {REPORT_ARTIFACT_KIND!r}; " + "the response cannot be trusted as a translation report." + ) + + artifact_version = payload.get("artifactVersion") + if artifact_version not in SUPPORTED_REPORT_VERSIONS: + raise TranslationAttestationError( + f"The server returned artifactVersion {artifact_version!r}; this client reads " + f"{', '.join(sorted(SUPPORTED_REPORT_VERSIONS))}." + ) + + # A report is only evidence about the artifact it was asked about. Matching + # tool names are not enough: two toolboxes can share tool names, and a stale + # or misrouted response would then attest the wrong artifact entirely. The + # server echoes both fields back for exactly this purpose, so bind to them. + reported_toolbox = payload.get("toolboxName") + if not isinstance(reported_toolbox, str) or reported_toolbox.strip() != batch.toolbox_name.strip(): + raise TranslationAttestationError( + f"The server report is for toolbox {reported_toolbox!r}, not the submitted " + f"{batch.toolbox_name!r}; it cannot attest this toolbox." + ) + + reported_format = payload.get("sourceFormat") + if ( + not isinstance(reported_format, str) + # The server normalises sourceFormat to lower case on the way out. + or reported_format.strip().casefold() != batch.source_format.strip().casefold() + ): + raise TranslationAttestationError( + f"The server report is for sourceFormat {reported_format!r}, not the submitted " + f"{batch.source_format!r}; it cannot attest this toolbox." + ) + + tools = payload.get("tools") + if not isinstance(tools, list): + raise TranslationAttestationError("The server report carries no 'tools' array.") + + submitted = {tool.tool_name for tool in batch.tools} + verdicts: dict[str, JsonObject] = {} + for entry in tools: + if not isinstance(entry, dict): + raise TranslationAttestationError("The server report contains a non-object tool entry.") + tool_name = entry.get("toolName") + classification = entry.get("classification") + if not isinstance(tool_name, str) or not isinstance(classification, str): + raise TranslationAttestationError( + "The server report contains a tool entry without a toolName/classification pair." + ) + if classification not in CLASSIFICATIONS: + # Accepting it would put a value in `classification` that no summary + # counter tallies, producing an attested report that does not add up. + raise TranslationAttestationError( + f"The server classified {tool_name!r} as {classification!r}, which is outside this " + f"report format's vocabulary ({', '.join(sorted(CLASSIFICATIONS))})." + ) + if tool_name not in submitted: + raise TranslationAttestationError( + f"The server report classifies {tool_name!r}, which was not submitted." + ) + verdicts[tool_name] = entry + return verdicts + + +def _merge_verdict(tool: TranslationToolProposal, entry: JsonObject) -> AttestedToolVerdict: + """Apply one server verdict over a tool's local classification.""" + + server_classification = str(entry["classification"]) + agreement = ( + AGREEMENT_AGREED + if server_classification == tool.local_classification + else AGREEMENT_DISAGREED + ) + process_id = entry.get("processId") + return AttestedToolVerdict( + tool_name=tool.tool_name, + # The server owns the canonical catalog, so its verdict is the effective + # one even when it contradicts the SDK. The local value stays on the + # record beside it rather than being overwritten. + classification=server_classification, + local_classification=tool.local_classification, + server_classification=server_classification, + agreement=agreement, + process_id=process_id if isinstance(process_id, str) else None, + parameter_bindings=_object_list(entry.get("parameterBindings")), + issues=_object_list(entry.get("issues")), + ) + + +def _local_only( + manifest: TranslationManifest, + reason: str, + *, + server: str | None, +) -> AttestedTranslationReport: + """Build the unattested report, with the reason stated rather than implied.""" + + return AttestedTranslationReport( + manifest=manifest, + verdict_source=LOCAL_ONLY, + tools=tuple( + AttestedToolVerdict( + tool_name=tool.tool_name, + classification=tool.local_classification, + local_classification=tool.local_classification, + agreement=AGREEMENT_NOT_ATTESTED, + process_id=tool.target_process_id, + ) + for tool in manifest.tools + ), + fallback_reason=reason, + server=server, + ) + + +def _describe_failure(exc: BaseException) -> str: + detail = str(exc).strip() + label = type(exc).__name__ + return f"{label}: {detail}" if detail else label + + +def _object_list(value: Any) -> tuple[JsonObject, ...]: + if not isinstance(value, list): + return () + return tuple(dict(item) for item in value if isinstance(item, dict)) + + +# --------------------------------------------------------------------------- +# Manifest construction +# --------------------------------------------------------------------------- + + +def source_format_for_path(path: str | Path) -> str | None: + """Return the manifest ``sourceFormat`` for a toolbox path, or ``None``. + + ``None`` means the file is not one of the toolbox containers the validation + endpoint accepts -- a bare ``.py`` arcpy script, for example, which is a + script rather than a toolbox. + """ + + return _SUFFIX_SOURCE_FORMATS.get(Path(path).suffix.lower()) + + +def build_pyt_translation_manifest( + toolbox: PytToolbox, + *, + source_format: str = SOURCE_FORMAT_PYT, +) -> TranslationManifest: + """Build the validation manifest for a parsed ``.pyt`` Python toolbox. + + A ``.pyt`` declares its tools as ``self.tools = [A, B]``. A name listed there + whose class is *not* defined in the same file -- imported from another module, + or simply absent -- is recorded in + :attr:`~honua_sdk.migration.PytToolbox.declared_tool_names` but never + materialises into :attr:`~honua_sdk.migration.PytToolbox.tools`, because the + reader has no ``execute`` body to classify. + + Those names still go into the manifest, with no proposed target. Submitting + only the materialised tools would let the server return a clean report for a + toolbox whose imported tools were never classified, and the attestation would + then cover a strict subset while claiming to cover all of it -- the same hole + ``.atbx`` script tools opened (honua-sdk-python#188 review). + """ + + materialised = {tool.class_name for tool in toolbox.tools} + return TranslationManifest( + toolbox_name=_toolbox_name(toolbox.label or toolbox.alias, toolbox.filename), + source_format=source_format, + source_label=_source_label(toolbox.filename), + tools=tuple( + _flatten( + [ + *( + # Every discovered call, not just the translatable ones: a tool the + # SDK cannot map still belongs in the report as explicitly + # unsupported rather than missing from the tool count. + _proposals_for_tool(tool.class_name, tool.label, tool.report.calls) + for tool in toolbox.tools + ), + _unresolved_tool_proposals( + ( + name + for name in toolbox.declared_tool_names + if name.strip() not in materialised + ), + reason=( + "is declared in the toolbox's self.tools but its class is not defined " + "in this .pyt (typically imported from another module), so the reader " + "found no execute() body to classify and no native process mapping has " + "been established. Scan the module that defines it with the arcpy .py " + "scanner." + ), + ), + ] + ) + ), + ) + + +def build_atbx_translation_manifest( + toolbox: ModelBuilderToolbox, + *, + source_format: str = SOURCE_FORMAT_ATBX, +) -> TranslationManifest: + """Build the validation manifest for a parsed ``.atbx`` ModelBuilder toolbox. + + A ``.atbx`` yields three kinds of tool, and all three go into the manifest: + + * ModelBuilder **models** carry their geoprocessing steps inline and are + translated normally. + * **Script tools** only reference an external Python body the reader + deliberately does not follow, so they arrive as names in + :attr:`~honua_sdk.migration.ModelBuilderToolbox.script_tool_names`. + * **Unresolved tools** are declared models whose definition yielded no + recognizable step, surfaced in + :attr:`~honua_sdk.migration.ModelBuilderToolbox.unresolved_tool_names`. + + Submitting only the models would let the server return a clean report for a + toolbox whose other tools were never classified, and the attestation would + then cover a strict subset while claiming to cover all of it. The two + name-only kinds are therefore submitted with no proposed target -- the + honest statement, since the translator has established nothing about them -- + and the server reports them ``unsupported``. + """ + + return TranslationManifest( + toolbox_name=_toolbox_name(None, toolbox.filename), + source_format=source_format, + source_label=_source_label(toolbox.filename), + tools=tuple( + _flatten( + [ + *( + _proposals_for_tool(model.name, model.label, [step.call for step in model.steps]) + for model in toolbox.models + ), + # Further groups, so a script tool sharing a model's name is + # disambiguated by _flatten rather than rejected by the server. + _unresolved_tool_proposals( + toolbox.script_tool_names, + reason=( + "is a script tool: its geoprocessing logic lives in an external Python " + "script the .atbx reader does not follow, so no native process mapping " + "has been established. Scan that script with the arcpy .py scanner." + ), + ), + _unresolved_tool_proposals( + toolbox.unresolved_tool_names, + reason=( + "is declared by the toolbox but its definition yielded no recognizable " + "geoprocessing step, so there is nothing to map onto a native process. " + "Review the model in ArcGIS Pro to confirm what it does." + ), + ), + ] + ) + ), + ) + + +def _unresolved_tool_proposals( + names: Iterable[str], + *, + reason: str, +) -> tuple[TranslationToolProposal, ...]: + """Propose tools the reader discovered by name but could not read a body for. + + Both toolbox formats have this shape: a ``.atbx`` script tool points at an + external Python script the reader does not follow, and a ``.pyt`` may list a + tool class that is imported rather than defined in the file. Either way the + translator has established nothing, so no target is proposed and no coverage + is claimed -- guessing one would reintroduce the same false confidence + through a different door. The tool still has to appear in the manifest so + the report's tool count matches the toolbox and the server marks it + ``unsupported``. + """ + + return tuple( + TranslationToolProposal( + tool_name=name.strip(), + local_classification=CLASSIFICATION_UNSUPPORTED, + unsupported_constructs=(f"'{name.strip()}' {reason}",), + ) + for name in names + if name and name.strip() + ) + + +def _flatten(groups: Iterable[Sequence[TranslationToolProposal]]) -> Iterator[TranslationToolProposal]: + """Concatenate per-tool proposals, keeping every manifest tool name unique. + + The endpoint rejects a manifest containing a duplicate ``toolName``, and the + report is keyed by that name, so a repeated name would cost attestation for + the whole toolbox. Disambiguate instead of failing. + """ + + seen: dict[str, int] = {} + for group in groups: + for proposal in group: + count = seen.get(proposal.tool_name, 0) + seen[proposal.tool_name] = count + 1 + if count == 0: + yield proposal + else: + yield _renamed(proposal, f"{proposal.tool_name}~{count + 1}", seen) + + +def _renamed( + proposal: TranslationToolProposal, + candidate: str, + seen: dict[str, int], +) -> TranslationToolProposal: + name = candidate + suffix = 1 + while name in seen: + suffix += 1 + name = f"{candidate}.{suffix}" + seen[name] = 1 + return TranslationToolProposal( + tool_name=name, + local_classification=proposal.local_classification, + display_name=proposal.display_name, + target_process_id=proposal.target_process_id, + parameter_mappings=proposal.parameter_mappings, + unsupported_constructs=proposal.unsupported_constructs, + ) + + +def _proposals_for_tool( + tool_name: str, + display_name: str | None, + calls: Sequence[ArcPyCall], +) -> tuple[TranslationToolProposal, ...]: + """Turn one toolbox tool into the proposals the server validates. + + The server certifies *one* native process per manifest tool, so a tool whose + body runs several geoprocessing calls is submitted as one proposal per call, + suffixed to keep manifest tool names unique. A tool with no recognized call + is still submitted -- with no target -- so it appears in the report as + explicitly unsupported rather than being dropped from the count. + """ + + if not calls: + return ( + TranslationToolProposal( + tool_name=tool_name.strip(), + local_classification=CLASSIFICATION_UNSUPPORTED, + display_name=display_name, + unsupported_constructs=( + "The tool body contains no geoprocessing call the translator recognizes.", + ), + ), + ) + + multiple = len(calls) > 1 + return tuple( + _proposal_for_call( + f"{tool_name.strip()}#{index + 1}" if multiple else tool_name.strip(), + display_name, + call, + ) + for index, call in enumerate(calls) + ) + + +def _proposal_for_call( + tool_name: str, + display_name: str | None, + call: ArcPyCall, +) -> TranslationToolProposal: + """Translate one geoprocessing call into a server-validatable proposal.""" + + constructs: list[str] = [] + mappings: list[TranslationParameterMapping] = [] + + for binding in resolve_argument_bindings(call): + if binding.kind == "output": + # Outputs name a destination dataset, not a canonical process input; + # submitting them would only produce unknown-parameter noise. + continue + mappings.append( + TranslationParameterMapping( + source_name=binding.source_name, + target_parameter=binding.target_parameter, + ) + ) + if not binding.declared: + constructs.append( + f"Argument {binding.source_name!r} is not part of the registered " + f"{call.qualified_name} signature and is passed through unmapped." + ) + + reason = call.manual_review_reason + if reason is not None: + constructs.append(reason) + elif call.process_id is None: + constructs.append( + f"No Honua process mapping is registered for {call.qualified_name}." + ) + + return TranslationToolProposal( + tool_name=tool_name, + local_classification=_local_classification(call, constructs), + display_name=display_name, + target_process_id=call.job_process_id, + parameter_mappings=tuple(mappings), + unsupported_constructs=tuple(constructs), + ) + + +def _local_classification(call: ArcPyCall, constructs: Sequence[str]) -> str: + """The SDK's own verdict for one call, in the server's vocabulary.""" + + if not call.translatable: + # Both "no mapping at all" and "mapped but the server cannot job-execute + # it" mean the tool is not runnable, which is the server's `unsupported`. + return CLASSIFICATION_UNSUPPORTED + if constructs: + return CLASSIFICATION_PARTIALLY_TRANSLATED + return CLASSIFICATION_TRANSLATED + + +def _toolbox_name(declared: str | None, filename: str | None) -> str: + if declared and declared.strip(): + return declared.strip() + if filename: + stem = Path(filename).stem + if stem: + return stem + return "toolbox" + + +def _source_label(filename: str | None) -> str | None: + """Basename only: the directory path is the operator's, not the server's.""" + + return Path(filename).name if filename else None diff --git a/packages/honua-sdk/honua_sdk/migration/modelbuilder.py b/packages/honua-sdk/honua_sdk/migration/modelbuilder.py index 7a3c297..8b9edb5 100644 --- a/packages/honua-sdk/honua_sdk/migration/modelbuilder.py +++ b/packages/honua-sdk/honua_sdk/migration/modelbuilder.py @@ -53,6 +53,7 @@ _translate_call, build_parity_evidence, ) +from .pyt import BINARY_TOOLBOX_EXPORT_GUIDANCE _BINARY_TOOLBOX_SUFFIXES = frozenset({".tbx"}) _ATBX_SUFFIX = ".atbx" @@ -234,12 +235,20 @@ class ModelBuilderToolbox: models: tuple[ModelBuilderModel, ...] script_tool_names: tuple[str, ...] = field(default_factory=tuple) parse_error: str | None = None + #: Tools the archive declares that did not become entries in + #: :attr:`models` -- a model whose definition yielded no recognizable + #: geoprocessing step. They are excluded from ``models`` (there is nothing + #: to translate) but must stay *discoverable*, because a consumer that only + #: sees ``models`` would under-count the toolbox and could certify coverage + #: it does not have. See ``honua_sdk.migration.attestation``. + unresolved_tool_names: tuple[str, ...] = field(default_factory=tuple) def to_dict(self) -> JsonObject: result: JsonObject = { "schema": "honua.migration.arcpy.modelbuilder-toolbox/v1", "filename": self.filename, "scriptToolNames": list(self.script_tool_names), + "unresolvedToolNames": list(self.unresolved_tool_names), "models": [model.to_dict() for model in self.models], } if self.parse_error is not None: @@ -388,8 +397,8 @@ def parse_atbx_toolbox(path: str | Path) -> ModelBuilderToolbox: if file_path.suffix.lower() in _BINARY_TOOLBOX_SUFFIXES: raise UnsupportedModelFormatError( f"Binary toolbox parsing for {file_path.suffix!r} is not supported " - "(proprietary format -- not clean-room parseable). Export to .atbx " - "and re-run." + "(proprietary format -- not clean-room parseable). " + f"{BINARY_TOOLBOX_EXPORT_GUIDANCE}" ) try: @@ -403,6 +412,7 @@ def parse_atbx_toolbox(path: str | Path) -> ModelBuilderToolbox: models: list[ModelBuilderModel] = [] script_tool_names: list[str] = [] + unresolved_tool_names: list[str] = [] with archive: for info in archive.infolist(): if info.is_dir(): @@ -438,11 +448,18 @@ def parse_atbx_toolbox(path: str | Path) -> ModelBuilderToolbox: ) if model.steps: models.append(model) + elif model.name and model.name not in unresolved_tool_names: + # A model with no recognizable step has nothing to translate, so + # it stays out of `models` -- but it is still a tool the toolbox + # declares, and dropping the name entirely would let a caller + # under-count the toolbox and certify coverage it does not have. + unresolved_tool_names.append(model.name) return ModelBuilderToolbox( filename=str(file_path), models=tuple(models), script_tool_names=tuple(sorted(script_tool_names)), + unresolved_tool_names=tuple(sorted(unresolved_tool_names)), ) diff --git a/packages/honua-sdk/honua_sdk/migration/pyt.py b/packages/honua-sdk/honua_sdk/migration/pyt.py index 1799b54..87cc6b4 100644 --- a/packages/honua-sdk/honua_sdk/migration/pyt.py +++ b/packages/honua-sdk/honua_sdk/migration/pyt.py @@ -32,9 +32,36 @@ class whose ``self.tools`` attribute lists tool classes, where each tool class translate_arcpy_source, ) +#: Concrete migration instruction attached to every binary-``.tbx`` refusal. +#: +#: The refusal is a standing policy decision, not an unfinished stub: Honua does +#: not reverse-engineer proprietary Esri containers (the same rule that governs +#: ``.loc``/``.lox`` locator files), and the migration path is always +#: export-to-open-format. Because the answer will never be "we added a parser", +#: the error has to carry the steps that actually unblock the migration instead +#: of reading as a dead end. +BINARY_TOOLBOX_EXPORT_GUIDANCE = ( + "Export the toolbox to an open format first, then re-run against that file:\n" + " 1. Open the .tbx in ArcGIS Pro (Catalog pane > Toolboxes).\n" + " 2. Right-click the toolbox > Save As > New ArcGIS Toolbox (.atbx). " + "ModelBuilder models and their tool/parameter metadata come across, and " + ".atbx is a published zip-of-JSON container this SDK reads with " + "honua_sdk.migration.parse_atbx_toolbox.\n" + " 3. For a script tool whose logic must migrate too, open its Properties > " + "Execution, copy the Python body into a .pyt Python toolbox (or export the " + "script), and run honua_sdk.migration.parse_pyt_file against that .pyt.\n" + "Both .atbx and .pyt are open formats and are fully supported here; the " + "binary .tbx container is deliberately never parsed." +) + class UnsupportedToolboxError(NotImplementedError): - """Raised when a binary toolbox format is not parseable as source.""" + """Raised when a binary toolbox format is not parseable as source. + + For a proprietary binary ``.tbx`` this is a policy refusal, not a missing + feature -- the message carries :data:`BINARY_TOOLBOX_EXPORT_GUIDANCE`, the + concrete ArcGIS Pro export steps that produce a readable ``.atbx``/``.pyt``. + """ @dataclass(frozen=True) @@ -196,9 +223,8 @@ def parse_binary_toolbox(path: str | Path) -> PytToolbox: ) raise UnsupportedToolboxError( f"Binary toolbox parsing for {suffix!r} is not supported " - "(proprietary binary format -- not clean-room parseable). Export to a " - ".atbx ModelBuilder toolbox (parse_atbx_toolbox) or a .pyt Python " - "toolbox (parse_pyt_file)." + "(proprietary binary format -- not clean-room parseable). " + f"{BINARY_TOOLBOX_EXPORT_GUIDANCE}" ) diff --git a/tests/admin/test_toolbox_translation.py b/tests/admin/test_toolbox_translation.py new file mode 100644 index 0000000..113b4d1 --- /dev/null +++ b/tests/admin/test_toolbox_translation.py @@ -0,0 +1,251 @@ +"""Tests for the admin toolbox-translation validation endpoint client. + +Wire contract: ``POST /api/v1/admin/import/toolbox/translation/validate`` +(honua-server#2145 / #3040). The server owns the round-trip proof against the +canonical process catalog; this client only has to speak the manifest/report +shapes exactly and go through the existing admin credential path. +""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from honua_admin import ( + TOOLBOX_TRANSLATION_MANIFEST_KIND, + TOOLBOX_TRANSLATION_REPORT_KIND, + AsyncHonuaAdminClient, + HonuaAdminClient, + ToolboxParameterMapping, + ToolboxToolDescriptor, + ToolboxTranslationManifest, +) + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +_REPORT = { + "artifactKind": TOOLBOX_TRANSLATION_REPORT_KIND, + "artifactVersion": "1.0", + "toolboxName": "VectorAnalysisToolbox", + "sourceFormat": "pyt", + "summary": { + "toolCount": 2, + "translatedCount": 1, + "partiallyTranslatedCount": 0, + "unsupportedCount": 1, + }, + "tools": [ + { + "toolName": "BufferGeometry", + "classification": "translated", + "processId": "geometry.buffer", + "parameterBindings": [ + {"sourceName": "in_geometry", "targetParameter": "wkb", "valueType": "Wkb", "required": True}, + {"sourceName": "buffer_distance", "targetParameter": "distance", "valueType": "FloatingPoint", + "required": True}, + ], + "issues": [], + }, + { + "toolName": "RunCustomScript", + "classification": "unsupported", + "processId": None, + "parameterBindings": [], + "issues": [ + { + "code": "no-native-executor", + "message": "The scanner proposed no native Honua process for this tool.", + }, + { + "code": "unsupported-construct", + "message": "Source construct cannot be translated: custom Python execution body.", + "parameterName": None, + }, + ], + }, + ], +} + + +def _manifest() -> ToolboxTranslationManifest: + return ToolboxTranslationManifest( + toolbox_name="VectorAnalysisToolbox", + source_format="pyt", + source_label="vector_analysis.pyt", + tools=[ + ToolboxToolDescriptor( + tool_name="BufferGeometry", + display_name="Buffer Geometry", + target_process_id="geometry.buffer", + parameter_mappings=[ + ToolboxParameterMapping("in_geometry", "wkb", "GPGeometry"), + ToolboxParameterMapping("buffer_distance", "distance"), + ], + ), + ToolboxToolDescriptor(tool_name="RunCustomScript", unsupported_constructs=["custom Python execution body"]), + ], + ) + + +def test_manifest_serialises_to_the_server_wire_shape() -> None: + payload = _manifest().to_dict() + + assert payload["artifactKind"] == TOOLBOX_TRANSLATION_MANIFEST_KIND + assert payload["artifactVersion"] == "1.0" + assert list(payload)[:2] == ["artifactKind", "artifactVersion"] + assert payload["toolboxName"] == "VectorAnalysisToolbox" + assert payload["sourceFormat"] == "pyt" + assert payload["sourceLabel"] == "vector_analysis.pyt" + assert payload["tools"][0]["parameterMappings"] == [ + {"sourceName": "in_geometry", "targetParameter": "wkb", "sourceDataType": "GPGeometry"}, + {"sourceName": "buffer_distance", "targetParameter": "distance"}, + ] + # An untranslatable tool carries no target rather than a stub target. + assert "targetProcessId" not in payload["tools"][1] + assert payload["tools"][1]["unsupportedConstructs"] == ["custom Python execution body"] + + +def test_validate_toolbox_translation_posts_the_manifest_and_parses_the_report() -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["method"] = request.method + seen["path"] = request.url.path + seen["body"] = json.loads(request.content.decode("utf-8")) + return httpx.Response(200, json=_REPORT) + + transport = httpx.MockTransport(handler) + with HonuaAdminClient("http://test.honua.io", transport=transport) as client: + report = client.validate_toolbox_translation(_manifest()) + + assert seen["method"] == "POST" + assert seen["path"] == "/api/v1/admin/import/toolbox/translation/validate" + assert seen["body"]["artifactKind"] == TOOLBOX_TRANSLATION_MANIFEST_KIND + + assert report.artifact_kind == TOOLBOX_TRANSLATION_REPORT_KIND + assert report.toolbox_name == "VectorAnalysisToolbox" + assert report.summary.tool_count == 2 + assert report.summary.translated_count == 1 + assert report.summary.unsupported_count == 1 + assert [tool.classification for tool in report.tools] == ["translated", "unsupported"] + assert report.tools[0].process_id == "geometry.buffer" + assert report.tools[0].parameter_bindings[0].value_type == "Wkb" + assert report.tools[0].parameter_bindings[0].required is True + assert report.tools[1].issues[0].code == "no-native-executor" + + +def test_validate_toolbox_translation_sends_the_admin_api_key() -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["headers"] = request.headers + return httpx.Response(200, json=_REPORT) + + transport = httpx.MockTransport(handler) + with HonuaAdminClient("http://test.honua.io", api_key="admin-secret", transport=transport) as client: + client.validate_toolbox_translation(_manifest()) + + # The endpoint is in the admin import group, so it rides the existing admin + # credential path rather than a new auth mechanism. + assert seen["headers"]["x-api-key"] == "admin-secret" + + +def test_validate_toolbox_translation_accepts_per_call_options() -> None: + seen: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["headers"] = request.headers + seen["timeout"] = request.extensions.get("timeout", {}) + return httpx.Response(200, json=_REPORT) + + transport = httpx.MockTransport(handler) + with HonuaAdminClient("http://test.honua.io", transport=transport) as client: + client.validate_toolbox_translation( + _manifest(), + timeout=2.5, + extra_headers={"X-Trace-Id": "trace-9"}, + idempotency_key="toolbox-key", + ) + + assert seen["headers"]["x-trace-id"] == "trace-9" + assert seen["headers"]["idempotency-key"] == "toolbox-key" + assert seen["timeout"]["connect"] == 2.5 + + +def test_validate_toolbox_translation_raises_on_a_rejected_manifest() -> None: + from honua_sdk.errors import HonuaHttpError + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(400, json={"error": "sourceFormat must be one of: pyt, tbx, atbx."}) + + transport = httpx.MockTransport(handler) + with HonuaAdminClient("http://test.honua.io", transport=transport) as client, pytest.raises(HonuaHttpError): + client.validate_toolbox_translation( + ToolboxTranslationManifest(toolbox_name="T", source_format="docx", tools=[]) + ) + + +def test_report_from_dict_tolerates_a_sparse_payload() -> None: + from honua_admin import ToolboxTranslationReport + + report = ToolboxTranslationReport.from_dict({"toolboxName": "T", "sourceFormat": "pyt"}) + + assert report.summary.tool_count == 0 + assert report.tools == [] + + +def test_report_from_dict_never_synthesises_a_missing_artifact_identity() -> None: + """A 200 without artifactKind/artifactVersion must stay identity-less. + + Defaulting these to the expected values would make a malformed or + non-translation-report response indistinguishable from a genuine v1 report, + and the attestation layer -- which consumes ``to_dict()`` -- would then have + nothing left to reject it on (honua-sdk-python#188 review). + """ + + from honua_admin import ToolboxTranslationReport + + report = ToolboxTranslationReport.from_dict( + {"toolboxName": "T", "sourceFormat": "pyt", "tools": []} + ) + + assert report.artifact_kind is None + assert report.artifact_version is None + payload = report.to_dict() + assert "artifactKind" not in payload + assert "artifactVersion" not in payload + + +def test_report_round_trips_a_present_artifact_identity_first() -> None: + from honua_admin import ToolboxTranslationReport + + payload = ToolboxTranslationReport.from_dict(_REPORT).to_dict() + + assert list(payload)[:2] == ["artifactKind", "artifactVersion"] + assert payload["artifactKind"] == TOOLBOX_TRANSLATION_REPORT_KIND + assert payload["artifactVersion"] == "1.0" + + +@pytest.mark.anyio +async def test_async_validate_toolbox_translation() -> None: + seen: dict[str, Any] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + seen["path"] = request.url.path + seen["body"] = json.loads(request.content.decode("utf-8")) + return httpx.Response(200, json=_REPORT) + + transport = httpx.MockTransport(handler) + async with AsyncHonuaAdminClient("http://test.honua.io", transport=transport) as client: + report = await client.validate_toolbox_translation(_manifest()) + + assert seen["path"] == "/api/v1/admin/import/toolbox/translation/validate" + assert seen["body"]["tools"][0]["toolName"] == "BufferGeometry" + assert report.summary.translated_count == 1 diff --git a/tests/test_arcpy_migration_attestation.py b/tests/test_arcpy_migration_attestation.py new file mode 100644 index 0000000..d51a583 --- /dev/null +++ b/tests/test_arcpy_migration_attestation.py @@ -0,0 +1,988 @@ +"""Tests for server-attested toolbox translation reports (honua-sdk-python#188). + +The contract under test is narrow but load-bearing: a migration report must say +whether its verdict came from the server's canonical process catalog or only +from the SDK's local view of it, and a local fallback must never be dressed up +as attested. These tests pin all four paths the issue calls out -- attested +success, server-vs-local disagreement, an unreachable server, and a failed +(unauthorized) call -- plus the offline default. +""" + +from __future__ import annotations + +import pytest + +from honua_sdk.migration import ( + AGREEMENT_AGREED, + AGREEMENT_DISAGREED, + AGREEMENT_NOT_ATTESTED, + CLASSIFICATION_PARTIALLY_TRANSLATED, + CLASSIFICATION_TRANSLATED, + CLASSIFICATION_UNSUPPORTED, + LOCAL_ONLY, + MAX_MANIFEST_TOOLS, + SERVER_ATTESTED, + SOURCE_FORMAT_ATBX, + SOURCE_FORMAT_PYT, + SOURCE_FORMAT_TBX, + ModelBuilderToolbox, + TranslationManifest, + TranslationToolProposal, + attest_translation, + build_atbx_translation_manifest, + build_pyt_translation_manifest, + parse_pyt_source, + resolve_argument_bindings, + scan_arcpy_source, + source_format_for_path, +) + +# Buffer -> translatable (geometry.buffer); Erase -> manual-review (the server +# cannot job-execute it); Kriging -> unsupported (no registered mapping). +PYT_SOURCE = ''' +import arcpy + + +class Toolbox(object): + def __init__(self): + self.label = "Roads Toolbox" + self.alias = "roads" + self.tools = [BufferRoads, EraseRivers, InterpolateStations, EmptyTool] + + +class BufferRoads(object): + def __init__(self): + self.label = "Buffer Roads" + + def execute(self, parameters, messages): + arcpy.analysis.Buffer("roads", "roads_buffer", "25 Meters") + + +class EraseRivers(object): + def __init__(self): + self.label = "Erase Rivers" + + def execute(self, parameters, messages): + arcpy.analysis.Erase("a", "b", "c") + + +class InterpolateStations(object): + def __init__(self): + self.label = "Interpolate" + + def execute(self, parameters, messages): + arcpy.sa.Kriging("stations", "PredZ") + + +class EmptyTool(object): + def __init__(self): + self.label = "Empty" + + def execute(self, parameters, messages): + return None +''' + + +def _manifest() -> TranslationManifest: + toolbox = parse_pyt_source(PYT_SOURCE, filename="/home/operator/private/roads.pyt") + return build_pyt_translation_manifest(toolbox) + + +def _server_report(manifest: TranslationManifest, **classifications: str) -> dict[str, object]: + """A server report classifying every submitted tool. + + Anything not named in ``classifications`` echoes the local verdict back, so a + test only has to spell out the tools it wants the server to disagree about. + """ + + return { + "artifactKind": "honua.migration.toolbox-translation-report", + "artifactVersion": "1.0", + "toolboxName": manifest.toolbox_name, + "sourceFormat": manifest.source_format, + "summary": {}, + "tools": [ + { + "toolName": tool.tool_name, + "classification": classifications.get(tool.tool_name, tool.local_classification), + "processId": tool.target_process_id, + "parameterBindings": [], + "issues": [], + } + for tool in manifest.tools + ], + } + + +# --------------------------------------------------------------------------- +# Manifest construction +# --------------------------------------------------------------------------- + + +def test_build_pyt_manifest_declares_the_server_artifact_identity() -> None: + payload = _manifest().to_dict() + + assert payload["artifactKind"] == "honua.migration.toolbox-translation" + assert payload["artifactVersion"] == "1.0" + assert payload["sourceFormat"] == SOURCE_FORMAT_PYT + assert payload["toolboxName"] == "Roads Toolbox" + + +def test_build_pyt_manifest_redacts_the_local_directory_from_the_source_label() -> None: + payload = _manifest().to_dict() + + # The server echoes sourceLabel into operator-visible output, so only the + # basename travels -- never the operator's directory layout. + assert payload["sourceLabel"] == "roads.pyt" + assert "/home/operator" not in str(payload) + + +def test_build_pyt_manifest_classifies_each_tool_locally() -> None: + local = {tool.tool_name: tool.local_classification for tool in _manifest().tools} + + assert local["BufferRoads"] == CLASSIFICATION_TRANSLATED + # Registered but not job-executable is not runnable, so it is unsupported. + assert local["EraseRivers"] == CLASSIFICATION_UNSUPPORTED + assert local["InterpolateStations"] == CLASSIFICATION_UNSUPPORTED + # A tool with no recognized GP call still appears, rather than vanishing + # from the tool count. + assert local["EmptyTool"] == CLASSIFICATION_UNSUPPORTED + + +def test_build_pyt_manifest_proposes_a_native_target_and_parameter_mappings() -> None: + buffer_tool = next(tool for tool in _manifest().tools if tool.tool_name == "BufferRoads") + + assert buffer_tool.target_process_id == "geometry.buffer" + mappings = {mapping.source_name: mapping.target_parameter for mapping in buffer_tool.parameter_mappings} + assert mappings["in_features"] == "input_features" + assert mappings["buffer_distance_or_field"] == "distance" + # Output destinations are not canonical process inputs, so they are not + # proposed as parameter mappings. + assert "out_feature_class" not in mappings + + +def test_build_pyt_manifest_reports_an_unmapped_keyword_as_an_unsupported_construct() -> None: + source = ''' +import arcpy + + +class Toolbox(object): + def __init__(self): + self.tools = [T] + + +class T(object): + def execute(self, parameters, messages): + arcpy.analysis.Buffer("roads", "out", "25 Meters", not_a_real_arg=1) +''' + manifest = build_pyt_translation_manifest(parse_pyt_source(source, filename="t.pyt")) + tool = manifest.tools[0] + + assert any("not_a_real_arg" in construct for construct in tool.unsupported_constructs) + # A translatable call carrying an unmapped construct is partial, not clean. + assert tool.local_classification == CLASSIFICATION_PARTIALLY_TRANSLATED + + +def test_build_pyt_manifest_splits_a_multi_step_tool_into_unique_names() -> None: + source = ''' +import arcpy + + +class Toolbox(object): + def __init__(self): + self.tools = [T] + + +class T(object): + def execute(self, parameters, messages): + arcpy.analysis.Buffer("roads", "out", "25 Meters") + arcpy.analysis.Clip("out", "aoi", "clipped") +''' + manifest = build_pyt_translation_manifest(parse_pyt_source(source, filename="t.pyt")) + + # The endpoint certifies one native process per manifest tool, and rejects a + # duplicate toolName outright, so steps get distinct names. + names = [tool.tool_name for tool in manifest.tools] + assert names == ["T#1", "T#2"] + assert len(set(names)) == len(names) + + +def test_resolve_argument_bindings_returns_nothing_for_an_unmapped_call() -> None: + call = scan_arcpy_source("import arcpy\narcpy.sa.Kriging('a', 'b')\n").calls[0] + + assert resolve_argument_bindings(call) == () + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + ("toolbox.pyt", SOURCE_FORMAT_PYT), + ("Toolbox.ATBX", SOURCE_FORMAT_ATBX), + ("legacy.tbx", SOURCE_FORMAT_TBX), + ("workflow.py", None), + ("service.json", None), + ], +) +def test_source_format_for_path(path: str, expected: str | None) -> None: + assert source_format_for_path(path) == expected + + +def test_manifest_batches_respect_the_server_tool_cap() -> None: + tools = tuple( + TranslationToolProposal(tool_name=f"T{index}", local_classification=CLASSIFICATION_UNSUPPORTED) + for index in range(MAX_MANIFEST_TOOLS + 5) + ) + manifest = TranslationManifest(toolbox_name="Big", source_format=SOURCE_FORMAT_PYT, tools=tools) + + batches = manifest.batches() + + assert [len(batch.tools) for batch in batches] == [MAX_MANIFEST_TOOLS, 5] + assert [tool.tool_name for batch in batches for tool in batch.tools] == [t.tool_name for t in tools] + + +def test_manifest_batches_rejects_a_non_positive_size() -> None: + with pytest.raises(ValueError, match="at least 1"): + _manifest().batches(0) + + +# --------------------------------------------------------------------------- +# Attested success +# --------------------------------------------------------------------------- + + +def test_attest_translation_marks_a_server_verdict_as_attested() -> None: + manifest = _manifest() + submitted: list[TranslationManifest] = [] + + def validator(batch: TranslationManifest) -> dict[str, object]: + submitted.append(batch) + return _server_report(batch) + + report = attest_translation(manifest, validator=validator, server="https://honua.test") + + assert report.attested is True + assert report.verdict_source == SERVER_ATTESTED + assert report.fallback_reason is None + assert report.server == "https://honua.test" + assert [tool.tool_name for tool in submitted[0].tools] == [t.tool_name for t in manifest.tools] + assert all(verdict.agreement == AGREEMENT_AGREED for verdict in report.tools) + + document = report.to_dict() + assert document["verdictSource"] == SERVER_ATTESTED + assert document["attested"] is True + assert document["summary"]["toolCount"] == len(manifest.tools) + assert document["summary"]["disagreementCount"] == 0 + + +def test_attest_translation_carries_the_server_bindings_and_issues() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + payload = _server_report(batch) + tools = payload["tools"] + assert isinstance(tools, list) + tools[0]["parameterBindings"] = [ + {"sourceName": "in_features", "targetParameter": "wkb", "valueType": "Wkb", "required": True} + ] + tools[0]["issues"] = [{"code": "missing-required-parameter", "message": "srid is not mapped."}] + tools[0]["processId"] = "geometry.buffer" + return payload + + verdict = attest_translation(manifest, validator=validator).tools[0] + + assert verdict.process_id == "geometry.buffer" + assert verdict.parameter_bindings[0]["valueType"] == "Wkb" + assert verdict.issues[0]["code"] == "missing-required-parameter" + + +def test_attest_translation_submits_every_batch_of_a_large_toolbox() -> None: + tools = tuple( + TranslationToolProposal(tool_name=f"T{index}", local_classification=CLASSIFICATION_UNSUPPORTED) + for index in range(MAX_MANIFEST_TOOLS + 3) + ) + manifest = TranslationManifest(toolbox_name="Big", source_format=SOURCE_FORMAT_PYT, tools=tools) + calls = 0 + + def validator(batch: TranslationManifest) -> dict[str, object]: + nonlocal calls + calls += 1 + return _server_report(batch) + + report = attest_translation(manifest, validator=validator) + + assert calls == 2 + assert report.attested is True + assert len(report.tools) == len(tools) + + +# --------------------------------------------------------------------------- +# Disagreement: the server wins, and the disagreement is surfaced +# --------------------------------------------------------------------------- + + +def test_server_verdict_overrides_the_local_one_and_the_disagreement_is_surfaced() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + # The SDK called this one translated; the canonical catalog does not. + return _server_report(batch, BufferRoads=CLASSIFICATION_UNSUPPORTED) + + report = attest_translation(manifest, validator=validator) + buffer_verdict = next(verdict for verdict in report.tools if verdict.tool_name == "BufferRoads") + + assert buffer_verdict.classification == CLASSIFICATION_UNSUPPORTED + assert buffer_verdict.server_classification == CLASSIFICATION_UNSUPPORTED + # The local verdict is retained beside the server's, not overwritten, so the + # drift is auditable rather than invisible. + assert buffer_verdict.local_classification == CLASSIFICATION_TRANSLATED + assert buffer_verdict.agreement == AGREEMENT_DISAGREED + assert buffer_verdict.disagreed is True + + assert [verdict.tool_name for verdict in report.disagreements] == ["BufferRoads"] + document = report.to_dict() + assert document["summary"]["disagreementCount"] == 1 + assert document["disagreements"] == [ + {"toolName": "BufferRoads", "local": CLASSIFICATION_TRANSLATED, "server": CLASSIFICATION_UNSUPPORTED} + ] + + +def test_a_server_verdict_more_generous_than_the_local_one_also_wins() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + return _server_report(batch, EraseRivers=CLASSIFICATION_TRANSLATED) + + verdict = next(v for v in attest_translation(manifest, validator=validator).tools if v.tool_name == "EraseRivers") + + assert verdict.classification == CLASSIFICATION_TRANSLATED + assert verdict.local_classification == CLASSIFICATION_UNSUPPORTED + assert verdict.disagreed is True + + +# --------------------------------------------------------------------------- +# Offline / degraded paths: never presented as attested +# --------------------------------------------------------------------------- + + +def test_no_validator_produces_a_local_only_report_with_a_stated_reason() -> None: + report = attest_translation(_manifest()) + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason is not None + assert "has not been attested" in report.fallback_reason + # The offline report is still complete and usable, just not attested. + assert len(report.tools) == len(_manifest().tools) + assert all(verdict.agreement == AGREEMENT_NOT_ATTESTED for verdict in report.tools) + assert all(verdict.server_classification is None for verdict in report.tools) + + document = report.to_dict() + assert document["attested"] is False + assert document["verdictSource"] == LOCAL_ONLY + + +def test_an_unreachable_server_degrades_to_local_only_rather_than_failing() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + raise ConnectionError("All connection attempts failed") + + report = attest_translation(manifest, validator=validator, server="https://offline.test") + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason == "ConnectionError: All connection attempts failed" + # The local verdicts survive, so the run still produces a migration report. + assert {verdict.tool_name for verdict in report.tools} == {tool.tool_name for tool in manifest.tools} + assert next(v for v in report.tools if v.tool_name == "BufferRoads").classification == CLASSIFICATION_TRANSLATED + + +def test_an_unauthorized_call_is_never_reported_as_attested() -> None: + class HonuaAuthError(Exception): + pass + + def validator(batch: TranslationManifest) -> dict[str, object]: + raise HonuaAuthError("401 Unauthorized") + + report = attest_translation(_manifest(), validator=validator, server="https://honua.test") + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason == "HonuaAuthError: 401 Unauthorized" + assert report.to_dict()["attested"] is False + assert all(verdict.server_classification is None for verdict in report.tools) + + +def test_a_failure_with_no_message_still_names_the_failure_type() -> None: + def validator(batch: TranslationManifest) -> dict[str, object]: + raise TimeoutError + + report = attest_translation(_manifest(), validator=validator) + + assert report.fallback_reason == "TimeoutError" + + +# Everything a report needs before the per-tool checks are reached: the v1 +# artifact identity plus the binding back to the submitted toolbox. Cases below +# break exactly one thing at a time on top of this. +_VALID_IDENTITY = { + "artifactKind": "honua.migration.toolbox-translation-report", + "artifactVersion": "1.0", + "toolboxName": "Roads Toolbox", + "sourceFormat": "pyt", +} + + +@pytest.mark.parametrize( + ("payload", "expected"), + [ + pytest.param([], "list where a translation report object", id="not-an-object"), + pytest.param( + {**_VALID_IDENTITY, "artifactKind": "honua.migration.source-inventory", "tools": []}, + "artifactKind", + id="wrong-artifact", + ), + pytest.param({**_VALID_IDENTITY, "summary": {}}, "no 'tools' array", id="no-tools"), + pytest.param({**_VALID_IDENTITY, "tools": ["nope"]}, "non-object tool entry", id="non-object-entry"), + pytest.param( + {**_VALID_IDENTITY, "tools": [{"toolName": "BufferRoads"}]}, + "toolName/classification", + id="no-classification", + ), + ], +) +def test_a_malformed_server_report_degrades_to_local_only(payload: object, expected: str) -> None: + report = attest_translation(_manifest(), validator=lambda batch: payload) # type: ignore[arg-type,return-value] + + assert report.attested is False + assert report.fallback_reason is not None + assert expected in report.fallback_reason + + +def test_a_report_missing_a_submitted_tool_is_not_attested() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + payload = _server_report(batch) + tools = payload["tools"] + assert isinstance(tools, list) + # A partially-classified response cannot back a whole-toolbox claim. + return {**payload, "tools": tools[:1]} + + report = attest_translation(manifest, validator=validator) + + assert report.attested is False + assert report.fallback_reason is not None + assert "did not classify every submitted tool" in report.fallback_reason + + +def test_a_report_classifying_an_unsubmitted_tool_is_not_attested() -> None: + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + payload = _server_report(batch) + tools = payload["tools"] + assert isinstance(tools, list) + tools.append({"toolName": "NeverSubmitted", "classification": CLASSIFICATION_TRANSLATED}) + return payload + + report = attest_translation(manifest, validator=validator) + + assert report.attested is False + assert report.fallback_reason is not None + assert "which was not submitted" in report.fallback_reason + + +def test_a_batch_failing_after_a_successful_one_degrades_the_whole_report() -> None: + tools = tuple( + TranslationToolProposal(tool_name=f"T{index}", local_classification=CLASSIFICATION_UNSUPPORTED) + for index in range(MAX_MANIFEST_TOOLS + 1) + ) + manifest = TranslationManifest(toolbox_name="Big", source_format=SOURCE_FORMAT_PYT, tools=tools) + calls = 0 + + def validator(batch: TranslationManifest) -> dict[str, object]: + nonlocal calls + calls += 1 + if calls == 1: + return _server_report(batch) + raise ConnectionError("dropped mid-toolbox") + + report = attest_translation(manifest, validator=validator) + + # There is no partial attestation: one failed batch un-attests the toolbox. + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert all(verdict.agreement == AGREEMENT_NOT_ATTESTED for verdict in report.tools) + + +# --------------------------------------------------------------------------- +# Review regressions (honua-sdk-python#188): three ways a report could have +# claimed `attested: true` while not actually being fully attested. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("payload", "expected"), + [ + pytest.param( + {"artifactVersion": "1.0", "toolboxName": "T", "sourceFormat": "pyt", "tools": []}, + "artifactKind None", + id="missing-artifact-kind", + ), + pytest.param( + { + "artifactKind": "honua.migration.toolbox-translation-report", + "toolboxName": "T", + "sourceFormat": "pyt", + "tools": [], + }, + "artifactVersion None", + id="missing-artifact-version", + ), + pytest.param( + {**_VALID_IDENTITY, "artifactVersion": "2.0", "tools": []}, + "artifactVersion '2.0'", + id="unreadable-artifact-version", + ), + ], +) +def test_a_report_without_a_usable_artifact_identity_is_not_attested(payload: dict, expected: str) -> None: + """A 200 that does not identify itself as a v1 report cannot back attestation. + + The genuine endpoint always stamps both identity fields. Treating them as + optional -- or letting the response model default them in -- would make an + error envelope, a proxy page, or a different API's payload indistinguishable + from a real report. + """ + + report = attest_translation(_manifest(), validator=lambda batch: payload) + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason is not None + assert expected in report.fallback_reason + assert all(verdict.server_classification is None for verdict in report.tools) + + +def test_an_admin_report_missing_its_identity_is_not_attested_end_to_end() -> None: + """The admin response model must not paper over a missing identity. + + This is the real CLI path: the admin client parses the response into + ``ToolboxTranslationReport`` and hands ``to_dict()`` to the attestation + layer. If the model defaulted the identity fields, that layer would receive + a perfectly-formed v1 report and mark a malformed response attested. + """ + + from honua_admin import ToolboxTranslationReport + + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + # A 200 body with no artifactKind/artifactVersion at all. + body = { + "toolboxName": batch.toolbox_name, + "sourceFormat": batch.source_format, + "summary": {}, + "tools": [ + {"toolName": tool.tool_name, "classification": CLASSIFICATION_TRANSLATED} + for tool in batch.tools + ], + } + return ToolboxTranslationReport.from_dict(body).to_dict() + + report = attest_translation(manifest, validator=validator, server="https://honua.test") + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert "artifactKind" in str(report.fallback_reason) + + +@pytest.mark.parametrize("classification", ["manual-review", "translated-v2", "TRANSLATED", ""]) +def test_a_classification_outside_the_vocabulary_is_not_attested(classification: str) -> None: + """An unknown classification would produce an attestation that does not add up. + + The summary counts tools by exactly the three declared values, so a fourth + value would appear as a tool's effective classification while no counter + included it. Refuse the report instead of emitting an inconsistent one. + """ + + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + payload = _server_report(batch) + tools = payload["tools"] + assert isinstance(tools, list) + tools[0]["classification"] = classification + return payload + + report = attest_translation(manifest, validator=validator, server="https://honua.test") + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason is not None + assert "outside this report format's vocabulary" in report.fallback_reason + # Nothing from the refused response leaks into the local-only report. + assert all(verdict.server_classification is None for verdict in report.tools) + assert all(verdict.agreement == AGREEMENT_NOT_ATTESTED for verdict in report.tools) + + +def test_an_attested_report_summary_accounts_for_every_tool() -> None: + """The invariant the vocabulary check exists to protect.""" + + manifest = _manifest() + report = attest_translation(manifest, validator=lambda batch: _server_report(batch)) + summary = report.to_dict()["summary"] + + assert report.attested is True + counted = ( + summary["translatedCount"] + summary["partiallyTranslatedCount"] + summary["unsupportedCount"] + ) + assert counted == summary["toolCount"] == len(manifest.tools) + + +def test_atbx_manifest_includes_script_tools_alongside_models() -> None: + """Script tools must be submitted, not silently omitted. + + ``parse_atbx_toolbox`` records a script tool by name only (its logic lives + in an external .py the reader does not follow). Building the manifest from + ``models`` alone let the server return a clean report for a strict subset of + the toolbox, which the CLI then presented as a whole-toolbox attestation + (honua-sdk-python#188 review). + """ + + toolbox = ModelBuilderToolbox( + filename="/home/operator/private/wf.atbx", + models=(), + script_tool_names=("LegacyScriptTool", "AnotherScript"), + ) + + manifest = build_atbx_translation_manifest(toolbox) + + assert [tool.tool_name for tool in manifest.tools] == ["LegacyScriptTool", "AnotherScript"] + for tool in manifest.tools: + # No native target may be proposed: the translator never read the body. + assert tool.target_process_id is None + assert tool.local_classification == CLASSIFICATION_UNSUPPORTED + assert any("script tool" in construct for construct in tool.unsupported_constructs) + assert any("arcpy .py scanner" in construct for construct in tool.unsupported_constructs) + + +def test_atbx_attestation_covers_every_discovered_tool(tmp_path) -> None: + """End-to-end: an .atbx holding both a model and a script tool.""" + + import io + import json as _json + import zipfile + + from honua_sdk.migration import parse_atbx_toolbox + + model = { + "name": "BufferModel", + "processes": [ + { + "toolName": "Buffer", + "toolbox": "analysis", + "parameters": { + "in_features": "a", + "out_feature_class": "b", + "buffer_distance_or_field": "5 Meters", + }, + } + ], + } + script_tool = {"name": "LegacyScriptTool", "type": "script", "script": "legacy.py"} + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("BufferModel.tool/tool.content", _json.dumps(model)) + archive.writestr("LegacyScriptTool.tool/tool.content", _json.dumps(script_tool)) + path = tmp_path / "wf.atbx" + path.write_bytes(buffer.getvalue()) + + toolbox = parse_atbx_toolbox(path) + assert toolbox.models, "fixture should parse one model" + assert toolbox.script_tool_names == ("LegacyScriptTool",), "fixture should record one script tool" + + manifest = build_atbx_translation_manifest(toolbox) + submitted = [tool.tool_name for tool in manifest.tools] + assert "BufferModel" in submitted + assert "LegacyScriptTool" in submitted + + report = attest_translation(manifest, validator=lambda batch: _server_report(batch)) + + assert report.attested is True + # The attestation covers the whole toolbox, not just its models. + assert {verdict.tool_name for verdict in report.tools} == {"BufferModel", "LegacyScriptTool"} + assert report.to_dict()["summary"]["toolCount"] == 2 + + +def test_pyt_manifest_includes_declared_tools_whose_class_is_not_in_the_file() -> None: + """A declared-but-unmaterialised tool must be submitted, not silently omitted. + + ``self.tools = [Present, ImportedTool]`` keeps both names in + ``declared_tool_names``, but only classes defined in the same file become + ``toolbox.tools`` -- there is no ``execute`` body to read for an imported + one. Building the manifest from ``toolbox.tools`` alone let the server + return a clean report covering only ``Present`` while the CLI presented it + as whole-toolbox attestation (honua-sdk-python#188 review). + """ + + source = ''' +import arcpy +from other_module import ImportedTool + + +class Toolbox(object): + def __init__(self): + self.label = "T" + self.tools = [Present, ImportedTool] + + +class Present(object): + def execute(self, parameters, messages): + arcpy.analysis.Buffer("a", "b", "1 Meter") +''' + toolbox = parse_pyt_source(source, filename="t.pyt") + # Precondition: the reader really does drop the imported name. + assert toolbox.declared_tool_names == ("Present", "ImportedTool") + assert [tool.class_name for tool in toolbox.tools] == ["Present"] + + manifest = build_pyt_translation_manifest(toolbox) + + assert [tool.tool_name for tool in manifest.tools] == ["Present", "ImportedTool"] + imported = manifest.tools[1] + # Nothing was read, so nothing may be proposed. + assert imported.target_process_id is None + assert imported.local_classification == CLASSIFICATION_UNSUPPORTED + assert any("not defined in this .pyt" in c for c in imported.unsupported_constructs) + assert any("arcpy .py scanner" in c for c in imported.unsupported_constructs) + + +def test_pyt_attestation_covers_declared_tools_the_reader_could_not_materialise() -> None: + """End-to-end: attestation must not certify a strict subset of the toolbox.""" + + source = ''' +import arcpy +from vendor.tools import VendorTool + + +class Toolbox(object): + def __init__(self): + self.label = "Mixed" + self.tools = [Local, VendorTool] + + +class Local(object): + def execute(self, parameters, messages): + arcpy.analysis.Buffer("a", "b", "1 Meter") +''' + manifest = build_pyt_translation_manifest(parse_pyt_source(source, filename="mixed.pyt")) + submitted = [tool.tool_name for tool in manifest.tools] + assert submitted == ["Local", "VendorTool"] + + report = attest_translation(manifest, validator=lambda batch: _server_report(batch)) + + assert report.attested is True + assert {verdict.tool_name for verdict in report.tools} == {"Local", "VendorTool"} + summary = report.to_dict()["summary"] + assert summary["toolCount"] == 2 + assert ( + summary["translatedCount"] + summary["partiallyTranslatedCount"] + summary["unsupportedCount"] + == 2 + ) + + +def test_a_pyt_toolbox_whose_tools_are_all_imported_still_submits_them() -> None: + """The degenerate case: nothing materialises, so nothing would be submitted. + + Before the fix this produced an empty manifest, which the endpoint rejects + ("tools is required") -- turning a real coverage gap into an unexplained + attestation failure instead of an honest unsupported report. + """ + + source = ''' +from vendor.tools import AlphaTool, BetaTool + + +class Toolbox(object): + def __init__(self): + self.label = "All imported" + self.tools = [AlphaTool, BetaTool] +''' + manifest = build_pyt_translation_manifest(parse_pyt_source(source, filename="vendor.pyt")) + + assert [tool.tool_name for tool in manifest.tools] == ["AlphaTool", "BetaTool"] + assert all(tool.target_process_id is None for tool in manifest.tools) + assert all(tool.local_classification == CLASSIFICATION_UNSUPPORTED for tool in manifest.tools) + + +def _atbx_with(tmp_path, entries: dict) -> object: + import io + import json as _json + import zipfile + + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + for name, payload in entries.items(): + archive.writestr(name, _json.dumps(payload)) + path = tmp_path / "wf.atbx" + path.write_bytes(buffer.getvalue()) + return path + + +def test_atbx_manifest_includes_models_that_yielded_no_steps(tmp_path) -> None: + """A declared model with no recognizable step must still be submitted. + + ``parse_atbx_toolbox`` deliberately keeps a stepless model out of ``models`` + -- there is nothing to translate -- but the name is still a tool the toolbox + declares. Dropping it entirely let the manifest under-count the toolbox and + certify coverage it did not have (honua-sdk-python#188 review follow-up). + """ + + from honua_sdk.migration import parse_atbx_toolbox + + path = _atbx_with( + tmp_path, + { + "BufferModel.tool/tool.content": { + "name": "BufferModel", + "processes": [ + { + "toolName": "Buffer", + "toolbox": "analysis", + "parameters": { + "in_features": "a", + "out_feature_class": "b", + "buffer_distance_or_field": "5 Meters", + }, + } + ], + }, + "EmptyModel.tool/tool.content": {"type": "ModelTool", "processes": []}, + }, + ) + + toolbox = parse_atbx_toolbox(path) + # The reader's existing contract is unchanged: a stepless model is not a model. + assert [model.name for model in toolbox.models] == ["BufferModel"] + # ...but it is now discoverable rather than lost. + assert toolbox.unresolved_tool_names == ("EmptyModel",) + assert toolbox.to_dict()["unresolvedToolNames"] == ["EmptyModel"] + + manifest = build_atbx_translation_manifest(toolbox) + + assert [tool.tool_name for tool in manifest.tools] == ["BufferModel", "EmptyModel"] + empty = manifest.tools[1] + assert empty.target_process_id is None + assert empty.local_classification == CLASSIFICATION_UNSUPPORTED + assert any("no recognizable" in construct for construct in empty.unsupported_constructs) + + report = attest_translation(manifest, validator=lambda batch: _server_report(batch)) + + assert report.attested is True + assert {verdict.tool_name for verdict in report.tools} == {"BufferModel", "EmptyModel"} + assert report.to_dict()["summary"]["toolCount"] == 2 + + +def test_atbx_manifest_covers_models_script_tools_and_unresolved_together(tmp_path) -> None: + """All three .atbx tool kinds land in one manifest.""" + + from honua_sdk.migration import parse_atbx_toolbox + + path = _atbx_with( + tmp_path, + { + "BufferModel.tool/tool.content": { + "name": "BufferModel", + "processes": [ + { + "toolName": "Buffer", + "toolbox": "analysis", + "parameters": { + "in_features": "a", + "out_feature_class": "b", + "buffer_distance_or_field": "5 Meters", + }, + } + ], + }, + "LegacyScriptTool.tool/tool.content": { + "name": "LegacyScriptTool", + "type": "script", + "script": "legacy.py", + }, + "EmptyModel.tool/tool.content": {"type": "ModelTool", "processes": []}, + }, + ) + + toolbox = parse_atbx_toolbox(path) + manifest = build_atbx_translation_manifest(toolbox) + submitted = {tool.tool_name for tool in manifest.tools} + + assert submitted == {"BufferModel", "LegacyScriptTool", "EmptyModel"} + + report = attest_translation(manifest, validator=lambda batch: _server_report(batch)) + summary = report.to_dict()["summary"] + + assert report.attested is True + assert summary["toolCount"] == 3 + assert ( + summary["translatedCount"] + summary["partiallyTranslatedCount"] + summary["unsupportedCount"] + == 3 + ) + + +@pytest.mark.parametrize( + ("override", "expected"), + [ + pytest.param({"toolboxName": "SomeOtherToolbox"}, "not the submitted", id="wrong-toolbox"), + pytest.param({"toolboxName": None}, "not the submitted", id="missing-toolbox"), + pytest.param({"sourceFormat": "atbx"}, "sourceFormat", id="wrong-source-format"), + pytest.param({"sourceFormat": None}, "sourceFormat", id="missing-source-format"), + ], +) +def test_a_report_for_a_different_artifact_is_not_attested(override: dict, expected: str) -> None: + """A report only attests the artifact it was asked about. + + Matching tool names are not enough — two toolboxes can share them — so a + stale or misrouted response would otherwise attest the wrong artifact + entirely (honua-sdk-python#188 review). + """ + + manifest = _manifest() + + def validator(batch: TranslationManifest) -> dict[str, object]: + return {**_server_report(batch), **override} + + report = attest_translation(manifest, validator=validator, server="https://honua.test") + + assert report.attested is False + assert report.verdict_source == LOCAL_ONLY + assert report.fallback_reason is not None + assert expected in report.fallback_reason + assert all(verdict.server_classification is None for verdict in report.tools) + + +def test_report_binding_tolerates_the_servers_own_normalisation() -> None: + """The server trims the name and lower-cases the format; that must still bind.""" + + manifest = TranslationManifest( + toolbox_name="Roads Toolbox", + source_format="pyt", + tools=( + TranslationToolProposal( + tool_name="A", local_classification=CLASSIFICATION_UNSUPPORTED + ), + ), + ) + + def validator(batch: TranslationManifest) -> dict[str, object]: + payload = _server_report(batch) + # Exactly what the endpoint does: Trim() the name, ToLowerInvariant() the format. + payload["toolboxName"] = " Roads Toolbox " + payload["sourceFormat"] = "PYT" + return payload + + report = attest_translation(manifest, validator=validator) + + assert report.attested is True diff --git a/tests/test_arcpy_migration_cli.py b/tests/test_arcpy_migration_cli.py index e39f7ae..e8aec05 100644 --- a/tests/test_arcpy_migration_cli.py +++ b/tests/test_arcpy_migration_cli.py @@ -260,3 +260,322 @@ def test_cli_translate_reports_syntax_error_and_emits_no_plan(tmp_path: Path, ca assert not evidence_out.exists() assert "coverage:" not in captured.err assert captured.out == "" + + +# --------------------------------------------------------------------------- +# Server attestation (honua-sdk-python#188) +# --------------------------------------------------------------------------- + + +def _attesting_admin_client(monkeypatch, handler): + """Point the CLI's lazily-imported admin client at a MockTransport.""" + + import httpx + + import honua_admin + from honua_admin import HonuaAdminClient + + seen: dict = {} + + def fake_client(base_url, **kwargs): + seen["base_url"] = base_url + seen["api_key"] = kwargs.get("api_key") + seen["timeout"] = kwargs.get("timeout") + return HonuaAdminClient(base_url, transport=httpx.MockTransport(handler), **kwargs) + + monkeypatch.setattr(honua_admin, "HonuaAdminClient", fake_client) + return seen + + +def _report_for(request, classifications: dict[str, str] | None = None) -> dict: + """Echo back a server report classifying every tool in the posted manifest.""" + + manifest = json.loads(request.content.decode("utf-8")) + overrides = classifications or {} + return { + "artifactKind": "honua.migration.toolbox-translation-report", + "artifactVersion": "1.0", + "toolboxName": manifest["toolboxName"], + "sourceFormat": manifest["sourceFormat"], + "summary": {}, + "tools": [ + { + "toolName": tool["toolName"], + "classification": overrides.get(tool["toolName"], "translated"), + "processId": tool.get("targetProcessId"), + "parameterBindings": [], + "issues": [], + } + for tool in manifest["tools"] + ], + } + + +def test_cli_pyt_without_server_marks_the_report_local_only(tmp_path: Path, capsys) -> None: + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + + rc = main(["pyt", str(toolbox), "--output", str(out)]) + + assert rc == 0 + attestation = json.loads(out.read_text())["attestation"] + assert attestation["verdictSource"] == "local-only" + assert attestation["attested"] is False + assert attestation["fallbackReason"] + assert "local-only (NOT server-attested)" in capsys.readouterr().err + + +def test_cli_pyt_with_server_emits_a_server_attested_report(tmp_path: Path, monkeypatch, capsys) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + attested_out = tmp_path / "attestation.json" + paths: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + paths.append(request.url.path) + return httpx.Response(200, json=_report_for(request)) + + seen = _attesting_admin_client(monkeypatch, handler) + + rc = main( + [ + "pyt", + str(toolbox), + "--output", + str(out), + "--server", + "http://honua.test", + "--api-key", + "admin-secret", + "--attestation", + str(attested_out), + ] + ) + + assert rc == 0 + assert paths == ["/api/v1/admin/import/toolbox/translation/validate"] + assert seen["base_url"] == "http://honua.test" + assert seen["api_key"] == "admin-secret" + + attestation = json.loads(out.read_text())["attestation"] + assert attestation["verdictSource"] == "server-attested" + assert attestation["attested"] is True + assert attestation["server"] == "http://honua.test" + assert attestation["fallbackReason"] is None + # The standalone attestation artifact matches the embedded one. + assert json.loads(attested_out.read_text()) == attestation + assert "server-attested by http://honua.test" in capsys.readouterr().err + + +def test_cli_api_key_falls_back_to_the_environment(tmp_path: Path, monkeypatch) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=_report_for(request)) + + seen = _attesting_admin_client(monkeypatch, handler) + monkeypatch.setenv("HONUA_ADMIN_API_KEY", "from-env") + + rc = main(["pyt", str(toolbox), "--output", str(tmp_path / "tb.json"), "--server", "http://honua.test"]) + + assert rc == 0 + assert seen["api_key"] == "from-env" + + +def test_cli_surfaces_a_server_vs_local_disagreement(tmp_path: Path, monkeypatch, capsys) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + + def handler(request: httpx.Request) -> httpx.Response: + # The SDK classified tool A as translatable; the catalog disagrees. + return httpx.Response(200, json=_report_for(request, {"A": "unsupported"})) + + _attesting_admin_client(monkeypatch, handler) + + rc = main(["pyt", str(toolbox), "--output", str(out), "--server", "http://honua.test"]) + + assert rc == 0 + attestation = json.loads(out.read_text())["attestation"] + assert attestation["summary"]["disagreementCount"] == 1 + assert attestation["disagreements"] == [{"toolName": "A", "local": "translated", "server": "unsupported"}] + # The server's verdict is the effective one. + assert attestation["tools"][0]["classification"] == "unsupported" + assert attestation["tools"][0]["localClassification"] == "translated" + assert "disagreement: A local=translated server=unsupported (server wins)" in capsys.readouterr().err + + +def test_cli_unreachable_server_degrades_to_local_only(tmp_path: Path, monkeypatch, capsys) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused") + + _attesting_admin_client(monkeypatch, handler) + + rc = main(["pyt", str(toolbox), "--output", str(out), "--server", "http://offline.test"]) + + # An unreachable server is not a hard failure: the local report still emits. + assert rc == 0 + attestation = json.loads(out.read_text())["attestation"] + assert attestation["attested"] is False + assert attestation["verdictSource"] == "local-only" + assert "connection refused" in attestation["fallbackReason"] + assert attestation["tools"][0]["classification"] == "translated" + assert "local-only (NOT server-attested)" in capsys.readouterr().err + + +def test_cli_unauthorized_call_never_claims_attestation(tmp_path: Path, monkeypatch, capsys) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(401, json={"error": "Unauthorized"}) + + _attesting_admin_client(monkeypatch, handler) + + rc = main(["pyt", str(toolbox), "--output", str(out), "--server", "http://honua.test"]) + + assert rc == 0 + attestation = json.loads(out.read_text())["attestation"] + assert attestation["attested"] is False + assert attestation["verdictSource"] == "local-only" + assert attestation["fallbackReason"] + assert all(tool["serverClassification"] is None for tool in attestation["tools"]) + assert "local-only (NOT server-attested)" in capsys.readouterr().err + + +def test_cli_require_attested_fails_when_the_call_is_refused(tmp_path: Path, monkeypatch, capsys) -> None: + import httpx + + from honua_sdk.migration._cli import EXIT_NOT_ATTESTED + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(403, json={"error": "Forbidden"}) + + _attesting_admin_client(monkeypatch, handler) + + rc = main( + ["pyt", str(toolbox), "--output", str(out), "--server", "http://honua.test", "--require-attested"] + ) + + assert rc == EXIT_NOT_ATTESTED + assert "--require-attested" in capsys.readouterr().err + # The report is still written, marked local-only, so the failure is diagnosable. + assert json.loads(out.read_text())["attestation"]["attested"] is False + + +def test_cli_require_attested_offline_fails_without_contacting_a_server(tmp_path: Path) -> None: + from honua_sdk.migration._cli import EXIT_NOT_ATTESTED + + toolbox = _write(tmp_path, "tb.pyt", PYT) + + rc = main(["pyt", str(toolbox), "--output", str(tmp_path / "tb.json"), "--require-attested"]) + + assert rc == EXIT_NOT_ATTESTED + + +def test_cli_attestation_degrades_when_honua_admin_is_not_installed(tmp_path: Path, monkeypatch, capsys) -> None: + import builtins + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + real_import = builtins.__import__ + + def blocked_import(name, *args, **kwargs): + if name == "honua_admin": + raise ImportError("No module named 'honua_admin'") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", blocked_import) + + rc = main(["pyt", str(toolbox), "--output", str(out), "--server", "http://honua.test"]) + + # honua-admin is an optional dependency: without it the toolbox still + # translates, it just cannot be attested. + assert rc == 0 + attestation = json.loads(out.read_text())["attestation"] + assert attestation["attested"] is False + assert "honua_admin" in attestation["fallbackReason"] + + +def test_cli_atbx_with_server_attests_model_steps(tmp_path: Path, monkeypatch) -> None: + import httpx + + atbx = _write_atbx(tmp_path / "wf.atbx") + out = tmp_path / "wf.json" + bodies: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + bodies.append(json.loads(request.content.decode("utf-8"))) + return httpx.Response(200, json=_report_for(request)) + + _attesting_admin_client(monkeypatch, handler) + + rc = main(["atbx", str(atbx), "--output", str(out), "--server", "http://honua.test"]) + + assert rc == 0 + assert bodies[0]["sourceFormat"] == "atbx" + assert [tool["toolName"] for tool in bodies[0]["tools"]] == ["BufferModel"] + assert bodies[0]["tools"][0]["targetProcessId"] == "geometry.buffer" + assert json.loads(out.read_text())["attestation"]["verdictSource"] == "server-attested" + + +def test_cli_translate_routes_a_pyt_toolbox_through_attestation(tmp_path: Path, monkeypatch) -> None: + import httpx + + toolbox = _write(tmp_path, "tb.pyt", PYT) + out = tmp_path / "tb.json" + bodies: list[dict] = [] + + def handler(request: httpx.Request) -> httpx.Response: + bodies.append(json.loads(request.content.decode("utf-8"))) + return httpx.Response(200, json=_report_for(request)) + + _attesting_admin_client(monkeypatch, handler) + + rc = main(["translate", str(toolbox), "--output", str(out), "--server", "http://honua.test"]) + + assert rc == 0 + assert bodies[0]["sourceFormat"] == "pyt" + document = json.loads(out.read_text()) + assert document["attestation"]["verdictSource"] == "server-attested" + # The submitted manifest travels with the plan so the report is reproducible. + assert document["translationManifest"]["artifactKind"] == "honua.migration.toolbox-translation" + + +def test_cli_translate_rejects_attesting_a_bare_arcpy_script(tmp_path: Path, capsys) -> None: + script = _write(tmp_path, "wf.py", SCRIPT) + + rc = main(["translate", str(script), "--server", "http://honua.test"]) + + # Refusing beats inventing a toolbox source format the server would reject. + assert rc == 2 + assert "is a script rather than a toolbox" in capsys.readouterr().err + + +def test_cli_translate_of_a_binary_tbx_gives_export_instructions(tmp_path: Path, capsys) -> None: + binary = tmp_path / "legacy.tbx" + binary.write_bytes(b"\x00binary\x00") + + rc = main(["translate", str(binary)]) + + assert rc == 3 + err = capsys.readouterr().err + # The refusal is policy, so it has to read as a migration instruction. + assert "Export the toolbox to an open format first" in err + assert "New ArcGIS Toolbox (.atbx)" in err + assert "deliberately never parsed" in err diff --git a/tests/test_arcpy_migration_modelbuilder.py b/tests/test_arcpy_migration_modelbuilder.py index fb79ad9..b59aab4 100644 --- a/tests/test_arcpy_migration_modelbuilder.py +++ b/tests/test_arcpy_migration_modelbuilder.py @@ -400,3 +400,6 @@ def test_atbx_model_detected_by_tool_type(tmp_path) -> None: toolbox = parse_atbx_toolbox(atbx) assert toolbox.models == () assert toolbox.parse_error is None + # Excluded from models (nothing to translate) but still discoverable, so a + # consumer cannot under-count the toolbox (honua-sdk-python#188). + assert toolbox.unresolved_tool_names == ("EmptyModel",) diff --git a/tests/test_arcpy_migration_pyt.py b/tests/test_arcpy_migration_pyt.py index 7dd9b68..bdd031b 100644 --- a/tests/test_arcpy_migration_pyt.py +++ b/tests/test_arcpy_migration_pyt.py @@ -5,6 +5,7 @@ import pytest from honua_sdk.migration import ( + BINARY_TOOLBOX_EXPORT_GUIDANCE, UnsupportedToolboxError, build_pyt_parity_evidence, parse_binary_toolbox, @@ -239,3 +240,27 @@ def test_parse_binary_toolbox_still_stubs_binary_tbx(tmp_path) -> None: with pytest.raises(UnsupportedToolboxError) as excinfo: parse_binary_toolbox(tmp_path / "legacy.tbx") assert ".tbx" in str(excinfo.value) + + +def test_binary_tbx_refusal_carries_concrete_export_steps(tmp_path) -> None: + # Refusing to parse the proprietary container is a standing policy decision, + # not an unfinished stub, so the error has to read as a migration + # instruction rather than a dead end (honua-sdk-python#188). + with pytest.raises(UnsupportedToolboxError) as excinfo: + parse_binary_toolbox(tmp_path / "legacy.tbx") + + message = str(excinfo.value) + assert BINARY_TOOLBOX_EXPORT_GUIDANCE in message + assert "ArcGIS Pro" in message + assert "New ArcGIS Toolbox (.atbx)" in message + assert "parse_pyt_file" in message + assert "deliberately never parsed" in message + + +def test_atbx_reader_shares_the_same_binary_tbx_guidance(tmp_path) -> None: + from honua_sdk.migration import UnsupportedModelFormatError, parse_atbx_toolbox + + with pytest.raises(UnsupportedModelFormatError) as excinfo: + parse_atbx_toolbox(tmp_path / "legacy.tbx") + + assert BINARY_TOOLBOX_EXPORT_GUIDANCE in str(excinfo.value)