diff --git a/apps/api/src/cora/agent/adapters/postgres_language_model_lookup.py b/apps/api/src/cora/agent/adapters/postgres_language_model_lookup.py index 92265c0ceac..a9e97c79872 100644 --- a/apps/api/src/cora/agent/adapters/postgres_language_model_lookup.py +++ b/apps/api/src/cora/agent/adapters/postgres_language_model_lookup.py @@ -8,6 +8,21 @@ and deprecating a mistaken duplicate restores the previous Approved entry. The `language_model_id DESC` tiebreak makes equal-created_at rows deterministic. + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `LanguageModelLookupResult` (to keep +`cora.infrastructure.ports.language_model_lookup` import-free of +Agent BC types). The adapter constructs +`LanguageModelStatus(row["status"])` as a validation step: a +corrupted row whose `status` is not a known enum value surfaces as +`ValueError` from the adapter rather than as a silent wrong-status +match downstream. `.value` on the validated member narrows to +exactly the port's alias, so no cast is needed. The SQL filter +narrows to `'Approved'`; the alias still pins the full five-value +enum because a query filter is adapter behavior, not a constraint on +the field's type. """ from __future__ import annotations @@ -15,6 +30,7 @@ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false from typing import TYPE_CHECKING +from cora.agent.aggregates.language_model import LanguageModelStatus from cora.infrastructure.ports.language_model_lookup import LanguageModelLookupResult if TYPE_CHECKING: @@ -47,7 +63,7 @@ async def find_by_model( return None return LanguageModelLookupResult( language_model_id=row["language_model_id"], - status=row["status"], + status=LanguageModelStatus(row["status"]).value, data_tier=row["data_tier"], archivability=row["archivability"], snapshot_pin=row["snapshot_pin"], diff --git a/apps/api/src/cora/data/adapters/postgres_dataset_distribution_lookup.py b/apps/api/src/cora/data/adapters/postgres_dataset_distribution_lookup.py index 7447adf9101..8543496e94e 100644 --- a/apps/api/src/cora/data/adapters/postgres_dataset_distribution_lookup.py +++ b/apps/api/src/cora/data/adapters/postgres_dataset_distribution_lookup.py @@ -13,6 +13,19 @@ not a single canonical row. Lives in `cora.data.adapters` because it reads a Data-owned projection; it implements the infrastructure port so the Run BC consumes it without importing anything Data-internal. + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias on the +port's `DatasetDistributionLookupResult` (to keep +`cora.infrastructure.ports.dataset_distribution_lookup` import-free of Data +BC types). The adapter constructs `DistributionStatus(row["status"])` as a +validation step: a corrupted row whose `status` is not a known enum value +surfaces as `ValueError` from the adapter rather than as a silent +wrong-status match downstream. `.value` on the validated member narrows to +exactly the port's alias, so no cast is needed. The SQL filter excludes +Discarded rows; the alias still pins the full four-value enum because a +query filter is adapter behavior, not a constraint on the field's type. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -22,6 +35,7 @@ import asyncpg +from cora.data.aggregates.distribution import DistributionStatus from cora.infrastructure.ports.dataset_distribution_lookup import ( DatasetDistributionLookupResult, ) @@ -55,7 +69,7 @@ async def find_by_datasets( distribution_id=row["distribution_id"], dataset_id=row["dataset_id"], supply_id=row["supply_id"], - status=row["status"], + status=DistributionStatus(row["status"]).value, ) ) return {dataset_id: tuple(results) for dataset_id, results in grouped.items()} diff --git a/apps/api/src/cora/enclosure/adapters/postgres_enclosure_lookup.py b/apps/api/src/cora/enclosure/adapters/postgres_enclosure_lookup.py index 0369b0ce6ec..3b69dba7a02 100644 --- a/apps/api/src/cora/enclosure/adapters/postgres_enclosure_lookup.py +++ b/apps/api/src/cora/enclosure/adapters/postgres_enclosure_lookup.py @@ -38,8 +38,11 @@ `EnclosureLifecycle(row["lifecycle"])` as a validation step: a corrupted row whose value is not a known enum surfaces as `ValueError` from the adapter rather than as a silent wrong-status -match downstream. Both validated `StrEnum` values are `IS-A str`, -so assignment into the dataclass's `str`-typed fields is exact. +match downstream. The port's fields are `Literal` aliases rather +than bare `str`, and `.value` on a validated member narrows to +exactly that alias, so no cast is needed: the constructor rejects +an unknown value at runtime and the type checker confirms the +remainder statically. ## Timestamp coercion @@ -140,8 +143,8 @@ def _row_to_reference(row: Any) -> EnclosureLookupResult: return EnclosureLookupResult( enclosure_id=row["enclosure_id"], name=str(row["name"]), - permit_status=EnclosurePermitStatus(row["permit_status"]), - lifecycle=EnclosureLifecycle(row["lifecycle"]), + permit_status=EnclosurePermitStatus(row["permit_status"]).value, + lifecycle=EnclosureLifecycle(row["lifecycle"]).value, permit_status_changed_at=_format_changed_at(row["last_permit_status_changed_at"]), source_kind=row["last_source_kind"], source_id=row["last_source_id"], diff --git a/apps/api/src/cora/equipment/adapters/postgres_assembly_lookup.py b/apps/api/src/cora/equipment/adapters/postgres_assembly_lookup.py index d40d9436003..65231fec18d 100644 --- a/apps/api/src/cora/equipment/adapters/postgres_assembly_lookup.py +++ b/apps/api/src/cora/equipment/adapters/postgres_assembly_lookup.py @@ -5,6 +5,17 @@ handler so the role_kind satisfaction check ORs-in the Assembly path on top of the Family disjunction (see [[project-role-aggregate-design]] sub-slice 3C/3D for the worked Microscope-Assembly example). + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `AssemblyLookupResult` (to keep +`cora.infrastructure.ports.assembly_lookup` import-free of Equipment +BC types). The adapter constructs `AssemblyStatus(row["status"])` as +a validation step: a corrupted row whose `status` is not a known +enum value surfaces as `ValueError` from the adapter rather than as +a silent wrong-status match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -14,6 +25,7 @@ import asyncpg +from cora.equipment.aggregates.assembly import AssemblyStatus from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult _LOOKUP_SQL = """ @@ -42,7 +54,7 @@ def _row_to_result(row: Any) -> AssemblyLookupResult: return AssemblyLookupResult( id=row["assembly_id"], name=str(row["name"]), - status=str(row["status"]), + status=AssemblyStatus(row["status"]).value, presents_as=frozenset(row["presents_as"] or ()), ) diff --git a/apps/api/src/cora/equipment/adapters/postgres_asset_lookup.py b/apps/api/src/cora/equipment/adapters/postgres_asset_lookup.py index 877488e62d6..7a7728a91c9 100644 --- a/apps/api/src/cora/equipment/adapters/postgres_asset_lookup.py +++ b/apps/api/src/cora/equipment/adapters/postgres_asset_lookup.py @@ -37,15 +37,14 @@ ## Enum coercion `tier` and `lifecycle` are stored as `TEXT` columns and typed as -`str` on the port's `AssetLookupResult` (to keep +`Literal` aliases on the port's `AssetLookupResult` (to keep `cora.infrastructure.ports.asset_lookup` import-free of Equipment BC types). The adapter still constructs `AssetTier(row["tier"])` / `AssetLifecycle(row["lifecycle"])` as a validation step: a corrupted row whose `tier` or `lifecycle` is not a known enum value surfaces as `ValueError` from the adapter rather than as a -silent wrong-tier match downstream. The validated `StrEnum` value -IS-A `str`, so the assignment into the dataclass's `str`-typed -fields is exact. +silent wrong-tier match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -194,8 +193,8 @@ def _row_to_result(row: Any) -> AssetLookupResult: return AssetLookupResult( id=row["asset_id"], name=str(row["name"]), - tier=AssetTier(row["tier"]), - lifecycle=AssetLifecycle(row["lifecycle"]), + tier=AssetTier(row["tier"]).value, + lifecycle=AssetLifecycle(row["lifecycle"]).value, family_affordances=frozenset(str(a) for a in row["family_affordances"]), located_in_enclosure_id=row["located_in_enclosure_id"], ) diff --git a/apps/api/src/cora/equipment/adapters/postgres_family_lookup.py b/apps/api/src/cora/equipment/adapters/postgres_family_lookup.py index 9599efe6588..30f7b2f8dad 100644 --- a/apps/api/src/cora/equipment/adapters/postgres_family_lookup.py +++ b/apps/api/src/cora/equipment/adapters/postgres_family_lookup.py @@ -4,6 +4,17 @@ `Kernel.family_lookup` port. The 3B slice ships this adapter; 3D wires its `bind_plan_role` handler against it for the role_kind satisfaction-check path (Lock 17 ANY-single-family disjunction). + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `FamilyLookupResult` (to keep +`cora.infrastructure.ports.family_lookup` import-free of Equipment +BC types). The adapter constructs `FamilyStatus(row["status"])` as a +validation step: a corrupted row whose `status` is not a known enum +value surfaces as `ValueError` from the adapter rather than as a +silent wrong-status match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -13,6 +24,7 @@ import asyncpg +from cora.equipment.aggregates.family import FamilyStatus from cora.infrastructure.ports.family_lookup import FamilyLookupResult _LOOKUP_SQL = """ @@ -41,7 +53,7 @@ def _row_to_result(row: Any) -> FamilyLookupResult: return FamilyLookupResult( id=row["family_id"], name=str(row["name"]), - status=str(row["status"]), + status=FamilyStatus(row["status"]).value, affordances=frozenset(row["affordances"] or ()), presents_as=frozenset(row["presents_as"] or ()), ) diff --git a/apps/api/src/cora/federation/adapters/in_memory_permit_lookup.py b/apps/api/src/cora/federation/adapters/in_memory_permit_lookup.py index 190a1ebbe4d..e4e8126c0ff 100644 --- a/apps/api/src/cora/federation/adapters/in_memory_permit_lookup.py +++ b/apps/api/src/cora/federation/adapters/in_memory_permit_lookup.py @@ -10,8 +10,10 @@ from uuid import UUID from cora.infrastructure.ports.federation.permit_lookup import ( + AbiTierValue, PermitLookup, PermitLookupResult, + PermitStatusValue, ) from cora.shared.facility_code import FacilityCode @@ -62,8 +64,8 @@ def register_outbound( peer_facility_id: str | FacilityCode, artifact_kind: str, permit_id: UUID, - status: str = "Active", - abi_tier_floor: str = "Stable", + status: PermitStatusValue = "Active", + abi_tier_floor: AbiTierValue = "Stable", current_version: int = 0, ) -> PermitLookupResult: """Convenience: seed an outbound permit; returns the seeded result for assertions.""" @@ -90,8 +92,8 @@ def register_inbound( peer_facility_id: str | FacilityCode, artifact_kind: str, permit_id: UUID, - status: str = "Active", - abi_tier_floor: str = "Stable", + status: PermitStatusValue = "Active", + abi_tier_floor: AbiTierValue = "Stable", current_version: int = 0, ) -> PermitLookupResult: """Convenience: seed an inbound permit; returns the seeded result for assertions.""" diff --git a/apps/api/src/cora/federation/adapters/postgres_credential_lookup.py b/apps/api/src/cora/federation/adapters/postgres_credential_lookup.py index ad3f8addff7..82435689cee 100644 --- a/apps/api/src/cora/federation/adapters/postgres_credential_lookup.py +++ b/apps/api/src/cora/federation/adapters/postgres_credential_lookup.py @@ -26,16 +26,19 @@ ## Enum coercion -`purpose` and `status` are stored as `TEXT` columns and are -typed as `str` on the port's `CredentialLookupResult` (to keep +`purpose` and `status` are stored as `TEXT` columns. `purpose` stays +typed `str` on the port's `CredentialLookupResult` (to keep `cora.infrastructure.ports.credential_lookup` import-free of -Federation BC types). The adapter still constructs -`CredentialPurpose(row["purpose"])` / `CredentialStatus(row["status"])` -as a validation step: a corrupted row whose `purpose` or `status` -is not a known enum value surfaces as `ValueError` from the -adapter rather than as a silent wrong-purpose match downstream. -The validated `StrEnum` value is `IS-A str`, so the assignment -into the dataclass's `str`-typed fields is exact. +Federation BC types); the adapter constructs +`CredentialPurpose(row["purpose"])` as a validation step, and the +validated `StrEnum` value IS-A `str`, so the assignment into the +`str`-typed field is exact. `status` is typed as a `Literal` alias +pinning `CredentialStatus`'s value set; the adapter constructs +`CredentialStatus(row["status"])` the same way, but takes `.value` +so the result narrows to exactly the alias, no cast needed. Either +way, a corrupted row whose `purpose` or `status` is not a known +enum value surfaces as `ValueError` from the adapter rather than as +a silent wrong-value match downstream. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -79,7 +82,7 @@ def _row_to_result(row: Any) -> CredentialLookupResult: id=row["credential_id"], facility_id=FacilityCode(str(row["facility_id"])), purpose=CredentialPurpose(row["purpose"]), - status=CredentialStatus(row["status"]), + status=CredentialStatus(row["status"]).value, ) diff --git a/apps/api/src/cora/federation/adapters/postgres_facility_lookup.py b/apps/api/src/cora/federation/adapters/postgres_facility_lookup.py index 5a67f2a588c..0ae0b5a3e9f 100644 --- a/apps/api/src/cora/federation/adapters/postgres_facility_lookup.py +++ b/apps/api/src/cora/federation/adapters/postgres_facility_lookup.py @@ -33,15 +33,18 @@ ## Enum coercion -`kind` and `status` are stored as `TEXT` columns and are typed as +`kind` and `status` are stored as `TEXT` columns. `kind` stays typed `str` on the port's `FacilityLookupResult` (to keep `cora.infrastructure.ports.facility_lookup` import-free of Federation -BC types). The adapter still constructs `FacilityKind(row["kind"])` / -`FacilityStatus(row["status"])` as a validation step: a corrupted -row whose `kind` or `status` is not a known enum value surfaces as -`ValueError` from the adapter rather than as a silent wrong-kind -match downstream. The validated `StrEnum` value is `IS-A str`, so -the assignment into the dataclass's `str`-typed fields is exact. +BC types); the adapter constructs `FacilityKind(row["kind"])` as a +validation step, and the validated `StrEnum` value IS-A `str`, so +the assignment into the `str`-typed field is exact. `status` is +typed as a `Literal` alias pinning `FacilityStatus`'s value set; the +adapter constructs `FacilityStatus(row["status"])` the same way, but +takes `.value` so the result narrows to exactly the alias, no cast +needed. Either way, a corrupted row whose `kind` or `status` is not +a known enum value surfaces as `ValueError` from the adapter rather +than as a silent wrong-value match downstream. ## JSONB array decoding @@ -129,8 +132,8 @@ def _row_to_result(row: Any) -> FacilityLookupResult: return FacilityLookupResult( id=row["facility_id"], code=FacilityCode(str(row["code"])), - kind=FacilityKind(row["kind"]), - status=FacilityStatus(row["status"]), + kind=FacilityKind(row["kind"]).value, + status=FacilityStatus(row["status"]).value, trust_anchor_credential_ids=_decode_trust_anchor_ids(row["trust_anchor_credential_ids"]), ) diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_assembly_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_assembly_lookup.py index 29a348ef35f..5aef6320d4f 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_assembly_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_assembly_lookup.py @@ -13,7 +13,7 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult +from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult, AssemblyStatusValue class InMemoryAssemblyLookup: @@ -30,7 +30,7 @@ def register( self, assembly_id: UUID, name: str, - status: str = "Defined", + status: AssemblyStatusValue = "Defined", presents_as: Iterable[UUID] = (), ) -> None: """Test helper: install an Assembly summary keyed by `assembly_id`.""" diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_asset_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_asset_lookup.py index 16625774f2b..f7ea63140e1 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_asset_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_asset_lookup.py @@ -18,7 +18,9 @@ from cora.infrastructure.ports.asset_lookup import ( ANCESTOR_WALK_DEPTH_CAP, AncestorWalkDepthExceededError, + AssetLifecycleValue, AssetLookupResult, + AssetTierValue, ) @@ -37,8 +39,8 @@ def register( self, asset_id: UUID, name: str, - tier: str = "Unit", - lifecycle: str = "Active", + tier: AssetTierValue = "Unit", + lifecycle: AssetLifecycleValue = "Active", family_affordances: frozenset[str] | None = None, parent_id: UUID | None = None, located_in_enclosure_id: UUID | None = None, diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_clearance_template_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_clearance_template_lookup.py index 8c5a5014051..8c1d40d1f51 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_clearance_template_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_clearance_template_lookup.py @@ -13,7 +13,10 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.clearance_template_lookup import ClearanceTemplateLookupResult +from cora.infrastructure.ports.clearance_template_lookup import ( + ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, +) class InMemoryClearanceTemplateLookup: @@ -34,7 +37,7 @@ def register( *, facility_code: str = "aps", code: str = "default-template", - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> None: """Test helper: install a clearance-template summary keyed by `template_id`. diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_credential_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_credential_lookup.py index 32ecf9cf4a5..85760e2cce0 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_credential_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_credential_lookup.py @@ -13,7 +13,10 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.credential_lookup import CredentialLookupResult +from cora.infrastructure.ports.credential_lookup import ( + CredentialLookupResult, + CredentialStatusValue, +) from cora.shared.facility_code import FacilityCode @@ -32,7 +35,7 @@ def register( credential_id: UUID, facility_id: str | FacilityCode, purpose: str, - status: str, + status: CredentialStatusValue, ) -> None: """Test helper: install a credential summary keyed by `credential_id`. diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_enclosure_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_enclosure_lookup.py index 4d30af216f3..dc1313c09a8 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_enclosure_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_enclosure_lookup.py @@ -20,7 +20,11 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosureLookupResult, + EnclosurePermitStatusValue, +) class InMemoryEnclosureLookup: @@ -41,8 +45,8 @@ def register( self, enclosure_id: UUID, name: str, - permit_status: str = "Permitted", - lifecycle: str = "Active", + permit_status: EnclosurePermitStatusValue = "Permitted", + lifecycle: EnclosureLifecycleValue = "Active", permit_status_changed_at: str | None = None, source_kind: str | None = None, source_id: str | None = None, diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_facility_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_facility_lookup.py index f95832018b7..609c783c3d3 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_facility_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_facility_lookup.py @@ -14,7 +14,11 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.facility_lookup import FacilityLookupResult +from cora.infrastructure.ports.facility_lookup import ( + FacilityKindValue, + FacilityLookupResult, + FacilityStatusValue, +) from cora.shared.facility_code import FacilityCode @@ -35,8 +39,8 @@ def register( self, facility_id: UUID, code: str | FacilityCode, - kind: str, - status: str = "Active", + kind: FacilityKindValue, + status: FacilityStatusValue = "Active", trust_anchor_credential_ids: frozenset[UUID] = frozenset(), ) -> None: """Test helper: install a facility summary keyed by `facility_id`. diff --git a/apps/api/src/cora/infrastructure/adapters/in_memory_family_lookup.py b/apps/api/src/cora/infrastructure/adapters/in_memory_family_lookup.py index 5f7f886409e..65bd29e7a7e 100644 --- a/apps/api/src/cora/infrastructure/adapters/in_memory_family_lookup.py +++ b/apps/api/src/cora/infrastructure/adapters/in_memory_family_lookup.py @@ -13,7 +13,7 @@ from threading import Lock from uuid import UUID -from cora.infrastructure.ports.family_lookup import FamilyLookupResult +from cora.infrastructure.ports.family_lookup import FamilyLookupResult, FamilyStatusValue class InMemoryFamilyLookup: @@ -30,7 +30,7 @@ def register( self, family_id: UUID, name: str, - status: str = "Defined", + status: FamilyStatusValue = "Defined", affordances: Iterable[str] = (), presents_as: Iterable[UUID] = (), ) -> None: diff --git a/apps/api/src/cora/infrastructure/ports/assembly_lookup.py b/apps/api/src/cora/infrastructure/ports/assembly_lookup.py index bcccb8b4ace..f77d440ce94 100644 --- a/apps/api/src/cora/infrastructure/ports/assembly_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/assembly_lookup.py @@ -22,9 +22,13 @@ ## No BC imports in the port -`status` is typed `str` (not the `AssemblyStatus` StrEnum) so this -port stays inside `cora.infrastructure`'s `depends_on = []` tach -contract. Values match the StrEnum string values; consumers +`status` is a `Literal` alias rather than bare `str` (not the +`AssemblyStatus` StrEnum), so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so the alias pins the enum's value set without +importing the enum. A fitness test pins the alias to the enum and +fails if the two ever drift. Adapters produce a member's `.value`, +which narrows to exactly the alias, so no cast is needed. Consumers partition on the literal if they want to distinguish Defined / Versioned / Deprecated. @@ -42,9 +46,11 @@ """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +AssemblyStatusValue = Literal["Defined", "Versioned", "Deprecated"] + @dataclass(frozen=True) class AssemblyLookupResult: @@ -58,10 +64,10 @@ class AssemblyLookupResult: incrementally via `add_assembly_presents_as` / removes via `remove_assembly_presents_as`. - `status` is the FSM stage as a plain string ("Defined" / - "Versioned" / "Deprecated"); the bind_plan_role decider accepts - every status (mirrors the Family-path posture: deprecation is - advisory, not blocking). + `status` is a `Literal` alias pinning `AssemblyStatus`'s value + set ("Defined" / "Versioned" / "Deprecated"); the bind_plan_role + decider accepts every status (mirrors the Family-path posture: + deprecation is advisory, not blocking). `name` is the operator-readable display name; useful for surfacing in cross-BC error messages. @@ -69,7 +75,7 @@ class AssemblyLookupResult: id: UUID name: str - status: str + status: AssemblyStatusValue presents_as: frozenset[UUID] @@ -93,4 +99,4 @@ async def lookup(self, assembly_id: UUID) -> AssemblyLookupResult | None: ... -__all__ = ["AssemblyLookup", "AssemblyLookupResult"] +__all__ = ["AssemblyLookup", "AssemblyLookupResult", "AssemblyStatusValue"] diff --git a/apps/api/src/cora/infrastructure/ports/asset_lookup.py b/apps/api/src/cora/infrastructure/ports/asset_lookup.py index db3d6b27c50..942c1d9a3fe 100644 --- a/apps/api/src/cora/infrastructure/ports/asset_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/asset_lookup.py @@ -38,12 +38,16 @@ ## No BC imports in the port -`tier` and `lifecycle` are typed as `str` (not Equipment BC's -`AssetTier` / `AssetLifecycle` StrEnums) so this port stays inside -`cora.infrastructure`'s `depends_on = []` tach contract. The values -match the StrEnum string values; consumer deciders partition by -literal comparison (`tier == "Unit"`, `lifecycle == "Active"`) and -cast to typed enums at their boundary if they need the discipline. +`tier` and `lifecycle` are `Literal` aliases rather than bare `str` +(not Equipment BC's `AssetTier` / `AssetLifecycle` StrEnums), so this +port stays inside `cora.infrastructure`'s `depends_on = []` tach +contract: `Literal` comes from `typing`, so the alias pins each +enum's value set without importing the enum. A fitness test pins +each alias to its owning enum and fails if the two ever drift. +Adapters produce a member's `.value`, which narrows to exactly the +alias, so no cast is needed. Consumer deciders partition by literal +comparison (`tier == "Unit"`, `lifecycle == "Active"`) and cast to +typed enums at their boundary if they need the discipline. `id` is typed `UUID` (Equipment BC's Asset.id is bare UUID, not a NewType, so no cross-BC NewType to thread). Consumers that care @@ -51,9 +55,12 @@ """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +AssetTierValue = Literal["Unit", "Component", "Device"] +AssetLifecycleValue = Literal["Commissioned", "Active", "Maintenance", "Decommissioned"] + ANCESTOR_WALK_DEPTH_CAP = 50 """Maximum `parent_id` chain depth `ancestors_of` walks before failing. @@ -79,9 +86,10 @@ class AssetLookupResult: via `AssetLookup.lookup` and handed to the decider in the slice's context object (mirrors `FacilityLookupResult` shape). - `tier` and `lifecycle` are the StrEnum values as plain strings - (matches the projection's `TEXT` columns); the consumer decider - partitions on the literals it cares about. + `tier` and `lifecycle` are `Literal` aliases pinning + `AssetTier` / `AssetLifecycle`'s value sets (matches the + projection's `TEXT` columns); the consumer decider partitions on + the literals it cares about. `name` is the operator-readable display name (1-200 chars per `AssetName` VO); useful for surfacing in cross-BC error messages @@ -124,8 +132,8 @@ class AssetLookupResult: id: UUID name: str - tier: str - lifecycle: str + tier: AssetTierValue + lifecycle: AssetLifecycleValue family_affordances: frozenset[str] located_in_enclosure_id: UUID | None = None @@ -219,6 +227,8 @@ async def ancestors_of(self, asset_ids: frozenset[UUID]) -> frozenset[AssetLooku __all__ = [ "ANCESTOR_WALK_DEPTH_CAP", "AncestorWalkDepthExceededError", + "AssetLifecycleValue", "AssetLookup", "AssetLookupResult", + "AssetTierValue", ] diff --git a/apps/api/src/cora/infrastructure/ports/capability_lookup.py b/apps/api/src/cora/infrastructure/ports/capability_lookup.py index 5402918d94a..d3f611bd0c5 100644 --- a/apps/api/src/cora/infrastructure/ports/capability_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/capability_lookup.py @@ -28,12 +28,27 @@ vocabulary, without Equipment's domain code learning the word. The handler maps `CapabilityLookupResult` to Equipment's local `CapabilityView` response type. + +## No BC imports in the port + +`status` is a `Literal` alias rather than bare `str` (not Recipe +BC's `CapabilityStatus` StrEnum), so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so the alias pins the enum's value set without +importing the enum. A fitness test pins the alias to the enum and +fails if the two ever drift. The alias mirrors the FULL +`CapabilityStatus` value set (`Defined`, `Versioned`, `Deprecated`) +even though `find_applicable_by_affordances` filters to +`{Defined, Versioned}`: the SQL filter is adapter behavior, not a +constraint on what the field can hold. """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +CapabilityStatusValue = Literal["Defined", "Versioned", "Deprecated"] + @dataclass(frozen=True) class CapabilityLookupResult: @@ -43,12 +58,18 @@ class CapabilityLookupResult: map onto its public `CapabilityView` response shape. Adding fields to this dataclass is a port-version bump that touches the adapter plus every consumer. + + `status` is a `Literal` alias pinning `CapabilityStatus`'s full + value set; `find_applicable_by_affordances` filters to + `{Defined, Versioned}` at the adapter, so every row this port + actually returns today carries one of those two values, but the + field itself can hold any status the enum has. """ capability_id: UUID code: str name: str - status: str # "Defined" | "Versioned" + status: CapabilityStatusValue class CapabilityLookup(Protocol): diff --git a/apps/api/src/cora/infrastructure/ports/clearance_lookup.py b/apps/api/src/cora/infrastructure/ports/clearance_lookup.py index 04412139c67..b8fc937d15e 100644 --- a/apps/api/src/cora/infrastructure/ports/clearance_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/clearance_lookup.py @@ -41,14 +41,35 @@ column today; adding one needs a side-table or jsonb column. Defer until a concrete consumer (for example, a proposal-issued ExternalBinding-only Clearance) trips on the gap. + +## No BC imports in the port + +`status` is a `Literal` alias rather than bare `str` (not Safety +BC's `ClearanceStatus` StrEnum), so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so the alias pins the enum's value set without +importing the enum. A fitness test pins the alias to the enum and +fails if the two ever drift. Adapters produce a member's `.value`, +which narrows to exactly the alias, so no cast is needed. """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID from cora.infrastructure.routing import NIL_SENTINEL_ID +ClearanceStatusValue = Literal[ + "Defined", + "Submitted", + "UnderReview", + "Approved", + "Active", + "Expired", + "Rejected", + "Superseded", +] + @dataclass(frozen=True) class ClearanceLookupResult: @@ -59,13 +80,13 @@ class ClearanceLookupResult: `ClearanceLookup.find_covering` and handed to the decider in `RunStartContext.referencing_clearances`. - `status` is the StrEnum value as a plain string (matches the - projection's `TEXT` column); the decider treats it opaquely and - partitions on `"Active"`. + `status` is a `Literal` alias pinning `ClearanceStatus`'s value + set (matches the projection's `TEXT` column); the decider treats + it opaquely and partitions on `"Active"`. """ clearance_id: UUID - status: str + status: ClearanceStatusValue template_id: UUID template_code: str facility_code: str diff --git a/apps/api/src/cora/infrastructure/ports/clearance_template_lookup.py b/apps/api/src/cora/infrastructure/ports/clearance_template_lookup.py index d082657f8f7..f543eaa00f6 100644 --- a/apps/api/src/cora/infrastructure/ports/clearance_template_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/clearance_template_lookup.py @@ -39,13 +39,16 @@ ## No BC imports in the port -`status` is typed `str` (not Safety BC's `ClearanceTemplateStatus` -StrEnum) and `facility_code` is typed `str` (not Federation BC's -`FacilityCode` value object) so this port stays inside -`cora.infrastructure`'s `depends_on = []` tach contract. The values -match the StrEnum / VO string values; consumer deciders partition by -literal comparison and cast to typed enums / VOs at their boundary -if they need the discipline. +`status` is a `Literal` alias pinning Safety BC's +`ClearanceTemplateStatus` StrEnum value set (not the StrEnum itself) +and `facility_code` is typed `str` (not Federation BC's +`FacilityCode` value object), so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so the alias pins the enum's value set without +importing the enum. A fitness test pins the alias to the enum and +fails if the two ever drift. `facility_code` values match the VO's +string value; consumer deciders cast to the typed VO at their +boundary if they need the discipline. `id` is typed `UUID` (Safety BC's `ClearanceTemplate.id` is bare UUID, not a NewType, so no cross-BC NewType to thread). Consumers @@ -53,9 +56,11 @@ """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +ClearanceTemplateStatusValue = Literal["Draft", "Active", "Deprecated", "Withdrawn"] + @dataclass(frozen=True) class ClearanceTemplateLookupResult: @@ -67,10 +72,10 @@ class ClearanceTemplateLookupResult: the decider in the slice's context object (mirrors `AssetLookupResult` shape). - `status` is the `ClearanceTemplateStatus` StrEnum value as a - plain string (matches the projection's `TEXT` column); the - consumer decider partitions on the literals it cares about - ("Draft", "Active", "Deprecated", "Withdrawn"). + `status` is a `Literal` alias pinning `ClearanceTemplateStatus`'s + value set (matches the projection's `TEXT` column); the consumer + decider partitions on the literals it cares about ("Draft", + "Active", "Deprecated", "Withdrawn"). `facility_code` is the `FacilityCode` value object's string representation (matches the projection's `TEXT` column); the @@ -89,7 +94,7 @@ class ClearanceTemplateLookupResult: id: UUID facility_code: str code: str - status: str + status: ClearanceTemplateStatusValue version: int @@ -115,4 +120,8 @@ async def lookup(self, template_id: UUID) -> ClearanceTemplateLookupResult | Non ... -__all__ = ["ClearanceTemplateLookup", "ClearanceTemplateLookupResult"] +__all__ = [ + "ClearanceTemplateLookup", + "ClearanceTemplateLookupResult", + "ClearanceTemplateStatusValue", +] diff --git a/apps/api/src/cora/infrastructure/ports/credential_lookup.py b/apps/api/src/cora/infrastructure/ports/credential_lookup.py index 57cdbb6f005..d5087f902e4 100644 --- a/apps/api/src/cora/infrastructure/ports/credential_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/credential_lookup.py @@ -30,11 +30,16 @@ ## No BC imports in the port -`purpose` and `status` are typed as `str` (not the Federation BC's -`CredentialPurpose` / `CredentialStatus` StrEnums) so this port stays -inside `cora.infrastructure`'s `depends_on = []` tach contract. The -values match the StrEnum string values; deciders partition by literal -comparison (`purpose == "SealOnlineSigning"`, `status == "Active"`). +`purpose` stays typed `str` (not Federation BC's `CredentialPurpose` +StrEnum) and `status` is a `Literal` alias pinning Federation BC's +`CredentialStatus` StrEnum value set (not the StrEnum itself), so +this port stays inside `cora.infrastructure`'s `depends_on = []` +tach contract: `Literal` comes from `typing`, so the alias pins the +enum's value set without importing the enum. A fitness test pins +the alias to the enum and fails if the two ever drift. `purpose`'s +values match the StrEnum's string values; deciders partition by +literal comparison (`purpose == "SealOnlineSigning"`, `status == +"Active"`). `facility_id` is typed `FacilityCode` (not bare `str`) per the locked two-tier facility identity design ([[project-structural-scope-design]] @@ -47,11 +52,13 @@ """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID from cora.shared.facility_code import FacilityCode +CredentialStatusValue = Literal["Active", "Rotating", "Revoked"] + @dataclass(frozen=True) class CredentialLookupResult: @@ -62,10 +69,11 @@ class CredentialLookupResult: handler via `CredentialLookup.lookup` and handed to the decider in the seal-slice context object. - `purpose` and `status` are the StrEnum values as plain strings - (matches the projection's `TEXT` columns); the decider partitions - on `purpose == "SealOnlineSigning"` / `"SealOfflineRoot"` and - `status == "Active"`. + `purpose` is the `CredentialPurpose` StrEnum value as a plain + string; `status` is a `Literal` alias pinning `CredentialStatus`'s + value set (both match the projection's `TEXT` columns). The + decider partitions on `purpose == "SealOnlineSigning"` / + `"SealOfflineRoot"` and `status == "Active"`. `facility_id` is a `FacilityCode` value object per the two-tier facility-identity design; the adapter constructs the VO from the @@ -77,7 +85,7 @@ class CredentialLookupResult: id: UUID facility_id: FacilityCode purpose: str - status: str + status: CredentialStatusValue class CredentialLookup(Protocol): diff --git a/apps/api/src/cora/infrastructure/ports/dataset_distribution_lookup.py b/apps/api/src/cora/infrastructure/ports/dataset_distribution_lookup.py index 8f7474fc2b2..98ebebd1028 100644 --- a/apps/api/src/cora/infrastructure/ports/dataset_distribution_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/dataset_distribution_lookup.py @@ -18,17 +18,26 @@ partitions on `status`. It deliberately does NOT reuse the canonical-pick query, whose lowest-id row may be Stale while a higher-id Distribution is Verified. -`status` is the `DistributionStatus` value as a plain string (matches the -projection's TEXT column); `supply_id` is carried for the deferred reachability -check (which Storage Supply / tier the copy rests on); `distribution_id` is -carried for diagnostics and the eventual lineage record. +`status` is a `Literal` alias pinning `DistributionStatus`'s value set +(matches the projection's TEXT column); `Literal` comes from `typing`, so +the alias pins the enum's full four-value set (`Registered`, `Verified`, +`Stale`, `Discarded`) without importing the enum, keeping this port inside +`cora.infrastructure`'s `depends_on = []` tach contract. The adapter's SQL +filters out Discarded rows, but the alias still mirrors the full enum +because a query filter is adapter behavior, not a constraint on the field's +type. A fitness test pins the alias to the enum and fails if the two ever +drift. `supply_id` is carried for the deferred reachability check (which +Storage Supply / tier the copy rests on); `distribution_id` is carried for +diagnostics and the eventual lineage record. """ from collections.abc import Mapping from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +DistributionStatusValue = Literal["Registered", "Verified", "Stale", "Discarded"] + @dataclass(frozen=True) class DatasetDistributionLookupResult: @@ -37,7 +46,7 @@ class DatasetDistributionLookupResult: distribution_id: UUID dataset_id: UUID supply_id: UUID - status: str + status: DistributionStatusValue class DatasetDistributionLookup(Protocol): @@ -101,6 +110,7 @@ async def find_by_datasets( __all__ = [ "DatasetDistributionLookup", "DatasetDistributionLookupResult", + "DistributionStatusValue", "NoDatasetDistributionsLookup", "SeededDatasetDistributionLookup", ] diff --git a/apps/api/src/cora/infrastructure/ports/enclosure_lookup.py b/apps/api/src/cora/infrastructure/ports/enclosure_lookup.py index 08ec5781003..02e718d1351 100644 --- a/apps/api/src/cora/infrastructure/ports/enclosure_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/enclosure_lookup.py @@ -47,26 +47,34 @@ as audit (a tombstoned enclosure can still read `permit_status="Permitted"` from before it was retired). -Both axes reach the port surface as bare `str`. The decider's -gate check is `lifecycle == "Active" AND permit_status == -"Permitted"`. +Both axes reach the port surface as a `Literal` alias pinning the +owning StrEnum's value set. The decider's gate check is +`lifecycle == "Active" AND permit_status == "Permitted"`, and a +comparison against any value outside the alias is a type error at +the call site rather than a silently unreachable branch. ## No BC imports in the port -Every field on `EnclosureLookupResult` is typed as a bare `str` or -bare `UUID` (not the Enclosure BC's `EnclosureId` / -`EnclosurePermitStatus` / `EnclosureLifecycle` types) so this -port stays inside `cora.infrastructure.ports`'s `depends_on = []` -tach contract. The `permit_status` / `lifecycle` / `source_kind` -values match the StrEnum string values; deciders partition by -literal comparison. Enclosure BC callers cast at the boundary if -they want the NewType discipline. +No field on `EnclosureLookupResult` is typed with an Enclosure BC +type (not `EnclosureId`, `EnclosurePermitStatus` or +`EnclosureLifecycle`), so this port stays inside +`cora.infrastructure.ports`'s `depends_on = []` tach contract. +The two status axes are `Literal` aliases rather than bare `str`: +`Literal` comes from `typing`, so the alias pins the enum's value +set without importing the enum. A fitness test pins the alias to +the enum and fails if the two ever drift. Adapters produce a +member's `.value`, which narrows to exactly the alias, so no cast +is needed. Enclosure BC callers still wrap at the boundary if they +want the NewType discipline. """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +EnclosurePermitStatusValue = Literal["Permitted", "NotPermitted", "Unknown"] +EnclosureLifecycleValue = Literal["Active", "Decommissioned"] + @dataclass(frozen=True) class EnclosureLookupResult: @@ -106,8 +114,8 @@ class EnclosureLookupResult: enclosure_id: UUID name: str - permit_status: str - lifecycle: str + permit_status: EnclosurePermitStatusValue + lifecycle: EnclosureLifecycleValue permit_status_changed_at: str | None source_kind: str | None source_id: str | None @@ -224,6 +232,8 @@ async def lookup_by_name( __all__ = [ "AlwaysPermittedEnclosureLookup", + "EnclosureLifecycleValue", "EnclosureLookup", "EnclosureLookupResult", + "EnclosurePermitStatusValue", ] diff --git a/apps/api/src/cora/infrastructure/ports/facility_lookup.py b/apps/api/src/cora/infrastructure/ports/facility_lookup.py index 35751ff64fa..2760939059a 100644 --- a/apps/api/src/cora/infrastructure/ports/facility_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/facility_lookup.py @@ -45,11 +45,15 @@ ## No BC imports in the port -`kind` and `status` are typed as `str` (not the Federation BC's -`FacilityKind` / `FacilityStatus` StrEnums) so this port stays -inside `cora.infrastructure`'s `depends_on = []` tach contract. The -values match the StrEnum string values; deciders partition by literal -comparison (`kind == "Site"`, `status == "Active"`). +`kind` and `status` are `Literal` aliases pinning Federation BC's +`FacilityKind` and `FacilityStatus` StrEnum value sets, not the +StrEnums themselves, so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so an alias pins a value set without importing +the enum. A fitness test pins each alias to its enum and fails if +the two ever drift, so a decider partitioning by literal comparison +(`kind == "Site"`, `status == "Active"`) is a type error the moment +it names a value the enum does not have. `trust_anchor_credential_ids` is typed `frozenset[UUID]` (not `frozenset[CredentialId]`) for the same tach reason; Federation BC @@ -66,11 +70,14 @@ from collections.abc import Sequence from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID from cora.shared.facility_code import FacilityCode +FacilityKindValue = Literal["Site", "Area"] +FacilityStatusValue = Literal["Active", "Decommissioned"] + @dataclass(frozen=True) class FacilityLookupResult: @@ -81,9 +88,10 @@ class FacilityLookupResult: `FacilityLookup.lookup` and handed to the decider in the slice's context object (mirrors `CredentialLookupResult` shape). - `kind` and `status` are the StrEnum values as plain strings (matches - the projection's `TEXT` columns); the decider partitions on - `kind == "Site"` / `"Area"` and `status == "Active"`. + `kind` is the `FacilityKind` StrEnum value as a plain string; + `status` is a `Literal` alias pinning `FacilityStatus`'s value set + (both match the projection's `TEXT` columns). The decider + partitions on `kind == "Site"` / `"Area"` and `status == "Active"`. `code` is a `FacilityCode` value object per the two-tier facility-identity design; the adapter constructs the VO from the raw @@ -99,8 +107,8 @@ class FacilityLookupResult: id: UUID code: FacilityCode - kind: str - status: str + kind: FacilityKindValue + status: FacilityStatusValue trust_anchor_credential_ids: frozenset[UUID] @@ -153,4 +161,4 @@ async def list_active(self) -> Sequence[FacilityLookupResult]: ... -__all__ = ["FacilityLookup", "FacilityLookupResult"] +__all__ = ["FacilityLookup", "FacilityLookupResult", "FacilityStatusValue"] diff --git a/apps/api/src/cora/infrastructure/ports/family_lookup.py b/apps/api/src/cora/infrastructure/ports/family_lookup.py index b3f1b2b5b7b..b64fd7eef03 100644 --- a/apps/api/src/cora/infrastructure/ports/family_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/family_lookup.py @@ -31,9 +31,13 @@ match the Affordance StrEnum string values; consumer deciders cast to typed enums at their BC boundary if they want the discipline. -`status` is typed `str` (not the `FamilyStatus` StrEnum) for the -same tach reason. Consumers partition on the literal -("Defined" / "Versioned" / "Deprecated"). +`status` is a `Literal` alias rather than bare `str` (not the +`FamilyStatus` StrEnum): `Literal` comes from `typing`, so the alias +pins the enum's value set without importing the enum, for the same +tach reason. A fitness test pins the alias to the enum and fails if +the two ever drift. Adapters produce a member's `.value`, which +narrows to exactly the alias, so no cast is needed. Consumers +partition on the literal ("Defined" / "Versioned" / "Deprecated"). `presents_as` is typed `frozenset[UUID]` (not `frozenset[RoleId]`); consumers cast at their BC boundary if they need the typed @@ -41,9 +45,11 @@ """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +FamilyStatusValue = Literal["Defined", "Versioned", "Deprecated"] + @dataclass(frozen=True) class FamilyLookupResult: @@ -64,10 +70,10 @@ class FamilyLookupResult: incrementally via `add_family_presents_as` / removes via `remove_family_presents_as`. - `status` is the FSM stage as a plain string ("Defined" / - "Versioned" / "Deprecated"); consumer decides whether - Deprecated Families are acceptable bindings (today's posture: - accept; deprecation is advisory). + `status` is a `Literal` alias pinning `FamilyStatus`'s value set + ("Defined" / "Versioned" / "Deprecated"); consumer decides + whether Deprecated Families are acceptable bindings (today's + posture: accept; deprecation is advisory). `name` is the operator-readable display name; useful for surfacing in cross-BC error messages. @@ -75,7 +81,7 @@ class FamilyLookupResult: id: UUID name: str - status: str + status: FamilyStatusValue affordances: frozenset[str] presents_as: frozenset[UUID] @@ -99,4 +105,4 @@ async def lookup(self, family_id: UUID) -> FamilyLookupResult | None: ... -__all__ = ["FamilyLookup", "FamilyLookupResult"] +__all__ = ["FamilyLookup", "FamilyLookupResult", "FamilyStatusValue"] diff --git a/apps/api/src/cora/infrastructure/ports/federation/__init__.py b/apps/api/src/cora/infrastructure/ports/federation/__init__.py index 85cc86a93ab..c5dc86051d2 100644 --- a/apps/api/src/cora/infrastructure/ports/federation/__init__.py +++ b/apps/api/src/cora/infrastructure/ports/federation/__init__.py @@ -30,8 +30,11 @@ NoAdapterForFacilityError, ) from cora.infrastructure.ports.federation.permit_lookup import ( + AbiTierValue, + DirectionValue, PermitLookup, PermitLookupResult, + PermitStatusValue, ) from cora.infrastructure.ports.federation.publish_port import PublishPort from cora.infrastructure.ports.federation.pull_port import PullPort @@ -65,12 +68,14 @@ ) __all__ = [ + "AbiTierValue", "ArtifactReference", "AssistedBy", "CoDevelopedBy", "CoseSign1ScittEnvelope", "CredentialRef", "DcoEntry", + "DirectionValue", "DsseSigstoreKeylessEnvelope", "DsseStaticJwksEnvelope", "FederationAdoptionWindowClosedError", @@ -89,6 +94,7 @@ "NoAdapterForFacilityError", "PermitLookup", "PermitLookupResult", + "PermitStatusValue", "PublicationStatus", "PublishPort", "PublishReceipt", diff --git a/apps/api/src/cora/infrastructure/ports/federation/permit_lookup.py b/apps/api/src/cora/infrastructure/ports/federation/permit_lookup.py index df69be9e3bc..063528880b4 100644 --- a/apps/api/src/cora/infrastructure/ports/federation/permit_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/federation/permit_lookup.py @@ -46,11 +46,15 @@ """ from dataclasses import dataclass -from typing import Protocol, runtime_checkable +from typing import Literal, Protocol, runtime_checkable from uuid import UUID from cora.shared.facility_code import FacilityCode +DirectionValue = Literal["Outbound", "Inbound"] +PermitStatusValue = Literal["Defined", "Active", "Suspended", "Revoked"] +AbiTierValue = Literal["Testing", "Stable", "Obsolete", "Removed"] + @dataclass(frozen=True) class PermitLookupResult: @@ -72,9 +76,9 @@ class PermitLookupResult: permit_id: UUID peer_facility_id: FacilityCode - direction: str - status: str - abi_tier_floor: str + direction: DirectionValue + status: PermitStatusValue + abi_tier_floor: AbiTierValue current_version: int @@ -102,4 +106,10 @@ async def lookup_inbound( ) -> PermitLookupResult | None: ... -__all__ = ["PermitLookup", "PermitLookupResult"] +__all__ = [ + "AbiTierValue", + "DirectionValue", + "PermitLookup", + "PermitLookupResult", + "PermitStatusValue", +] diff --git a/apps/api/src/cora/infrastructure/ports/language_model_lookup.py b/apps/api/src/cora/infrastructure/ports/language_model_lookup.py index 15d2daf3026..4e1a338f8f9 100644 --- a/apps/api/src/cora/infrastructure/ports/language_model_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/language_model_lookup.py @@ -33,19 +33,41 @@ means "nothing currently approved for this identity" (never cataloged, or every entry for it is Defined or terminal), which the gate treats as refusal. + +## No BC imports in the port + +`status` is a `Literal` alias rather than bare `str` (not the Agent +BC's `LanguageModelStatus` StrEnum), so this port stays inside +`cora.infrastructure`'s `depends_on = []` tach contract: `Literal` +comes from `typing`, so the alias pins the enum's value set without +importing the enum. A fitness test pins the alias to the enum and +fails if the two ever drift. Adapters produce a member's `.value`, +which narrows to exactly the alias, so no cast is needed. +`data_tier` and `archivability` stay bare `str`: no consumer +partitions on either value set yet. """ from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +LanguageModelStatusValue = Literal[ + "Defined", "Approved", "RetirementAnnounced", "Retired", "Deprecated" +] + @dataclass(frozen=True) class LanguageModelLookupResult: - """The catalog's answer for one (provider, model) identity.""" + """The catalog's answer for one (provider, model) identity. + + `status` is a `Literal` alias pinning `LanguageModelStatus`'s + value set; `find_by_model` only ever returns an `"Approved"` row + (see the port docstring's Failure direction section), but the + field itself can hold any status the enum has. + """ language_model_id: UUID - status: str + status: LanguageModelStatusValue data_tier: str archivability: str snapshot_pin: str | None @@ -104,4 +126,5 @@ async def find_by_model( "AlwaysApprovedLanguageModelLookup", "LanguageModelLookup", "LanguageModelLookupResult", + "LanguageModelStatusValue", ] diff --git a/apps/api/src/cora/infrastructure/ports/supply_lookup.py b/apps/api/src/cora/infrastructure/ports/supply_lookup.py index 30e56bab1db..336e95bb7b8 100644 --- a/apps/api/src/cora/infrastructure/ports/supply_lookup.py +++ b/apps/api/src/cora/infrastructure/ports/supply_lookup.py @@ -26,13 +26,29 @@ layer per [[project_deregister_supply_design]] (tombstones should not count toward gate satisfaction). See [[project_supply_preflight_gate_design]] for the shared decision. + +## No BC imports in the port + +`status` is a `Literal` alias pinning Supply BC's `SupplyStatus` +StrEnum value set (not the StrEnum itself), so this port stays +inside `cora.infrastructure`'s `depends_on = []` tach contract: +`Literal` comes from `typing`, so the alias pins the enum's value +set without importing the enum. A fitness test pins the alias to +the enum and fails if the two ever drift. `kind` stays bare `str` +(Supply BC has not yet closed it to a StrEnum, per its own +roadmap); `facility_code` stays bare `str` per the +Facility-aggregate bare-str-on-the-wire convention. """ from collections.abc import Mapping from dataclasses import dataclass -from typing import Protocol +from typing import Literal, Protocol from uuid import UUID +SupplyStatusValue = Literal[ + "Unknown", "Available", "Degraded", "Unavailable", "Recovering", "Decommissioned" +] + @dataclass(frozen=True) class SupplyLookupResult: @@ -47,10 +63,10 @@ class SupplyLookupResult: `SupplyLookup.find_supplies_by_name` (by natural-key attributes) and handed to the consumer's decider. - `status` is the StrEnum value as a plain string (matches the - projection's `TEXT` column); consumers treat it opaquely and - partition on `"Available"` (pre-flight gate) or pass it through - (register_distribution status-agnostic bind). + `status` is a `Literal` alias pinning `SupplyStatus`'s value set + (matches the projection's `TEXT` column); consumers treat it + opaquely and partition on `"Available"` (pre-flight gate) or pass + it through (register_distribution status-agnostic bind). `kind` is the bare-str Supply.kind value (today; future closed-StrEnum move per Supply BC's own roadmap). Distribution's register decider @@ -68,7 +84,7 @@ class SupplyLookupResult: supply_id: UUID kind: str name: str - status: str + status: SupplyStatusValue facility_code: str @@ -108,14 +124,14 @@ async def lookup(self, supply_id: UUID) -> SupplyLookupResult | None: use this single-id query in preference to the grouped find_supplies_by_kind interface. - Supplies in EVERY status are returned (Available, Degraded, - Unavailable, Recovering, Decommissioned); the consumer - decider partitions on `status` if it needs to distinguish - "no Supply at all" from "Supply exists but in non-Available - state". register_distribution intentionally accepts every - status because a Distribution can legitimately be registered - against a Decommissioned Supply for archival completeness; - only `kind` is gated. + Supplies in EVERY status are returned (Unknown, Available, + Degraded, Unavailable, Recovering, Decommissioned); the + consumer decider partitions on `status` if it needs to + distinguish "no Supply at all" from "Supply exists but in + non-Available state". register_distribution intentionally + accepts every status because a Distribution can legitimately + be registered against a Decommissioned Supply for archival + completeness; only `kind` is gated. Mirrors `AssetLookup.lookup` shape one-for-one for cross-port symmetry. diff --git a/apps/api/src/cora/recipe/adapters/postgres_capability_lookup.py b/apps/api/src/cora/recipe/adapters/postgres_capability_lookup.py index 032c316cad1..12846df1529 100644 --- a/apps/api/src/cora/recipe/adapters/postgres_capability_lookup.py +++ b/apps/api/src/cora/recipe/adapters/postgres_capability_lookup.py @@ -25,6 +25,20 @@ upstream aggregate. `proj_recipe_capability_summary` is exactly that: a denormalized view maintained by Recipe's projection worker. The lookup adapter reads it directly via the shared asyncpg pool. + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `CapabilityLookupResult` (to keep +`cora.infrastructure.ports.capability_lookup` import-free of Recipe +BC types). The adapter constructs `CapabilityStatus(row["status"])` +as a validation step: a corrupted row whose `status` is not a known +enum value surfaces as `ValueError` from the adapter rather than as +a silent wrong-status match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. +The SQL filter narrows to `{Defined, Versioned}`; the alias still +pins the full three-value enum because a query filter is adapter +behavior, not a constraint on the field's type. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -34,6 +48,7 @@ import asyncpg from cora.infrastructure.ports.capability_lookup import CapabilityLookupResult +from cora.recipe.aggregates.capability import CapabilityStatus _FIND_APPLICABLE_BY_AFFORDANCES_SQL = """ SELECT capability_id, code, name, status @@ -67,5 +82,5 @@ def _row_to_reference(row: Any) -> CapabilityLookupResult: capability_id=row["capability_id"], code=str(row["code"]), name=str(row["name"]), - status=str(row["status"]), + status=CapabilityStatus(row["status"]).value, ) diff --git a/apps/api/src/cora/safety/adapters/postgres_clearance_lookup.py b/apps/api/src/cora/safety/adapters/postgres_clearance_lookup.py index 85eefa75d05..308214065b1 100644 --- a/apps/api/src/cora/safety/adapters/postgres_clearance_lookup.py +++ b/apps/api/src/cora/safety/adapters/postgres_clearance_lookup.py @@ -30,6 +30,17 @@ NOT NULL guard skips the subject_binding_ids match when subject_id is None. `asset_ids` may be empty (rare but valid); the `&&` overlap operator handles empty arrays correctly. + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `ClearanceLookupResult` (to keep +`cora.infrastructure.ports.clearance_lookup` import-free of Safety +BC types). The adapter constructs `ClearanceStatus(row["status"])` +as a validation step: a corrupted row whose `status` is not a known +enum value surfaces as `ValueError` from the adapter rather than as +a silent wrong-status match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -40,6 +51,7 @@ import asyncpg from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult +from cora.safety.aggregates.clearance import ClearanceStatus _FIND_REFERENCING_RUN_SQL = """ SELECT clearance_id, status, template_id, template_code, facility_code @@ -79,7 +91,7 @@ async def find_covering( def _row_to_reference(row: Any) -> ClearanceLookupResult: return ClearanceLookupResult( clearance_id=row["clearance_id"], - status=str(row["status"]), + status=ClearanceStatus(row["status"]).value, template_id=row["template_id"], template_code=str(row["template_code"]), facility_code=str(row["facility_code"]), diff --git a/apps/api/src/cora/safety/adapters/postgres_clearance_template_lookup.py b/apps/api/src/cora/safety/adapters/postgres_clearance_template_lookup.py index ba0c9962680..77ff7692efb 100644 --- a/apps/api/src/cora/safety/adapters/postgres_clearance_template_lookup.py +++ b/apps/api/src/cora/safety/adapters/postgres_clearance_template_lookup.py @@ -27,15 +27,15 @@ ## Enum coercion -`status` is stored as `TEXT` and typed as `str` on the port's -`ClearanceTemplateLookupResult` (to keep +`status` is stored as `TEXT` and typed as a `Literal` alias on the +port's `ClearanceTemplateLookupResult` (to keep `cora.infrastructure.ports.clearance_template_lookup` import-free -of Safety BC types). The adapter still constructs +of Safety BC types). The adapter constructs `ClearanceTemplateStatus(row["status"])` as a validation step: a corrupted row whose `status` is not a known enum value surfaces as `ValueError` from the adapter rather than as a silent wrong -status downstream. The validated `StrEnum` value IS-A `str`, so -the assignment into the dataclass's `str`-typed field is exact. +status downstream. `.value` on the validated member narrows to +exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false diff --git a/apps/api/src/cora/supply/adapters/postgres_supply_lookup.py b/apps/api/src/cora/supply/adapters/postgres_supply_lookup.py index 7beecb97932..d855502e608 100644 --- a/apps/api/src/cora/supply/adapters/postgres_supply_lookup.py +++ b/apps/api/src/cora/supply/adapters/postgres_supply_lookup.py @@ -67,6 +67,17 @@ AND status != 'Decommissioned' ORDER BY registered_at, supply_id ``` + +## Enum coercion + +`status` is stored as a `TEXT` column and typed as a `Literal` alias +on the port's `SupplyLookupResult` (to keep +`cora.infrastructure.ports.supply_lookup` import-free of Supply BC +types). The adapter constructs `SupplyStatus(row["status"])` as a +validation step: a corrupted row whose `status` is not a known enum +value surfaces as `ValueError` from the adapter rather than as a +silent wrong-status match downstream. `.value` on the validated +member narrows to exactly the port's alias, so no cast is needed. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -78,6 +89,7 @@ import asyncpg from cora.infrastructure.ports.supply_lookup import SupplyLookupResult +from cora.supply.aggregates.supply import SupplyStatus _FIND_SUPPLIES_BY_KIND_SQL = """ SELECT supply_id, kind, name, status, facility_code @@ -150,6 +162,6 @@ def _row_to_reference(row: Any) -> SupplyLookupResult: supply_id=row["supply_id"], kind=str(row["kind"]), name=str(row["name"]), - status=str(row["status"]), + status=SupplyStatus(row["status"]).value, facility_code=str(row["facility_code"]), ) diff --git a/apps/api/tach.toml b/apps/api/tach.toml index 525c1e64cc6..adf34142ecc 100644 --- a/apps/api/tach.toml +++ b/apps/api/tach.toml @@ -3,7 +3,7 @@ # Enforces: # 1. Layered: shared depends on nothing; infrastructure depends only # on shared; BCs depend on shared + infrastructure (+ any sibling -# aggregate kernels they integrate with); api depends on every BC. +# aggregate kernels they NAME); api depends on every BC. # `cora.shared` is the pure-value-object foundation (Identifier # VOs, NewType identity aliases, bounded-text validators, JSON # Schema helpers); `cora.infrastructure` is composition root, @@ -20,6 +20,33 @@ # enforced by `tests/architecture/` (AST-based) rather than tach, # because per-slice module declarations would balloon this file. # +# What this file does NOT say, because the distinction has misled readers +# (including during the audit that added this paragraph): +# +# It constrains IMPORTS. Naming is one of several ways one BC comes to +# depend on another, and the rest leave no import to constrain: +# +# - A Protocol in `cora.infrastructure.ports`, implemented by one BC's +# adapter and consumed by another. `cora.operation` reads +# `EnclosureLookup` and cannot start a Procedure without Enclosure's +# permit data, yet declares no edge to `cora.enclosure`: both sides +# name only `cora.infrastructure`, which every module does. 20 lookup +# adapters inside BCs bind to a protocol declared in that shared +# namespace, so this is a category, not one case. +# - An event subscription, which names its producer with a STRING. +# - Anything under `tests`, which `exclude` drops entirely. +# +# So read this as the doors that have been cut, not as a map of what +# depends on what. Where the two differ, the map is the larger one. Whether +# those ports should move into their owning BC (making the map honest at +# the cost of the isolation the shared namespace buys) is an open question +# and not something this file currently answers either way. +# +# `tests/architecture/test_tach_edges_are_used.py` fails on an entry no +# source file takes up, so a permission whose reason has gone away does not +# quietly stay. The contract went 146 additions to 1 removal before that +# existed. +# # Run: `uv run tach check` (also wired into pre-commit and CI). # Visualize: `uv run tach show` opens the dependency graph. @@ -96,19 +123,15 @@ depends_on = ["cora.infrastructure", "cora.shared"] [[modules]] path = "cora.enclosure.adapters" -depends_on = ["cora.infrastructure", "cora.shared", "cora.enclosure.aggregates"] +depends_on = ["cora.infrastructure", "cora.enclosure.aggregates"] [[modules]] path = "cora.equipment.aggregates" depends_on = ["cora.infrastructure", "cora.shared"] -[[modules]] -path = "cora.equipment.ports" -depends_on = ["cora.infrastructure", "cora.shared", "cora.equipment.aggregates"] - [[modules]] path = "cora.equipment.adapters" -depends_on = ["cora.infrastructure", "cora.shared", "cora.equipment.aggregates", "cora.equipment.ports"] +depends_on = ["cora.infrastructure", "cora.equipment.aggregates"] [[modules]] path = "cora.federation.aggregates" @@ -147,7 +170,7 @@ depends_on = ["cora.infrastructure", "cora.shared"] # with at handler-load time (cross-aggregate validation pattern). [[modules]] path = "cora.access" -depends_on = ["cora.infrastructure", "cora.shared", "cora.access.aggregates"] +depends_on = ["cora.infrastructure", "cora.access.aggregates"] # Data BC depends on its own aggregates plus Run + Subject aggregate # kernels for cross-aggregate validation at register_dataset time @@ -190,8 +213,6 @@ depends_on = [ "cora.infrastructure", "cora.shared", "cora.equipment.aggregates", - "cora.equipment.ports", - "cora.equipment.adapters", ] [[modules]] @@ -214,7 +235,7 @@ depends_on = ["cora.infrastructure", "cora.shared", "cora.supply.aggregates"] # deferred per the Stage-1 Sub-Slice A open ambiguity on Asset binding). [[modules]] path = "cora.enclosure" -depends_on = ["cora.infrastructure", "cora.shared", "cora.enclosure.aggregates", "cora.enclosure.adapters"] +depends_on = ["cora.infrastructure", "cora.shared", "cora.enclosure.aggregates"] [[modules]] path = "cora.trust" diff --git a/apps/api/tests/architecture/test_port_status_literals_match_owning_enums.py b/apps/api/tests/architecture/test_port_status_literals_match_owning_enums.py new file mode 100644 index 00000000000..ea7fb8a1262 --- /dev/null +++ b/apps/api/tests/architecture/test_port_status_literals_match_owning_enums.py @@ -0,0 +1,216 @@ +"""Architecture fitness: a port's status `Literal` must match its owning StrEnum. + +A cross-BC lookup port cannot import the owning BC's StrEnum: the ports package +holds `depends_on = []` in `tach.toml`, and that neutrality is what lets several +BCs answer the same question. So a status axis crosses the port surface as a +value, and a consumer in another BC partitions on it by writing the string out +by hand. + +Typing that field as a `Literal` alias instead of a bare `str` keeps the port +neutral (`Literal` is stdlib) while pinning the exact value set, so a consumer +comparing against a value the enum does not have is a type error at the call +site. That only holds while the alias and the enum agree, which is what the +first test below asserts. + +Why the value set and not "a value of SOME StrEnum": `"Active"` is a member of +14 different StrEnums in this codebase, so a check that only asked whether a +literal belongs to some enum would pass a `ClearanceStatus` value handed to an +enclosure gate. It would also have passed the `permit_status="Denied"` fixture +this pin caught on its first application, since `"Denied"` is a real +`RatificationStatus` value. + +The registry is hand-maintained, which is the failure mode described in +`project_field_drop_bug_class`: a new port field added without a registry entry +would be silently unguarded. `test_every_port_status_literal_is_registered` +closes that by enumerating the aliases from source, so the registry has to keep +up with the ports rather than the other way round. +""" + +import ast +from enum import StrEnum +from typing import Final, get_args + +import pytest + +from cora.agent.aggregates.language_model import LanguageModelStatus +from cora.data.aggregates.distribution import DistributionStatus +from cora.enclosure.aggregates.enclosure.state import ( + EnclosureLifecycle, + EnclosurePermitStatus, +) +from cora.equipment.aggregates.assembly import AssemblyStatus +from cora.equipment.aggregates.asset import AssetLifecycle, AssetTier +from cora.equipment.aggregates.family import FamilyStatus +from cora.federation.aggregates.credential import CredentialStatus +from cora.federation.aggregates.facility import FacilityKind, FacilityStatus +from cora.federation.aggregates.permit.state import ( + AbiTier, + Direction, + PermitStatus, +) +from cora.infrastructure.ports.assembly_lookup import AssemblyStatusValue +from cora.infrastructure.ports.asset_lookup import AssetLifecycleValue, AssetTierValue +from cora.infrastructure.ports.capability_lookup import CapabilityStatusValue +from cora.infrastructure.ports.clearance_lookup import ClearanceStatusValue +from cora.infrastructure.ports.clearance_template_lookup import ( + ClearanceTemplateStatusValue, +) +from cora.infrastructure.ports.credential_lookup import CredentialStatusValue +from cora.infrastructure.ports.dataset_distribution_lookup import ( + DistributionStatusValue, +) +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosurePermitStatusValue, +) +from cora.infrastructure.ports.facility_lookup import ( + FacilityKindValue, + FacilityStatusValue, +) +from cora.infrastructure.ports.family_lookup import FamilyStatusValue +from cora.infrastructure.ports.federation import ( + AbiTierValue, + DirectionValue, + PermitStatusValue, +) +from cora.infrastructure.ports.language_model_lookup import LanguageModelStatusValue +from cora.infrastructure.ports.supply_lookup import SupplyStatusValue +from cora.recipe.aggregates.capability import CapabilityStatus +from cora.safety.aggregates.clearance import ClearanceStatus +from cora.safety.aggregates.clearance_template import ClearanceTemplateStatus +from cora.supply.aggregates.supply import SupplyStatus +from tests.architecture.conftest import tracked_python_files + +_PORTS_DIR: Final = "infrastructure/ports" + +_ALIAS_SUFFIX: Final = "Value" + +REGISTRY: Final[tuple[tuple[str, object, type[StrEnum]], ...]] = ( + ("EnclosurePermitStatusValue", EnclosurePermitStatusValue, EnclosurePermitStatus), + ("EnclosureLifecycleValue", EnclosureLifecycleValue, EnclosureLifecycle), + ("AssetTierValue", AssetTierValue, AssetTier), + ("AssetLifecycleValue", AssetLifecycleValue, AssetLifecycle), + ("AssemblyStatusValue", AssemblyStatusValue, AssemblyStatus), + ("FamilyStatusValue", FamilyStatusValue, FamilyStatus), + ("CapabilityStatusValue", CapabilityStatusValue, CapabilityStatus), + ("LanguageModelStatusValue", LanguageModelStatusValue, LanguageModelStatus), + ("DistributionStatusValue", DistributionStatusValue, DistributionStatus), + ("ClearanceStatusValue", ClearanceStatusValue, ClearanceStatus), + ("ClearanceTemplateStatusValue", ClearanceTemplateStatusValue, ClearanceTemplateStatus), + ("CredentialStatusValue", CredentialStatusValue, CredentialStatus), + ("FacilityStatusValue", FacilityStatusValue, FacilityStatus), + ("FacilityKindValue", FacilityKindValue, FacilityKind), + ("SupplyStatusValue", SupplyStatusValue, SupplyStatus), + ("DirectionValue", DirectionValue, Direction), + ("PermitStatusValue", PermitStatusValue, PermitStatus), + ("AbiTierValue", AbiTierValue, AbiTier), +) + + +_AXIS_FIELD_NAMES: Final = frozenset( + {"status", "lifecycle", "tier", "state", "direction", "permit_status", "kind"} +) + +UNPINNED_AXIS_FIELDS: Final[dict[tuple[str, str], str]] = { + ("SupplyLookupResult", "kind"): ( + "Supply.kind is free-form text today; no SupplyKind StrEnum exists to " + "pin against. Pin this the moment that enum lands." + ), +} + + +def _declared_aliases() -> dict[str, str]: + """Every `Value = Literal[...]` alias defined under the ports package. + + Read from source rather than by importing the package, so an alias that + fails to import is a visible failure here rather than a silent absence. + """ + found: dict[str, str] = {} + for path in tracked_python_files(): + if _PORTS_DIR not in path.as_posix(): + continue + for node in ast.parse(path.read_text()).body: + if not isinstance(node, ast.Assign) or len(node.targets) != 1: + continue + target = node.targets[0] + if not isinstance(target, ast.Name) or not target.id.endswith(_ALIAS_SUFFIX): + continue + if not isinstance(node.value, ast.Subscript): + continue + if ast.unparse(node.value.value).split(".")[-1] != "Literal": + continue + found[target.id] = path.name + return found + + +@pytest.mark.architecture +@pytest.mark.parametrize( + ("name", "alias", "enum"), REGISTRY, ids=lambda v: getattr(v, "__name__", v) +) +def test_port_status_literal_matches_owning_enum( + name: str, alias: object, enum: type[StrEnum] +) -> None: + literal_values = set(get_args(alias)) + enum_values = {member.value for member in enum} + assert literal_values == enum_values, ( + f"{name} and {enum.__name__} have drifted: " + f"the alias has {sorted(literal_values)}, the enum has {sorted(enum_values)}. " + "Widen both together. The alias exists to pin the enum's value set on a " + "port that cannot import the enum, so a one-sided change silently " + "un-guards every cross-BC comparison against this field." + ) + + +@pytest.mark.architecture +def test_every_port_status_literal_is_registered() -> None: + declared = _declared_aliases() + registered = {name for name, _, _ in REGISTRY} + unregistered = {name: mod for name, mod in declared.items() if name not in registered} + assert not unregistered, ( + "port Literal aliases with no owning-enum pin: " + + ", ".join(f"{name} ({mod})" for name, mod in sorted(unregistered.items())) + + ". Add each to REGISTRY with the StrEnum it mirrors, or the alias is a " + "hand-written value set that nothing keeps in step with its source." + ) + + +@pytest.mark.architecture +def test_no_port_axis_field_is_left_as_bare_str() -> None: + """Range over port FIELDS, not over the aliases that happen to exist. + + The registry check above asks whether every alias is pinned, which is blind + to a field that never got an alias at all. That blindness is not + hypothetical: it is how `federation/permit_lookup.py` was missed when the + other ports were converted, and it is the shape described in + `project_aggregate_coverage_blindness`. The subject of this check is + therefore the field. + + A genuine exception is recorded in `UNPINNED_AXIS_FIELDS` with its reason, + so the absence of a pin is a written claim rather than a silent gap. + """ + bare: list[str] = [] + for path in tracked_python_files(): + if _PORTS_DIR not in path.as_posix(): + continue + for node in ast.walk(ast.parse(path.read_text())): + if not isinstance(node, ast.ClassDef): + continue + for stmt in node.body: + if not isinstance(stmt, ast.AnnAssign) or not isinstance(stmt.target, ast.Name): + continue + field = stmt.target.id + if field not in _AXIS_FIELD_NAMES: + continue + if ast.unparse(stmt.annotation) != "str": + continue + if (node.name, field) in UNPINNED_AXIS_FIELDS: + continue + bare.append(f"{path.name}:{node.name}.{field}") + assert not bare, ( + "port fields on a closed axis still typed as a bare str: " + + ", ".join(sorted(bare)) + + ". Give each a Literal alias pinning its owning StrEnum and register " + "it, or record it in UNPINNED_AXIS_FIELDS with the reason no enum " + "exists to pin against. A bare str here means a consumer in another BC " + "can compare it against a value the owner never defined." + ) diff --git a/apps/api/tests/architecture/test_tach_edges_are_used.py b/apps/api/tests/architecture/test_tach_edges_are_used.py new file mode 100644 index 00000000000..0062d381e36 --- /dev/null +++ b/apps/api/tests/architecture/test_tach_edges_are_used.py @@ -0,0 +1,94 @@ +"""A declared dependency nobody imports is a permission nobody asked for. + +`tach.toml` says who may know about whom, and every entry is a hole +deliberately left in a wall. Holes are added when a need appears and nothing +closes them when the need goes away: over the contract's first four months it +gained 146 entries and lost 1, and that single removal was a hand-written +refactor rather than anything the process noticed. + +That ratio is the problem this test exists for. A permission outlives its +reason silently, and the file that is supposed to answer "what may depend on +what" slowly turns into "what has ever depended on what". The audit that +prompted this found ten such entries, five of them guarding +`cora.equipment.ports`, a module holding a docstring and an empty `__all__` +after its one export was hoisted to `cora.shared.ports`. + +Nothing here judges whether a dependency SHOULD exist. It only asks whether +the code still takes the permission up, which is the half a machine can check. + +Git-tracked enumeration, not a filesystem walk, for the reason in the +`conftest` module docstring: pre-commit stashes only tracked files, so an +untracked module would be invisible here and the check would pass by not +looking. The schema-version pin was reported green that way an hour before +this test was written. +""" + +from __future__ import annotations + +import re +import tomllib +from collections import defaultdict +from typing import TYPE_CHECKING + +from tests.architecture.conftest import SRC_ROOT, tracked_python_files + +if TYPE_CHECKING: + from pathlib import Path + +_TACH = SRC_ROOT.parent / "tach.toml" + +_IMPORT = re.compile(r"^\s*(?:from|import)\s+(cora\.[A-Za-z0-9_.]*)", re.M) + + +def _declared() -> dict[str, list[str]]: + modules = tomllib.loads(_TACH.read_text())["modules"] + return {m["path"]: list(m.get("depends_on", ())) for m in modules} + + +def _dotted(path: Path) -> str: + rel = path.relative_to(SRC_ROOT).with_suffix("") + return ".".join(rel.parts).removesuffix(".__init__") + + +def _most_specific(candidates: list[str], target: str) -> str | None: + """The longest declared path that `target` sits under, or None. + + `cora.run.aggregates.run` belongs to `cora.run.aggregates` when both it and + `cora.run` are declared, so a module declaring BOTH is not credited for the + broad one by an import that the narrow one already covers. + """ + matches = [c for c in candidates if target == c or target.startswith(c + ".")] + return max(matches, key=len) if matches else None + + +def _imports_by_module() -> dict[str, set[str]]: + declared = _declared() + owners = sorted(declared, key=len, reverse=True) + found: dict[str, set[str]] = defaultdict(set) + for path in tracked_python_files(): + owner = _most_specific(owners, _dotted(path)) + if owner is None: + continue + found[owner].update(_IMPORT.findall(path.read_text())) + return found + + +def test_every_declared_dependency_is_actually_imported() -> None: + declared = _declared() + imports = _imports_by_module() + unused = [ + (module, dep) + for module, deps in sorted(declared.items()) + for dep in deps + if not any(_most_specific(deps, i) == dep for i in imports.get(module, ())) + ] + assert not unused, ( + "tach.toml declares dependencies that no tracked source file takes up:\n" + + "\n".join(f" {module} -> {dep}" for module, dep in unused) + + "\n\nEach is a hole in a wall with nothing passing through it. Delete " + "the entry; add it back in the commit that needs it, which is one line " + "and puts the reason next to the use. If the import is real but this " + "cannot see it (a dynamic import, say), that is worth a comment here " + "rather than an exemption, since a permission no reader can trace to a " + "caller is one nobody can ever retire." + ) diff --git a/apps/api/tests/contract/test_start_procedure_enclosure_preflight.py b/apps/api/tests/contract/test_start_procedure_enclosure_preflight.py index 6bc01cb990a..0bbcf66baf9 100644 --- a/apps/api/tests/contract/test_start_procedure_enclosure_preflight.py +++ b/apps/api/tests/contract/test_start_procedure_enclosure_preflight.py @@ -47,6 +47,7 @@ from cora.infrastructure.adapters.in_memory_enclosure_lookup import ( InMemoryEnclosureLookup, ) +from cora.infrastructure.ports.enclosure_lookup import EnclosurePermitStatusValue from tests.contract._subject_helpers import register_active_asset @@ -226,7 +227,7 @@ def test_post_start_procedure_returns_204_when_binding_enclosure_is_permitted() @pytest.mark.contract @pytest.mark.parametrize("permit_status", ["NotPermitted", "Unknown"]) def test_post_start_procedure_returns_409_when_binding_enclosure_is_not_permitted( - permit_status: str, + permit_status: EnclosurePermitStatusValue, ) -> None: """A non-Permitted located-in Enclosure raises 409 ProcedureRequiresPermittedEnclosureError.""" diff --git a/apps/api/tests/contract/test_start_run_enclosure_preflight.py b/apps/api/tests/contract/test_start_run_enclosure_preflight.py index a0c24fe23d2..417480c6861 100644 --- a/apps/api/tests/contract/test_start_run_enclosure_preflight.py +++ b/apps/api/tests/contract/test_start_run_enclosure_preflight.py @@ -44,7 +44,10 @@ from cora.infrastructure.adapters.in_memory_enclosure_lookup import ( InMemoryEnclosureLookup, ) -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLookupResult, + EnclosurePermitStatusValue, +) from tests.contract._helpers import create_capability_via_api from tests.contract._subject_helpers import register_active_asset @@ -175,7 +178,7 @@ def test_post_runs_returns_201_when_binding_enclosure_is_permitted_and_active() @pytest.mark.contract @pytest.mark.parametrize("permit_status", ["NotPermitted", "Unknown"]) def test_post_runs_returns_409_when_binding_enclosure_is_not_permitted( - permit_status: str, + permit_status: EnclosurePermitStatusValue, ) -> None: """A non-Permitted located-in Enclosure raises 409 RunRequiresPermittedEnclosureError.""" diff --git a/apps/api/tests/unit/calibration/test_publish_revision_decider.py b/apps/api/tests/unit/calibration/test_publish_revision_decider.py index f48479768fd..73c89861970 100644 --- a/apps/api/tests/unit/calibration/test_publish_revision_decider.py +++ b/apps/api/tests/unit/calibration/test_publish_revision_decider.py @@ -25,6 +25,7 @@ from cora.infrastructure.ports.federation import ( DsseStaticJwksEnvelope, PermitLookupResult, + PermitStatusValue, ) from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId @@ -85,7 +86,7 @@ def _command( ) -def _permit_result(*, status: str = "Active") -> PermitLookupResult: +def _permit_result(*, status: PermitStatusValue = "Active") -> PermitLookupResult: return PermitLookupResult( permit_id=_PERMIT_ID, peer_facility_id=FacilityCode(_PEER), diff --git a/apps/api/tests/unit/calibration/test_publish_revision_decider_properties.py b/apps/api/tests/unit/calibration/test_publish_revision_decider_properties.py index 3208de0e1f6..b2ca9c62c0d 100644 --- a/apps/api/tests/unit/calibration/test_publish_revision_decider_properties.py +++ b/apps/api/tests/unit/calibration/test_publish_revision_decider_properties.py @@ -27,7 +27,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, get_args from uuid import UUID, uuid4 import pytest @@ -54,6 +54,7 @@ from cora.infrastructure.ports.federation import ( DsseStaticJwksEnvelope, PermitLookupResult, + PermitStatusValue, ) from cora.shared.facility_code import FACILITY_CODE_MAX_LENGTH, FacilityCode from cora.shared.identity import ActorId @@ -72,8 +73,10 @@ _SIGNATURE_KID = printable_ascii_text(min_size=1, max_size=128) _SIGNING_VERSION = st.sampled_from(["cora/v1"]) _PAYLOAD_BYTES = st.binary(min_size=1, max_size=512) -_PERMIT_NON_ACTIVE_STATUS = st.sampled_from(["Defined", "Suspended", "Revoked"]) -_PERMIT_ANY_STATUS = st.sampled_from(["Defined", "Active", "Suspended", "Revoked"]) +_PERMIT_ANY_STATUS = st.sampled_from(get_args(PermitStatusValue)) +_PERMIT_NON_ACTIVE_STATUS = st.sampled_from( + tuple(value for value in get_args(PermitStatusValue) if value != "Active") +) def _revision( @@ -122,7 +125,9 @@ def _envelope(payload_bytes: bytes, signing_version: str) -> DsseStaticJwksEnvel ) -def _permit(*, permit_id: UUID, peer_facility_id: str, status: str) -> PermitLookupResult: +def _permit( + *, permit_id: UUID, peer_facility_id: str, status: PermitStatusValue +) -> PermitLookupResult: return PermitLookupResult( permit_id=permit_id, peer_facility_id=FacilityCode(peer_facility_id), @@ -188,7 +193,7 @@ def test_decide_with_unknown_revision_always_raises_revision_not_found( calibration_id: UUID, known_revision_id: UUID, queried_revision_id: UUID, - permit_status: str, + permit_status: PermitStatusValue, content_hash: str, peer_facility_id: str, receipt_id: UUID, @@ -244,7 +249,7 @@ def test_decide_with_unknown_revision_always_raises_revision_not_found( def test_decide_with_legacy_revision_always_raises_cannot_publish( calibration_id: UUID, revision_id: UUID, - permit_status: str, + permit_status: PermitStatusValue, peer_facility_id: str, receipt_id: UUID, now: datetime, @@ -300,7 +305,7 @@ def test_decide_with_inactive_permit_always_raises_permit_not_active( calibration_id: UUID, revision_id: UUID, content_hash: str, - permit_status: str, + permit_status: PermitStatusValue, peer_facility_id: str, receipt_id: UUID, now: datetime, diff --git a/apps/api/tests/unit/data/test_discard_distribution_decider.py b/apps/api/tests/unit/data/test_discard_distribution_decider.py index ab78c5e5bc3..6798db33172 100644 --- a/apps/api/tests/unit/data/test_discard_distribution_decider.py +++ b/apps/api/tests/unit/data/test_discard_distribution_decider.py @@ -42,6 +42,7 @@ from cora.data.features.discard_distribution.context import DiscardDistributionContext from cora.infrastructure.ports.dataset_distribution_lookup import ( DatasetDistributionLookupResult, + DistributionStatusValue, ) from cora.shared.identity import ActorId from cora.shared.text_bounds import REASON_MAX_LENGTH @@ -91,7 +92,7 @@ def _sibling( *, distribution_id: UUID, supply_id: UUID, - status: str, + status: DistributionStatusValue, ) -> DatasetDistributionLookupResult: return DatasetDistributionLookupResult( distribution_id=distribution_id, @@ -322,7 +323,7 @@ def test_decide_is_pure_same_inputs_same_outputs() -> None: def target_result( - target: Distribution, *, status: str = "Registered" + target: Distribution, *, status: DistributionStatusValue = "Registered" ) -> DatasetDistributionLookupResult: """The target copy's own projection row, present in the sibling set returned by find_by_datasets (the decider filters it out by id).""" diff --git a/apps/api/tests/unit/data/test_discard_distribution_decider_properties.py b/apps/api/tests/unit/data/test_discard_distribution_decider_properties.py index acac6847cb6..aea8ecf83a4 100644 --- a/apps/api/tests/unit/data/test_discard_distribution_decider_properties.py +++ b/apps/api/tests/unit/data/test_discard_distribution_decider_properties.py @@ -58,6 +58,7 @@ from cora.data.features.discard_distribution.context import DiscardDistributionContext from cora.infrastructure.ports.dataset_distribution_lookup import ( DatasetDistributionLookupResult, + DistributionStatusValue, ) from cora.shared.identity import ActorId from cora.shared.text_bounds import REASON_MAX_LENGTH @@ -96,7 +97,9 @@ def _distribution(*, distribution_id: UUID, supply_id: UUID) -> Distribution: ) -def _sibling(*, supply_id: UUID, status: str) -> DatasetDistributionLookupResult: +def _sibling( + *, supply_id: UUID, status: DistributionStatusValue +) -> DatasetDistributionLookupResult: return DatasetDistributionLookupResult( distribution_id=uuid4(), dataset_id=_DATASET_ID, diff --git a/apps/api/tests/unit/data/test_register_distribution_decider.py b/apps/api/tests/unit/data/test_register_distribution_decider.py index b4eb027a2a7..91ed432b886 100644 --- a/apps/api/tests/unit/data/test_register_distribution_decider.py +++ b/apps/api/tests/unit/data/test_register_distribution_decider.py @@ -41,7 +41,7 @@ DistributionRegistrationContext, RegisterDistribution, ) -from cora.infrastructure.ports.supply_lookup import SupplyLookupResult +from cora.infrastructure.ports.supply_lookup import SupplyLookupResult, SupplyStatusValue from cora.shared.identity import ActorId _GOOD_SHA256 = "a" * DATASET_CHECKSUM_SHA256_HEX_LENGTH @@ -89,7 +89,7 @@ def _supply( supply_id: UUID, *, kind: str = "Storage", - status: str = "Available", + status: SupplyStatusValue = "Available", ) -> SupplyLookupResult: return SupplyLookupResult( supply_id=supply_id, diff --git a/apps/api/tests/unit/data/test_register_distribution_handler.py b/apps/api/tests/unit/data/test_register_distribution_handler.py index 1a9acfd6dee..a5b1fb540c1 100644 --- a/apps/api/tests/unit/data/test_register_distribution_handler.py +++ b/apps/api/tests/unit/data/test_register_distribution_handler.py @@ -34,6 +34,7 @@ from cora.infrastructure.ports.supply_lookup import ( SingleSupplyLookup, SupplyLookupResult, + SupplyStatusValue, UnknownSupplyLookup, ) from cora.shared.identity import ActorId @@ -70,7 +71,7 @@ def _storage_supply_ref( *, supply_id: UUID = _SUPPLY_ID, kind: str = "Storage", - status: str = "Available", + status: SupplyStatusValue = "Available", ) -> SupplyLookupResult: return SupplyLookupResult( supply_id=supply_id, diff --git a/apps/api/tests/unit/federation/test_initialize_seal_decider.py b/apps/api/tests/unit/federation/test_initialize_seal_decider.py index 054c322f47a..2738f5605f2 100644 --- a/apps/api/tests/unit/federation/test_initialize_seal_decider.py +++ b/apps/api/tests/unit/federation/test_initialize_seal_decider.py @@ -37,7 +37,10 @@ ) from cora.federation.features import initialize_seal from cora.federation.features.initialize_seal import InitializeSeal -from cora.infrastructure.ports.credential_lookup import CredentialLookupResult +from cora.infrastructure.ports.credential_lookup import ( + CredentialLookupResult, + CredentialStatusValue, +) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId @@ -78,7 +81,7 @@ def _online_cred( credential_id: UUID = _ONLINE_KEY_REF, *, purpose: str = CredentialPurpose.SEAL_ONLINE_SIGNING.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, facility_id: str = _FACILITY_CODE, ) -> CredentialLookupResult: return CredentialLookupResult( @@ -93,7 +96,7 @@ def _offline_cred( credential_id: UUID = _OFFLINE_KEY_REF, *, purpose: str = CredentialPurpose.SEAL_OFFLINE_ROOT.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, facility_id: str = _FACILITY_CODE, ) -> CredentialLookupResult: return CredentialLookupResult( diff --git a/apps/api/tests/unit/federation/test_initialize_seal_decider_properties.py b/apps/api/tests/unit/federation/test_initialize_seal_decider_properties.py index 332902e82c7..e99b41af088 100644 --- a/apps/api/tests/unit/federation/test_initialize_seal_decider_properties.py +++ b/apps/api/tests/unit/federation/test_initialize_seal_decider_properties.py @@ -45,7 +45,10 @@ ) from cora.federation.features import initialize_seal from cora.federation.features.initialize_seal import InitializeSeal -from cora.infrastructure.ports.credential_lookup import CredentialLookupResult +from cora.infrastructure.ports.credential_lookup import ( + CredentialLookupResult, + CredentialStatusValue, +) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId @@ -87,7 +90,7 @@ def _online_cred( credential_id: UUID = _ONLINE_KEY_REF, *, purpose: str = CredentialPurpose.SEAL_ONLINE_SIGNING.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, facility_id: str = _FACILITY_CODE, ) -> CredentialLookupResult: return CredentialLookupResult( @@ -102,7 +105,7 @@ def _offline_cred( credential_id: UUID = _OFFLINE_KEY_REF, *, purpose: str = CredentialPurpose.SEAL_OFFLINE_ROOT.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, facility_id: str = _FACILITY_CODE, ) -> CredentialLookupResult: return CredentialLookupResult( diff --git a/apps/api/tests/unit/federation/test_register_facility_decider.py b/apps/api/tests/unit/federation/test_register_facility_decider.py index 640cca90454..4f2774a95e5 100644 --- a/apps/api/tests/unit/federation/test_register_facility_decider.py +++ b/apps/api/tests/unit/federation/test_register_facility_decider.py @@ -27,6 +27,10 @@ from cora.federation.features import register_facility from cora.federation.features.register_facility import RegisterFacility from cora.infrastructure.ports import FacilityLookupResult +from cora.infrastructure.ports.facility_lookup import ( + FacilityKindValue, + FacilityStatusValue, +) from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId @@ -66,8 +70,8 @@ def _site_parent_lookup( *, facility_id: FacilityId = _PARENT_FACILITY_ID, code: FacilityCode = _DEFAULT_PARENT_LOOKUP_CODE, - kind: str = FacilityKind.SITE.value, - status: str = FacilityStatus.ACTIVE.value, + kind: FacilityKindValue = FacilityKind.SITE.value, + status: FacilityStatusValue = FacilityStatus.ACTIVE.value, ) -> FacilityLookupResult: """Test helper: build a Site-tier parent FacilityLookupResult. diff --git a/apps/api/tests/unit/federation/test_rotate_seal_online_key_decider.py b/apps/api/tests/unit/federation/test_rotate_seal_online_key_decider.py index a9a7c7f2cc8..dca5bb25c01 100644 --- a/apps/api/tests/unit/federation/test_rotate_seal_online_key_decider.py +++ b/apps/api/tests/unit/federation/test_rotate_seal_online_key_decider.py @@ -46,7 +46,10 @@ ) from cora.federation.features import rotate_seal_online_key from cora.federation.features.rotate_seal_online_key import RotateSealOnlineKey -from cora.infrastructure.ports.credential_lookup import CredentialLookupResult +from cora.infrastructure.ports.credential_lookup import ( + CredentialLookupResult, + CredentialStatusValue, +) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId @@ -113,7 +116,7 @@ def _credential( credential_id: UUID = _NEW_ONLINE_KEY, *, purpose: str = CredentialPurpose.SEAL_ONLINE_SIGNING.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, facility_id: str = _FACILITY_CODE, ) -> CredentialLookupResult: return CredentialLookupResult( diff --git a/apps/api/tests/unit/federation/test_rotate_seal_online_key_handler.py b/apps/api/tests/unit/federation/test_rotate_seal_online_key_handler.py index 2b8f9cba35e..d3cb6ceeadd 100644 --- a/apps/api/tests/unit/federation/test_rotate_seal_online_key_handler.py +++ b/apps/api/tests/unit/federation/test_rotate_seal_online_key_handler.py @@ -42,6 +42,7 @@ InMemoryFacilityLookup, ) from cora.infrastructure.kernel import Kernel +from cora.infrastructure.ports.credential_lookup import CredentialStatusValue from cora.shared.facility_code import FacilityCode from tests.unit._helpers import build_deps as _build_deps_shared from tests.unit.federation._helpers import ( @@ -74,7 +75,7 @@ def _build_lookup( *, register_default: bool = True, purpose: str = CredentialPurpose.SEAL_ONLINE_SIGNING.value, - status: str = CredentialStatus.ACTIVE.value, + status: CredentialStatusValue = CredentialStatus.ACTIVE.value, new_online_credential_id: UUID = _NEW_ONLINE_KEY, ) -> InMemoryCredentialLookup: """Build an `InMemoryCredentialLookup` seeded for the happy path. diff --git a/apps/api/tests/unit/operation/test_start_procedure_beam_gate_decider.py b/apps/api/tests/unit/operation/test_start_procedure_beam_gate_decider.py index 19d23dd4734..83b06924b14 100644 --- a/apps/api/tests/unit/operation/test_start_procedure_beam_gate_decider.py +++ b/apps/api/tests/unit/operation/test_start_procedure_beam_gate_decider.py @@ -236,7 +236,7 @@ def test_not_required_does_not_relax_the_enclosure_gate() -> None: EnclosureLookupResult( enclosure_id=UUID("00000000-0000-0000-0000-0000000000cc"), name="2-BM-B", - permit_status="Denied", + permit_status="NotPermitted", lifecycle="Active", permit_status_changed_at=_NOW.isoformat(), source_kind="EpicsPv", diff --git a/apps/api/tests/unit/operation/test_start_procedure_enclosure_gate_decider.py b/apps/api/tests/unit/operation/test_start_procedure_enclosure_gate_decider.py index afd42352519..1c75bdd3d69 100644 --- a/apps/api/tests/unit/operation/test_start_procedure_enclosure_gate_decider.py +++ b/apps/api/tests/unit/operation/test_start_procedure_enclosure_gate_decider.py @@ -23,7 +23,11 @@ import pytest -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosureLookupResult, + EnclosurePermitStatusValue, +) from cora.operation.aggregates.procedure import ( Procedure, ProcedureEnclosureCoverageMismatchError, @@ -39,8 +43,8 @@ def _enclosure_ref( *, - permit_status: str = "Permitted", - lifecycle: str = "Active", + permit_status: EnclosurePermitStatusValue = "Permitted", + lifecycle: EnclosureLifecycleValue = "Active", ) -> EnclosureLookupResult: return EnclosureLookupResult( enclosure_id=uuid4(), @@ -106,7 +110,9 @@ def test_decide_passes_when_every_referencing_enclosure_is_permitted_and_active( @pytest.mark.unit @pytest.mark.parametrize("permit_status", ["NotPermitted", "Unknown"]) -def test_decide_raises_requires_permitted_when_every_row_fails(permit_status: str) -> None: +def test_decide_raises_requires_permitted_when_every_row_fails( + permit_status: EnclosurePermitStatusValue, +) -> None: """Every referencing row fails -> ProcedureRequiresPermittedEnclosureError.""" only = _enclosure_ref(permit_status=permit_status) procedure = _procedure() diff --git a/apps/api/tests/unit/operation/test_start_procedure_handler.py b/apps/api/tests/unit/operation/test_start_procedure_handler.py index a5a9d1f57b5..41f82abb8b3 100644 --- a/apps/api/tests/unit/operation/test_start_procedure_handler.py +++ b/apps/api/tests/unit/operation/test_start_procedure_handler.py @@ -26,7 +26,9 @@ from cora.infrastructure.ports.asset_lookup import ( ANCESTOR_WALK_DEPTH_CAP, AncestorWalkDepthExceededError, + AssetLifecycleValue, AssetLookupResult, + AssetTierValue, ) from cora.infrastructure.ports.beam_availability_lookup import BeamAvailabilityLookupResult from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult @@ -326,8 +328,8 @@ async def test_handler_propagates_causation_id() -> None: def _result( asset_id: UUID, - lifecycle: str, - tier: str = "Unit", + lifecycle: AssetLifecycleValue, + tier: AssetTierValue = "Unit", located_in_enclosure_id: UUID | None = None, ) -> AssetLookupResult: return AssetLookupResult( diff --git a/apps/api/tests/unit/operation/test_start_procedure_supply_gate_decider.py b/apps/api/tests/unit/operation/test_start_procedure_supply_gate_decider.py index 668980eff83..78090a55451 100644 --- a/apps/api/tests/unit/operation/test_start_procedure_supply_gate_decider.py +++ b/apps/api/tests/unit/operation/test_start_procedure_supply_gate_decider.py @@ -19,7 +19,7 @@ import pytest -from cora.infrastructure.ports.supply_lookup import SupplyLookupResult +from cora.infrastructure.ports.supply_lookup import SupplyLookupResult, SupplyStatusValue from cora.operation.aggregates.procedure import ( Procedure, ProcedureName, @@ -33,7 +33,7 @@ _NOW = datetime(2026, 5, 28, 12, 0, 0, tzinfo=UTC) -def _ref(kind: str, status: str) -> SupplyLookupResult: +def _ref(kind: str, status: SupplyStatusValue) -> SupplyLookupResult: return SupplyLookupResult( supply_id=uuid4(), kind=kind, @@ -114,7 +114,9 @@ def test_decide_raises_requires_available_when_kind_absent_from_satisfaction() - @pytest.mark.unit @pytest.mark.parametrize("status", ["Unknown", "Degraded", "Unavailable", "Recovering"]) -def test_decide_raises_coverage_mismatch_when_no_supply_is_available(status: str) -> None: +def test_decide_raises_coverage_mismatch_when_no_supply_is_available( + status: SupplyStatusValue, +) -> None: """Kind exists in satisfaction but none AVAILABLE -> ProcedureSupplyCoverageMismatchError.""" proc = _procedure() only_supply = _ref("LiquidNitrogen", status) diff --git a/apps/api/tests/unit/run/test_record_witnessed_run_decider.py b/apps/api/tests/unit/run/test_record_witnessed_run_decider.py index f0a7dadfe32..21bd36cc9b1 100644 --- a/apps/api/tests/unit/run/test_record_witnessed_run_decider.py +++ b/apps/api/tests/unit/run/test_record_witnessed_run_decider.py @@ -20,7 +20,11 @@ ) from cora.infrastructure.ports.beam_availability_lookup import BeamAvailabilityLookupResult from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosureLookupResult, + EnclosurePermitStatusValue, +) from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import ( CapturePreconditionBypassSnapshot, @@ -128,7 +132,9 @@ def _beam( ) -def _enclosure(permit_status: str, lifecycle: str) -> EnclosureLookupResult: +def _enclosure( + permit_status: EnclosurePermitStatusValue, lifecycle: EnclosureLifecycleValue +) -> EnclosureLookupResult: return EnclosureLookupResult( enclosure_id=uuid4(), name="2-BM-B", diff --git a/apps/api/tests/unit/run/test_record_witnessed_run_decider_properties.py b/apps/api/tests/unit/run/test_record_witnessed_run_decider_properties.py index d97d3713156..fd773ca43f6 100644 --- a/apps/api/tests/unit/run/test_record_witnessed_run_decider_properties.py +++ b/apps/api/tests/unit/run/test_record_witnessed_run_decider_properties.py @@ -17,14 +17,14 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, get_args from uuid import UUID import pytest from hypothesis import given from hypothesis import strategies as st -from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult +from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult, ClearanceStatusValue from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import ( CapturePreconditionBypassSnapshot, @@ -49,6 +49,9 @@ _NAME = printable_ascii_text(min_size=1, max_size=200) _CAPTURE_CODE = printable_ascii_text(min_size=1, max_size=50) _MONITOR_SOURCE_ID = MonitorSourceId(UUID("01900000-0000-7000-8000-000063617001")) +_NON_ACTIVE_CLEARANCE_STATUSES = tuple( + value for value in get_args(ClearanceStatusValue) if value != "Active" +) _BYPASS_SNAPSHOTS = st.one_of( st.none(), st.builds( @@ -224,7 +227,7 @@ def test_witnessed_without_referencing_clearance_always_raises_requires_clearanc name=_NAME, plan_id=st.uuids(), capture_code=_CAPTURE_CODE, - clearance_status=st.text(min_size=1, max_size=20).filter(lambda s: s != "Active"), + clearance_status=st.sampled_from(_NON_ACTIVE_CLEARANCE_STATUSES), now=aware_datetimes(), new_id=st.uuids(), ) @@ -232,7 +235,7 @@ def test_witnessed_clearance_present_but_never_active_always_raises_coverage_mis name: str, plan_id: UUID, capture_code: str, - clearance_status: str, + clearance_status: ClearanceStatusValue, now: datetime, new_id: UUID, ) -> None: diff --git a/apps/api/tests/unit/run/test_safety_envelope.py b/apps/api/tests/unit/run/test_safety_envelope.py index 761a0f15fc5..c2516ca03db 100644 --- a/apps/api/tests/unit/run/test_safety_envelope.py +++ b/apps/api/tests/unit/run/test_safety_envelope.py @@ -20,9 +20,13 @@ import pytest from cora.infrastructure.ports.beam_availability_lookup import BeamAvailabilityLookupResult -from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult -from cora.infrastructure.ports.supply_lookup import SupplyLookupResult +from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult, ClearanceStatusValue +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosureLookupResult, + EnclosurePermitStatusValue, +) +from cora.infrastructure.ports.supply_lookup import SupplyLookupResult, SupplyStatusValue from cora.run.aggregates.run import ( RunBeamAvailabilityUnknownError, RunClearanceCoverageMismatchError, @@ -41,7 +45,7 @@ _RUN_ID = UUID("01900000-0000-7000-8000-0000000005a1") -def _clearance(status: str) -> ClearanceLookupResult: +def _clearance(status: ClearanceStatusValue) -> ClearanceLookupResult: return ClearanceLookupResult( clearance_id=uuid4(), status=status, @@ -51,7 +55,7 @@ def _clearance(status: str) -> ClearanceLookupResult: ) -def _supply(status: str) -> SupplyLookupResult: +def _supply(status: SupplyStatusValue) -> SupplyLookupResult: return SupplyLookupResult( supply_id=uuid4(), kind="LN2", @@ -61,7 +65,9 @@ def _supply(status: str) -> SupplyLookupResult: ) -def _enclosure(permit_status: str, lifecycle: str) -> EnclosureLookupResult: +def _enclosure( + permit_status: EnclosurePermitStatusValue, lifecycle: EnclosureLifecycleValue +) -> EnclosureLookupResult: return EnclosureLookupResult( enclosure_id=uuid4(), name="2-BM-A", diff --git a/apps/api/tests/unit/run/test_start_run_clearance_gate_decider.py b/apps/api/tests/unit/run/test_start_run_clearance_gate_decider.py index 72e503bb99f..8a7e1016fc1 100644 --- a/apps/api/tests/unit/run/test_start_run_clearance_gate_decider.py +++ b/apps/api/tests/unit/run/test_start_run_clearance_gate_decider.py @@ -23,7 +23,7 @@ AssetName, AssetTier, ) -from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult +from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult, ClearanceStatusValue from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import ( RunClearanceCoverageMismatchError, @@ -74,7 +74,7 @@ def _context( return context, frozenset({cap}) -def _ref(status: str) -> ClearanceLookupResult: +def _ref(status: ClearanceStatusValue) -> ClearanceLookupResult: return ClearanceLookupResult( clearance_id=uuid4(), status=status, @@ -112,7 +112,9 @@ def test_decide_raises_requires_active_when_no_clearance_references_the_run() -> "status", ["Defined", "Submitted", "UnderReview", "Approved", "Expired", "Rejected", "Superseded"], ) -def test_decide_raises_coverage_mismatch_when_no_clearance_is_active(status: str) -> None: +def test_decide_raises_coverage_mismatch_when_no_clearance_is_active( + status: ClearanceStatusValue, +) -> None: """Clearances reference the Run but none Active -> CoverageMismatch error.""" context, needs = _context(referencing_clearances=(_ref(status),)) new_id = uuid4() diff --git a/apps/api/tests/unit/run/test_start_run_enclosure_gate_decider.py b/apps/api/tests/unit/run/test_start_run_enclosure_gate_decider.py index 3dc16265813..16b47f0a706 100644 --- a/apps/api/tests/unit/run/test_start_run_enclosure_gate_decider.py +++ b/apps/api/tests/unit/run/test_start_run_enclosure_gate_decider.py @@ -29,7 +29,11 @@ AssetTier, ) from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult -from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult +from cora.infrastructure.ports.enclosure_lookup import ( + EnclosureLifecycleValue, + EnclosureLookupResult, + EnclosurePermitStatusValue, +) from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import ( RunEnclosureCoverageMismatchError, @@ -44,8 +48,8 @@ def _enclosure_ref( *, - permit_status: str = "Permitted", - lifecycle: str = "Active", + permit_status: EnclosurePermitStatusValue = "Permitted", + lifecycle: EnclosureLifecycleValue = "Active", ) -> EnclosureLookupResult: return EnclosureLookupResult( enclosure_id=uuid4(), @@ -151,7 +155,9 @@ def test_decide_passes_when_every_referencing_enclosure_is_permitted_and_active( @pytest.mark.unit @pytest.mark.parametrize("permit_status", ["NotPermitted", "Unknown"]) -def test_decide_raises_requires_permitted_when_every_row_fails(permit_status: str) -> None: +def test_decide_raises_requires_permitted_when_every_row_fails( + permit_status: EnclosurePermitStatusValue, +) -> None: """Every referencing Enclosure fails -> RunRequiresPermittedEnclosureError. Parametrized over NotPermitted and Unknown to pin the default-strict diff --git a/apps/api/tests/unit/run/test_start_run_input_gate_decider.py b/apps/api/tests/unit/run/test_start_run_input_gate_decider.py index b098391b98a..9406e1b04fa 100644 --- a/apps/api/tests/unit/run/test_start_run_input_gate_decider.py +++ b/apps/api/tests/unit/run/test_start_run_input_gate_decider.py @@ -29,6 +29,7 @@ from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult from cora.infrastructure.ports.dataset_distribution_lookup import ( DatasetDistributionLookupResult, + DistributionStatusValue, ) from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import RunInputNotReachableError, RunInputNotVerifiedError @@ -40,7 +41,7 @@ def _distribution( - dataset_id: UUID, status: str, *, supply_id: UUID | None = None + dataset_id: UUID, status: DistributionStatusValue, *, supply_id: UUID | None = None ) -> DatasetDistributionLookupResult: return DatasetDistributionLookupResult( distribution_id=uuid4(), @@ -153,7 +154,9 @@ def test_decide_passes_when_input_has_a_verified_distribution() -> None: @pytest.mark.unit @pytest.mark.parametrize("status", ["Registered", "Stale"]) -def test_decide_raises_when_input_has_no_verified_distribution(status: str) -> None: +def test_decide_raises_when_input_has_no_verified_distribution( + status: DistributionStatusValue, +) -> None: """Input with only a non-Verified Distribution -> RunInputNotVerifiedError.""" dataset_id = uuid4() context, needs = _context( diff --git a/apps/api/tests/unit/run/test_start_run_supply_gate_decider.py b/apps/api/tests/unit/run/test_start_run_supply_gate_decider.py index 223d4ea7c01..13e6b71c623 100644 --- a/apps/api/tests/unit/run/test_start_run_supply_gate_decider.py +++ b/apps/api/tests/unit/run/test_start_run_supply_gate_decider.py @@ -26,7 +26,7 @@ AssetTier, ) from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult -from cora.infrastructure.ports.supply_lookup import SupplyLookupResult +from cora.infrastructure.ports.supply_lookup import SupplyLookupResult, SupplyStatusValue from cora.recipe.aggregates.plan import Plan, PlanName, PlanStatus from cora.run.aggregates.run import ( RunRequiresAvailableSupplyError, @@ -39,7 +39,7 @@ _NOW = datetime(2026, 5, 28, 12, 0, 0, tzinfo=UTC) -def _ref(kind: str, status: str) -> SupplyLookupResult: +def _ref(kind: str, status: SupplyStatusValue) -> SupplyLookupResult: return SupplyLookupResult( supply_id=uuid4(), kind=kind, @@ -162,7 +162,9 @@ def test_decide_raises_requires_available_when_kind_absent_from_satisfaction() - @pytest.mark.unit @pytest.mark.parametrize("status", ["Unknown", "Degraded", "Unavailable", "Recovering"]) -def test_decide_raises_coverage_mismatch_when_no_supply_is_available(status: str) -> None: +def test_decide_raises_coverage_mismatch_when_no_supply_is_available( + status: SupplyStatusValue, +) -> None: """Kind exists in satisfaction but none AVAILABLE -> RunSupplyCoverageMismatchError. Parametrized over every non-AVAILABLE non-Decommissioned status to diff --git a/apps/api/tests/unit/run/test_start_run_widened_asset_ids.py b/apps/api/tests/unit/run/test_start_run_widened_asset_ids.py index 6e2ab6d7847..94491bbd2e4 100644 --- a/apps/api/tests/unit/run/test_start_run_widened_asset_ids.py +++ b/apps/api/tests/unit/run/test_start_run_widened_asset_ids.py @@ -30,7 +30,9 @@ from cora.infrastructure.ports.asset_lookup import ( ANCESTOR_WALK_DEPTH_CAP, AncestorWalkDepthExceededError, + AssetLifecycleValue, AssetLookupResult, + AssetTierValue, ) from cora.infrastructure.ports.clearance_lookup import ClearanceLookupResult from cora.infrastructure.ports.enclosure_lookup import EnclosureLookupResult @@ -54,8 +56,8 @@ def _result( asset_id: UUID, - lifecycle: str, - tier: str = "Unit", + lifecycle: AssetLifecycleValue, + tier: AssetTierValue = "Unit", located_in_enclosure_id: UUID | None = None, ) -> AssetLookupResult: return AssetLookupResult( diff --git a/apps/api/tests/unit/safety/test_amend_clearance_decider.py b/apps/api/tests/unit/safety/test_amend_clearance_decider.py index bf3c575c38c..4d289e7a0d6 100644 --- a/apps/api/tests/unit/safety/test_amend_clearance_decider.py +++ b/apps/api/tests/unit/safety/test_amend_clearance_decider.py @@ -11,7 +11,10 @@ import pytest -from cora.infrastructure.ports.clearance_template_lookup import ClearanceTemplateLookupResult +from cora.infrastructure.ports.clearance_template_lookup import ( + ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, +) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.safety.aggregates.clearance import ( Clearance, @@ -59,7 +62,7 @@ def _template_lookup_result( facility_code: str = "aps", code: str = "ESAF", *, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> ClearanceTemplateLookupResult: return ClearanceTemplateLookupResult( diff --git a/apps/api/tests/unit/safety/test_amend_clearance_decider_properties.py b/apps/api/tests/unit/safety/test_amend_clearance_decider_properties.py index 852dc071871..2897a532f84 100644 --- a/apps/api/tests/unit/safety/test_amend_clearance_decider_properties.py +++ b/apps/api/tests/unit/safety/test_amend_clearance_decider_properties.py @@ -39,7 +39,10 @@ from hypothesis import given from hypothesis import strategies as st -from cora.infrastructure.ports.clearance_template_lookup import ClearanceTemplateLookupResult +from cora.infrastructure.ports.clearance_template_lookup import ( + ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, +) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.safety.aggregates.clearance import ( Clearance, @@ -90,7 +93,7 @@ def _template_lookup_result( facility_code: str = "aps", code: str = "ESAF", *, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> ClearanceTemplateLookupResult: return ClearanceTemplateLookupResult( @@ -233,7 +236,7 @@ def test_amend_missing_template_lookup_always_raises_template_not_found( def test_amend_non_active_template_always_raises_not_bindable( parent_id: UUID, new_id: UUID, - template_status: str, + template_status: ClearanceTemplateStatusValue, now: datetime, ) -> None: """A template that exists but is not Active refuses binding for the command's id.""" diff --git a/apps/api/tests/unit/safety/test_register_clearance_decider.py b/apps/api/tests/unit/safety/test_register_clearance_decider.py index 3479456ebc8..9e4fa46fc56 100644 --- a/apps/api/tests/unit/safety/test_register_clearance_decider.py +++ b/apps/api/tests/unit/safety/test_register_clearance_decider.py @@ -7,6 +7,7 @@ from cora.infrastructure.ports.clearance_template_lookup import ( ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, ) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.safety.aggregates.clearance import ( @@ -53,7 +54,7 @@ def _template_lookup_result( facility_code: str = "aps", code: str = "ESAF", *, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> ClearanceTemplateLookupResult: """Build a stub ClearanceTemplateLookupResult for decider tests.""" diff --git a/apps/api/tests/unit/safety/test_register_clearance_decider_properties.py b/apps/api/tests/unit/safety/test_register_clearance_decider_properties.py index bce03d42559..5203effca9a 100644 --- a/apps/api/tests/unit/safety/test_register_clearance_decider_properties.py +++ b/apps/api/tests/unit/safety/test_register_clearance_decider_properties.py @@ -35,6 +35,7 @@ from cora.infrastructure.ports.clearance_template_lookup import ( ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, ) from cora.infrastructure.ports.facility_lookup import FacilityLookupResult from cora.safety.aggregates.clearance import ( @@ -85,7 +86,7 @@ def _template_lookup_result( facility_code: str = "aps", code: str = "ESAF", *, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> ClearanceTemplateLookupResult: """Build a stub ClearanceTemplateLookupResult for decider tests.""" @@ -205,7 +206,7 @@ def test_register_without_template_lookup_always_raises_template_not_found( def test_register_with_non_active_template_always_raises_not_bindable( title: str, run_id: UUID, - template_status: str, + template_status: ClearanceTemplateStatusValue, now: datetime, new_id: UUID, ) -> None: diff --git a/apps/api/tests/unit/safety/test_version_clearance_template_decider.py b/apps/api/tests/unit/safety/test_version_clearance_template_decider.py index c932289a43a..1fe28f4597a 100644 --- a/apps/api/tests/unit/safety/test_version_clearance_template_decider.py +++ b/apps/api/tests/unit/safety/test_version_clearance_template_decider.py @@ -14,6 +14,7 @@ from cora.infrastructure.ports.clearance_template_lookup import ( ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, ) from cora.safety.aggregates.clearance_template import ( ClearanceTemplate, @@ -43,7 +44,7 @@ def _lookup_result( template_id: UUID, facility_code: str, version: int, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", code: str = "esaf", ) -> ClearanceTemplateLookupResult: return ClearanceTemplateLookupResult( diff --git a/apps/api/tests/unit/safety/test_version_clearance_template_handler.py b/apps/api/tests/unit/safety/test_version_clearance_template_handler.py index 66375c9adf4..8498cc2c710 100644 --- a/apps/api/tests/unit/safety/test_version_clearance_template_handler.py +++ b/apps/api/tests/unit/safety/test_version_clearance_template_handler.py @@ -28,6 +28,7 @@ from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.kernel import Kernel +from cora.infrastructure.ports.clearance_template_lookup import ClearanceTemplateStatusValue from cora.safety.aggregates.clearance_template import ( ClearanceTemplateActivated, ClearanceTemplateCannotVersionError, @@ -188,7 +189,7 @@ def _seed_parent_lookup( parent_id: UUID = _PARENT_TEMPLATE_ID, facility_code: str = _FACILITY_CODE, code: str = _PARENT_TEMPLATE_CODE, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", version: int = 1, ) -> InMemoryClearanceTemplateLookup: lookup = InMemoryClearanceTemplateLookup() diff --git a/apps/api/tests/unit/safety/test_version_clearance_template_self_supersede.py b/apps/api/tests/unit/safety/test_version_clearance_template_self_supersede.py index c16c799c1fa..e07dfb7f5fe 100644 --- a/apps/api/tests/unit/safety/test_version_clearance_template_self_supersede.py +++ b/apps/api/tests/unit/safety/test_version_clearance_template_self_supersede.py @@ -18,6 +18,7 @@ from cora.infrastructure.ports.clearance_template_lookup import ( ClearanceTemplateLookupResult, + ClearanceTemplateStatusValue, ) from cora.safety.aggregates.clearance_template import ( ClearanceTemplate, @@ -45,7 +46,7 @@ def _lookup_result( template_id: UUID, facility_code: str, version: int, - status: str = "Active", + status: ClearanceTemplateStatusValue = "Active", code: str = "esaf", ) -> ClearanceTemplateLookupResult: return ClearanceTemplateLookupResult( diff --git a/apps/api/tests/unit/supply/test_register_supply_decider.py b/apps/api/tests/unit/supply/test_register_supply_decider.py index 9290896169d..74f0f9f1a97 100644 --- a/apps/api/tests/unit/supply/test_register_supply_decider.py +++ b/apps/api/tests/unit/supply/test_register_supply_decider.py @@ -5,8 +5,16 @@ import pytest -from cora.infrastructure.ports.asset_lookup import AssetLookupResult -from cora.infrastructure.ports.facility_lookup import FacilityLookupResult +from cora.infrastructure.ports.asset_lookup import ( + AssetLifecycleValue, + AssetLookupResult, + AssetTierValue, +) +from cora.infrastructure.ports.facility_lookup import ( + FacilityKindValue, + FacilityLookupResult, + FacilityStatusValue, +) from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId from cora.supply.aggregates.supply import ( @@ -32,8 +40,8 @@ def _facility_lookup_result( *, - kind: str = "Site", - status: str = "Active", + kind: FacilityKindValue = "Site", + status: FacilityStatusValue = "Active", ) -> FacilityLookupResult: return FacilityLookupResult( id=_FACILITY_ID, @@ -48,8 +56,8 @@ def _asset_lookup_result( *, asset_id: UUID = _CONTAINING_ASSET_ID, name: str = "2-BM", - tier: str = "Unit", - lifecycle: str = "Active", + tier: AssetTierValue = "Unit", + lifecycle: AssetLifecycleValue = "Active", ) -> AssetLookupResult: return AssetLookupResult( id=asset_id, diff --git a/apps/api/tests/unit/supply/test_register_supply_handler.py b/apps/api/tests/unit/supply/test_register_supply_handler.py index fbe2019535a..158d6d6a5bd 100644 --- a/apps/api/tests/unit/supply/test_register_supply_handler.py +++ b/apps/api/tests/unit/supply/test_register_supply_handler.py @@ -22,6 +22,8 @@ InMemoryFacilityLookup, ) from cora.infrastructure.kernel import Kernel +from cora.infrastructure.ports.asset_lookup import AssetLifecycleValue, AssetTierValue +from cora.infrastructure.ports.facility_lookup import FacilityStatusValue from cora.shared.facility_code import FacilityCode from cora.shared.identity import ActorId from cora.supply.aggregates.supply import ( @@ -43,7 +45,9 @@ _CONTAINING_ASSET_ID = UUID("01900000-0000-7000-8000-000000000a55") -def _seeded_facility_lookup(*, code: str = "aps", status: str = "Active") -> InMemoryFacilityLookup: +def _seeded_facility_lookup( + *, code: str = "aps", status: FacilityStatusValue = "Active" +) -> InMemoryFacilityLookup: lookup = InMemoryFacilityLookup() lookup.register(facility_id=_FACILITY_ID, code=code, kind="Site", status=status) return lookup @@ -53,8 +57,8 @@ def _seeded_asset_lookup( *, asset_id: UUID = _CONTAINING_ASSET_ID, name: str = "2-BM", - tier: str = "Unit", - lifecycle: str = "Active", + tier: AssetTierValue = "Unit", + lifecycle: AssetLifecycleValue = "Active", ) -> InMemoryAssetLookup: lookup = InMemoryAssetLookup() lookup.register(asset_id=asset_id, name=name, tier=tier, lifecycle=lifecycle)