refactor: resolve sonar complexity and length violations in package modules

implementations/python/packages/raes/composition/_expand.py

@@ -45,9 +45,10 @@
     SDLParserLimits,
     SDLSourceParseOptions,
 )
-from ..instantiate import _bind_scenario_content
+from ..instantiate import _bind_scenario_content, _BoundScenarioResult
 from ..module_registry import (
     Lockfile,
+    ResolvedModule,
     TrustPolicy,
     _VerifiedSourceBundle,
     load_lockfile,
@@ -327,6 +328,25 @@ def _expand_one_import(
     context.budget.check_namespaces(namespaced_payload, path=import_path)
     merged = _merge_sections(merged, namespaced_payload, path=import_path)
 
+    return (merged, *_import_provenance_additions(resolved_import, import_decl, bound, inner_provenance, symbols))
+
+
+def _import_provenance_additions(
+    resolved_import: ResolvedModule,
+    import_decl: ImportDecl,
+    bound: _BoundScenarioResult,
+    inner_provenance: ExpansionProvenance,
+    symbols: dict[str, dict[str, str] | set[str]],
+) -> tuple[
+    list[ResolvedImportProvenance],
+    list[CapabilityConstraint],
+    list[ExplicitnessProvenanceRecord],
+    list[RealizationDesignationRecord],
+    list[RealizationConstraintRecord],
+]:
+    """Build the namespaced provenance additions contributed by one import."""
+
+    namespace = import_decl.namespace
     import_records: list[ResolvedImportProvenance] = [
         _resolved_import_record(resolved_import, requested=import_decl, bindings=bound)
     ]
@@ -349,7 +369,6 @@ def _expand_one_import(
         for record in inner_provenance.realization_constraints
     ]
     return (
-        merged,
         import_records,
         capability_constraints,
         explicitness_records,

implementations/python/packages/raes/phase_contracts.py

@@ -153,6 +153,26 @@ def _validate_record(self) -> ResolvedImportProvenance:
         return self
 
 
+def _is_ordinary_constraint_pointer(parts: list[str]) -> bool:
+    return len(parts) == 4 and (parts[1], parts[3]) in {
+        ("nodes", "os"),
+        ("nodes", "os_distribution"),
+        ("nodes", "os_version"),
+        ("nodes", "architecture"),
+        ("infrastructure", "count"),
+    }
+
+
+def _is_process_limit_constraint_pointer(parts: list[str]) -> bool:
+    return (
+        len(parts) == 9
+        and parts[1] == "nodes"
+        and parts[3:7] == ["runtime", "operational_policy", "resource_limits", "process_limits"]
+        and parts[7].isdigit()
+        and parts[8] in {"soft", "hard"}
+    )
+
+
 class CapabilityConstraint(FrozenPhaseModel):
     """Pre-instantiation constraint retained for one concrete field."""
 
@@ -169,21 +189,7 @@ def _validate_constraint(self) -> CapabilityConstraint:
         if not self.field_pointer.startswith("/") or _JSON_POINTER_RE.fullmatch(self.field_pointer) is None:
             raise ValueError("field_pointer must be a non-root RFC 6901 JSON Pointer")
         parts = self.field_pointer.split("/")
-        ordinary_pointer = len(parts) == 4 and (parts[1], parts[3]) in {
-            ("nodes", "os"),
-            ("nodes", "os_distribution"),
-            ("nodes", "os_version"),
-            ("nodes", "architecture"),
-            ("infrastructure", "count"),
-        }
-        process_limit_pointer = (
-            len(parts) == 9
-            and parts[1] == "nodes"
-            and parts[3:7] == ["runtime", "operational_policy", "resource_limits", "process_limits"]
-            and parts[7].isdigit()
-            and parts[8] in {"soft", "hard"}
-        )
-        if not ordinary_pointer and not process_limit_pointer:
+        if not _is_ordinary_constraint_pointer(parts) and not _is_process_limit_constraint_pointer(parts):
             raise ValueError(
                 "field_pointer must address /nodes/<id>/(os|os_distribution|os_version|architecture), "
                 "/infrastructure/<id>/count, or "

implementations/python/packages/raes_backend_libvirt/capability_envelope.py

@@ -186,25 +186,35 @@ def _out_of_envelope_terms(
             if term and term not in supported:
                 yield (dimension.code, address, term), _envelope_diagnostic(dimension, address, term)
     if resource_type == NODE_RESOURCE_TYPE:
-        family = _os_family(payload)
-        distribution = _os_distribution(payload)
-        version = _os_version(payload) or None
-        if distribution and not capabilities.supports_operating_system(
-            family=family,
-            distribution=distribution,
-            version=version,
-        ):
-            identity = "/".join((family, distribution, version or "<open-release>"))
-            yield (
-                (_CODE_UNSUPPORTED_OPERATING_SYSTEM, address, identity),
-                Diagnostic(
-                    code=_CODE_UNSUPPORTED_OPERATING_SYSTEM,
-                    domain=_DOMAIN,
-                    address=address,
-                    message=f"Libvirt backend does not realize operating-system identity '{identity}'.",
-                    severity=Severity.ERROR,
-                ),
-            )
+        yield from _out_of_envelope_operating_system(address, payload, capabilities)
+
+
+def _out_of_envelope_operating_system(
+    address: str,
+    payload: Mapping[str, object],
+    capabilities: ProvisionerCapabilities,
+) -> Iterator[tuple[tuple[str, str, str], Diagnostic]]:
+    """Yield the unsupported operating-system identity of a node payload, if any."""
+
+    family = _os_family(payload)
+    distribution = _os_distribution(payload)
+    version = _os_version(payload) or None
+    if distribution and not capabilities.supports_operating_system(
+        family=family,
+        distribution=distribution,
+        version=version,
+    ):
+        identity = "/".join((family, distribution, version or "<open-release>"))
+        yield (
+            (_CODE_UNSUPPORTED_OPERATING_SYSTEM, address, identity),
+            Diagnostic(
+                code=_CODE_UNSUPPORTED_OPERATING_SYSTEM,
+                domain=_DOMAIN,
+                address=address,
+                message=f"Libvirt backend does not realize operating-system identity '{identity}'.",
+                severity=Severity.ERROR,
+            ),
+        )
 
 
 def _materialized_payloads(plan: ProvisioningPlan) -> Iterator[tuple[str, str, Mapping[str, object]]]:

implementations/python/packages/raes_backend_protocols/provisioner_capabilities.py

@@ -45,6 +45,40 @@ def _require_string_values(name: str, values: frozenset[str], *, required: bool
         raise ValueError(f"ProvisionerCapabilities.{name} must not contain empty strings")
 
 
+def _validate_operating_system_rows(capabilities: "ProvisionerCapabilities") -> None:
+    os_keys = [(entry.family, entry.distribution) for entry in capabilities.operating_systems]
+    if len(os_keys) != len(set(os_keys)):
+        raise ValueError(
+            "ProvisionerCapabilities.operating_systems must not contain duplicate family/distribution rows"
+        )
+    undeclared_families = {
+        entry.family
+        for entry in capabilities.operating_systems
+        if entry.family not in capabilities.supported_os_families
+    }
+    if undeclared_families:
+        raise ValueError(
+            "ProvisionerCapabilities.operating_systems families must be present in supported_os_families: "
+            + ", ".join(sorted(undeclared_families))
+        )
+
+
+def _validated_artifact_kinds(capabilities: "ProvisionerCapabilities") -> frozenset[GeneratedArtifactKind]:
+    try:
+        normalized = frozenset(GeneratedArtifactKind(kind) for kind in capabilities.supported_generated_artifact_kinds)
+    except ValueError as exc:
+        raise ValueError("ProvisionerCapabilities contains an unknown generated artifact kind") from exc
+    if capabilities.supports_generated_artifacts and not normalized:
+        raise ValueError(
+            "ProvisionerCapabilities that support generated artifacts must declare supported_generated_artifact_kinds"
+        )
+    if not capabilities.supports_generated_artifacts and normalized:
+        raise ValueError(
+            "ProvisionerCapabilities supported_generated_artifact_kinds require supports_generated_artifacts=True"
+        )
+    return normalized
+
+
 def _validate_account_support(capabilities: "ProvisionerCapabilities") -> None:
     if capabilities.supports_accounts and not capabilities.supported_account_features:
         raise ValueError("ProvisionerCapabilities that support accounts must declare supported_account_features")
@@ -92,19 +126,7 @@ def __post_init__(self) -> None:
             "capabilities.provisioner.supported_os_families",
             self.supported_os_families,
         )
-        os_keys = [(entry.family, entry.distribution) for entry in self.operating_systems]
-        if len(os_keys) != len(set(os_keys)):
-            raise ValueError(
-                "ProvisionerCapabilities.operating_systems must not contain duplicate family/distribution rows"
-            )
-        undeclared_families = {
-            entry.family for entry in self.operating_systems if entry.family not in self.supported_os_families
-        }
-        if undeclared_families:
-            raise ValueError(
-                "ProvisionerCapabilities.operating_systems families must be present in supported_os_families: "
-                + ", ".join(sorted(undeclared_families))
-            )
+        _validate_operating_system_rows(self)
         validate_controlled_vocabulary_scope_values(
             "capabilities.provisioner.supported_node_architectures",
             self.supported_node_architectures,
@@ -128,22 +150,7 @@ def __post_init__(self) -> None:
         if self.max_total_nodes is not None and self.max_total_nodes < 1:
             raise ValueError("ProvisionerCapabilities.max_total_nodes must be positive when provided")
         _validate_account_support(self)
-        try:
-            normalized_artifact_kinds = frozenset(
-                GeneratedArtifactKind(kind) for kind in self.supported_generated_artifact_kinds
-            )
-        except ValueError as exc:
-            raise ValueError("ProvisionerCapabilities contains an unknown generated artifact kind") from exc
-        object.__setattr__(self, "supported_generated_artifact_kinds", normalized_artifact_kinds)
-        if self.supports_generated_artifacts and not normalized_artifact_kinds:
-            raise ValueError(
-                "ProvisionerCapabilities that support generated artifacts must declare "
-                "supported_generated_artifact_kinds"
-            )
-        if not self.supports_generated_artifacts and normalized_artifact_kinds:
-            raise ValueError(
-                "ProvisionerCapabilities supported_generated_artifact_kinds require supports_generated_artifacts=True"
-            )
+        object.__setattr__(self, "supported_generated_artifact_kinds", _validated_artifact_kinds(self))
 
     def supports_operating_system(
         self,

implementations/python/packages/raes_backend_stubs/stubs.py

@@ -43,6 +43,34 @@
 ]
 
 
+def _applied_entries(
+    plan: ProvisioningPlan,
+    snapshot: RuntimeSnapshot,
+) -> tuple[dict[str, SnapshotEntry], list[str]]:
+    """Fold the plan's operations into snapshot entries and changed addresses."""
+
+    entries = dict(snapshot.entries)
+    changed_addresses: list[str] = []
+    for op in plan.operations:
+        if op.action == ChangeAction.DELETE:
+            entries.pop(op.address, None)
+            changed_addresses.append(op.address)
+            continue
+        status = "unchanged" if op.action == ChangeAction.UNCHANGED else "applied"
+        entries[op.address] = SnapshotEntry(
+            address=op.address,
+            domain=RuntimeDomain.PROVISIONING,
+            resource_type=op.resource_type,
+            payload=op.payload,
+            ordering_dependencies=op.ordering_dependencies,
+            refresh_dependencies=op.refresh_dependencies,
+            status=status,
+        )
+        if op.action != ChangeAction.UNCHANGED:
+            changed_addresses.append(op.address)
+    return entries, changed_addresses
+
+
 class StubProvisioner:
     """In-memory provisioner."""
 
@@ -74,25 +102,7 @@ def apply(
                     )
                 ],
             )
-        entries = dict(snapshot.entries)
-        changed_addresses: list[str] = []
-        for op in plan.operations:
-            if op.action == ChangeAction.DELETE:
-                entries.pop(op.address, None)
-                changed_addresses.append(op.address)
-                continue
-            status = "unchanged" if op.action == ChangeAction.UNCHANGED else "applied"
-            entries[op.address] = SnapshotEntry(
-                address=op.address,
-                domain=RuntimeDomain.PROVISIONING,
-                resource_type=op.resource_type,
-                payload=op.payload,
-                ordering_dependencies=op.ordering_dependencies,
-                refresh_dependencies=op.refresh_dependencies,
-                status=status,
-            )
-            if op.action != ChangeAction.UNCHANGED:
-                changed_addresses.append(op.address)
+        entries, changed_addresses = _applied_entries(plan, snapshot)
 
         if self._realization_envelope is None:
             return ApplyResult(

implementations/python/packages/raes_conformance/conformance/snapshot_semantics.py

@@ -38,9 +38,8 @@
 from raes_conformance.conformance.diagnostics import _SEMANTIC_INVALID_DIAGNOSTIC_CODE, _diagnostic
 
 
-def _snapshot_from_envelope(payload: dict[str, Any]) -> RuntimeSnapshot:
-    validated = RuntimeSnapshotEnvelopeModel.model_validate(payload)
-    entries = {
+def _snapshot_entries(validated: RuntimeSnapshotEnvelopeModel) -> dict[str, SnapshotEntry]:
+    return {
         address: SnapshotEntry(
             address=entry.address,
             domain=RuntimeDomain(entry.domain),
@@ -52,8 +51,44 @@ def _snapshot_from_envelope(payload: dict[str, Any]) -> RuntimeSnapshot:
         )
         for address, entry in validated.entries.items()
     }
+
+
+def _realization_disclosures(
+    validated: RuntimeSnapshotEnvelopeModel,
+) -> tuple[RealizationObservationDisclosure, ...]:
+    return tuple(
+        RealizationObservationDisclosure(
+            address=entry.address,
+            field_path=entry.field_path,
+            domain=entry.domain,
+            requirement_kind=entry.requirement_kind,
+            verification_scope=entry.verification_scope,
+            observation_strength=entry.observation_strength,
+            observed_value=entry.observed_value,
+            operating_system=(
+                ObservedOperatingSystemIdentity(
+                    family=entry.operating_system.family,
+                    distribution=entry.operating_system.distribution,
+                    version=entry.operating_system.version,
+                )
+                if entry.operating_system is not None
+                else None
+            ),
+            operation_id=entry.operation_id,
+            envelope_digest=entry.envelope_digest,
+            configuration_digest=entry.configuration_digest,
+            observer_version=entry.observer_version,
+            sequence=entry.sequence,
+            binding_verified=entry.binding_verified,
+        )
+        for entry in validated.realization_observations
+    )
+
+
+def _snapshot_from_envelope(payload: dict[str, Any]) -> RuntimeSnapshot:
+    validated = RuntimeSnapshotEnvelopeModel.model_validate(payload)
     return RuntimeSnapshot(
-        entries=entries,
+        entries=_snapshot_entries(validated),
         orchestration_results={
             address: result.model_dump(mode="json") for address, result in validated.orchestration_results.items()
         },
@@ -119,33 +154,7 @@ def _snapshot_from_envelope(payload: dict[str, Any]) -> RuntimeSnapshot:
             for context_id, context in validated.time_management_contexts.items()
         },
         time_model_state=validated.time_model_state,
-        realization_observations=tuple(
-            RealizationObservationDisclosure(
-                address=entry.address,
-                field_path=entry.field_path,
-                domain=entry.domain,
-                requirement_kind=entry.requirement_kind,
-                verification_scope=entry.verification_scope,
-                observation_strength=entry.observation_strength,
-                observed_value=entry.observed_value,
-                operating_system=(
-                    ObservedOperatingSystemIdentity(
-                        family=entry.operating_system.family,
-                        distribution=entry.operating_system.distribution,
-                        version=entry.operating_system.version,
-                    )
-                    if entry.operating_system is not None
-                    else None
-                ),
-                operation_id=entry.operation_id,
-                envelope_digest=entry.envelope_digest,
-                configuration_digest=entry.configuration_digest,
-                observer_version=entry.observer_version,
-                sequence=entry.sequence,
-                binding_verified=entry.binding_verified,
-            )
-            for entry in validated.realization_observations
-        ),
+        realization_observations=_realization_disclosures(validated),
         metadata=dict(validated.metadata),
     )