#!/usr/bin/env python3 from __future__ import annotations import argparse import importlib.metadata as metadata import os import re import sys import tempfile import unittest from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass from pathlib import Path CORE_RELEASE_TEST_MODULES = ( "tests.test_module_system", "tests.test_access_contracts", "tests.test_identity_organization_contracts", "tests.test_core_events", "tests.test_policy_contracts", ) # These tests intentionally compare the current source workspace with sibling # source repositories. The release gate instead installs independently tagged # wheels and verifies the equivalent contracts below from artifact metadata and # public module capability factories. SOURCE_COUPLED_CORE_TESTS: Mapping[str, str] = { "tests.test_module_system.ModuleSystemTests.test_discovers_installed_core_and_product_module_manifests": ( "asserts the current Campaign source manifest rather than the independently tagged artifact" ), "tests.test_module_system.ModuleSystemTests.test_module_manifest_versions_match_source_project_versions": ( "requires sibling source pyproject.toml files that wheels do not contain" ), "tests.test_module_system.ModuleSystemTests.test_release_catalog_generator_reads_manifest_interface_contracts": ( "requires the sibling meta/source workspace and is retained by the normal source gate" ), "tests.test_identity_organization_contracts.IdentityOrganizationContractTests." "test_sql_identity_and_organization_directories_return_dtos": ( "directly constructs the current IDM source implementation instead of using its public capability factory" ), } _RELEASE_REQUIREMENT_PATTERN = re.compile( r"^(govoplan-[a-z0-9-]+)\s+@\s+git\+ssh://.+?\.git@v[0-9]+\.[0-9]+\.[0-9]+$" ) @dataclass(frozen=True, slots=True) class InstalledModuleArtifact: package_name: str package_version: str manifest_id: str manifest_version: str def release_package_names(path: Path) -> tuple[str, ...]: names: list[str] = [] for line in path.read_text(encoding="utf-8").splitlines(): match = _RELEASE_REQUIREMENT_PATTERN.match(line.strip()) if match: names.append(match.group(1)) return tuple(dict.fromkeys(names)) def artifact_contract_issues( artifacts: Sequence[InstalledModuleArtifact], *, release_packages: Sequence[str], available_manifest_versions: Mapping[str, str], ) -> tuple[str, ...]: issues: list[str] = [] artifacts_by_package: dict[str, list[InstalledModuleArtifact]] = {} artifacts_by_module: dict[str, list[InstalledModuleArtifact]] = {} for artifact in artifacts: artifacts_by_package.setdefault(artifact.package_name, []).append(artifact) artifacts_by_module.setdefault(artifact.manifest_id, []).append(artifact) if artifact.package_version != artifact.manifest_version: issues.append( f"{artifact.package_name} metadata version {artifact.package_version!r} does not match " f"manifest {artifact.manifest_id!r} version {artifact.manifest_version!r}" ) available_version = available_manifest_versions.get(artifact.manifest_id) if available_version is None: issues.append(f"Installed manifest {artifact.manifest_id!r} is not discoverable through Core") elif available_version != artifact.manifest_version: issues.append( f"Discovered manifest {artifact.manifest_id!r} version {available_version!r} does not match " f"its {artifact.package_name} entry point version {artifact.manifest_version!r}" ) for package_name in release_packages: if package_name not in artifacts_by_package: issues.append(f"{package_name} does not publish a govoplan.modules entry point") for module_id, providers in artifacts_by_module.items(): if len(providers) > 1: packages = ", ".join(sorted(item.package_name for item in providers)) issues.append(f"Module id {module_id!r} is published by multiple release packages: {packages}") return tuple(issues) def iter_test_cases(suite: unittest.TestSuite) -> Iterator[unittest.TestCase]: for test in suite: if isinstance(test, unittest.TestSuite): yield from iter_test_cases(test) else: yield test def filter_test_suite( suite: unittest.TestSuite, excluded_ids: Sequence[str], ) -> tuple[unittest.TestSuite, tuple[str, ...]]: excluded = set(excluded_ids) kept: list[unittest.TestCase] = [] removed: list[str] = [] for test in iter_test_cases(suite): test_id = test.id() if test_id in excluded: removed.append(test_id) else: kept.append(test) return unittest.TestSuite(kept), tuple(removed) def _collect_installed_artifacts( package_names: Sequence[str], ) -> tuple[tuple[InstalledModuleArtifact, ...], tuple[str, ...]]: artifacts: list[InstalledModuleArtifact] = [] issues: list[str] = [] for package_name in package_names: try: distribution = metadata.distribution(package_name) except metadata.PackageNotFoundError: issues.append(f"Release package is not installed: {package_name}") continue entry_points = tuple( entry_point for entry_point in distribution.entry_points if entry_point.group == "govoplan.modules" ) if not entry_points: continue for entry_point in entry_points: try: factory = entry_point.load() manifest = factory() manifest_id = str(manifest.id) manifest_version = str(manifest.version) except Exception as exc: issues.append(f"Could not load {package_name} module entry point {entry_point.name!r}: {exc}") continue artifacts.append( InstalledModuleArtifact( package_name=package_name, package_version=distribution.version, manifest_id=manifest_id, manifest_version=manifest_version, ) ) return tuple(artifacts), tuple(issues) def _require_equal(actual: object, expected: object, message: str) -> None: if actual != expected: raise RuntimeError(f"{message}: expected {expected!r}, got {actual!r}") def _run_directory_capability_smoke(registry: object) -> None: from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, IdentityDirectory from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory from govoplan_core.core.modules import ModuleContext from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory from govoplan_core.db.base import Base from govoplan_core.db.session import configure_database, reset_database, set_database from govoplan_core.tenancy.scope import create_scope_tables from govoplan_identity.backend.db.models import Identity, IdentityAccountLink from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment from govoplan_organizations.backend.db.models import ( OrganizationFunction, OrganizationFunctionType, OrganizationUnit, OrganizationUnitType, ) from govoplan_tenancy.backend.db.models import Tenant registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) identity_directory = registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY) organization_directory = registry.require_capability(CAPABILITY_ORGANIZATION_DIRECTORY) idm_directory = registry.require_capability(CAPABILITY_IDM_DIRECTORY) if not isinstance(identity_directory, IdentityDirectory): raise RuntimeError("Installed Identity capability does not satisfy IdentityDirectory") if not isinstance(organization_directory, OrganizationDirectory): raise RuntimeError("Installed Organizations capability does not satisfy OrganizationDirectory") if not isinstance(idm_directory, IdmDirectory): raise RuntimeError("Installed IDM capability does not satisfy IdmDirectory") with tempfile.TemporaryDirectory(prefix="govoplan-release-directory-") as temp_dir: database = configure_database( f"sqlite:///{Path(temp_dir) / 'directory.sqlite3'}", dispose_previous=True, ) set_database(database) try: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant( id="release-tenant", slug="release-tenant", name="Release Tenant", settings={}, ) identity = Identity( id="release-identity", display_name="Release Person", source="local", ) link = IdentityAccountLink( identity_id=identity.id, account_id="release-account", is_primary=True, ) unit_type = OrganizationUnitType( id="release-unit-type", tenant_id=tenant.id, slug="office", name="Office", ) unit = OrganizationUnit( id="release-unit", tenant_id=tenant.id, unit_type_id=unit_type.id, slug="registry", name="Registry", ) function_type = OrganizationFunctionType( id="release-function-type", tenant_id=tenant.id, slug="clerk", name="Clerk", organization_unit_type_id=unit_type.id, ) function = OrganizationFunction( id="release-function", tenant_id=tenant.id, function_type_id=function_type.id, organization_unit_id=unit.id, slug="registry-clerk", name="Registry Clerk", ) assignment = IdmOrganizationFunctionAssignment( id="release-assignment", tenant_id=tenant.id, identity_id=identity.id, function_id=function.id, organization_unit_id=unit.id, applies_to_subunits=True, ) session.add_all((tenant, identity, link, unit_type, unit, function_type, function, assignment)) session.commit() identity_ref = identity_directory.identity_for_account("release-account") _require_equal( identity_ref.id if identity_ref is not None else None, "release-identity", "Identity capability account projection", ) function_ref = organization_directory.get_function("release-function") _require_equal( function_ref.name if function_ref is not None else None, "Registry Clerk", "Organizations capability function projection", ) identity_assignments = idm_directory.organization_function_assignments_for_identity( "release-identity", tenant_id="release-tenant", ) account_assignments = idm_directory.organization_function_assignments_for_account( "release-account", tenant_id="release-tenant", ) _require_equal( tuple(item.id for item in identity_assignments), ("release-assignment",), "IDM identity assignment projection", ) _require_equal( tuple(item.id for item in account_assignments), ("release-assignment",), "IDM account assignment projection", ) finally: reset_database(dispose=True) def run_installed_artifact_checks(requirements_path: Path) -> int: from govoplan_core.server.registry import available_module_manifests, build_platform_registry package_names = release_package_names(requirements_path) if not package_names: print(f"No tagged GovOPlaN packages found in {requirements_path}", file=sys.stderr) return 1 artifacts, collection_issues = _collect_installed_artifacts(package_names) available = available_module_manifests() contract_issues = artifact_contract_issues( artifacts, release_packages=package_names, available_manifest_versions={module_id: str(manifest.version) for module_id, manifest in available.items()}, ) issues = (*collection_issues, *contract_issues) if issues: print("\n".join(issues), file=sys.stderr) return 1 module_ids = tuple(dict.fromkeys(artifact.manifest_id for artifact in artifacts)) registry = build_platform_registry(module_ids) snapshot = registry.validate() discovered_ids = {manifest.id for manifest in snapshot.manifests} if discovered_ids != set(module_ids): print( "Release registry module mismatch: " f"expected {', '.join(sorted(module_ids))}; got {', '.join(sorted(discovered_ids))}", file=sys.stderr, ) return 1 _run_directory_capability_smoke(registry) versions = ", ".join( f"{artifact.manifest_id}={artifact.manifest_version}" for artifact in sorted(artifacts, key=lambda item: item.manifest_id) ) print(f"Installed release artifact contracts passed: {versions}") return 0 def run_core_release_tests(core_root: Path) -> int: resolved_root = core_root.resolve() os.chdir(resolved_root) sys.path.insert(0, str(resolved_root)) loader = unittest.defaultTestLoader loaded = loader.loadTestsFromNames(CORE_RELEASE_TEST_MODULES) suite, removed = filter_test_suite(loaded, tuple(SOURCE_COUPLED_CORE_TESTS)) missing_exclusions = set(SOURCE_COUPLED_CORE_TESTS) - set(removed) if missing_exclusions: print( "Release source-test exclusions no longer match the Core suite: " + ", ".join(sorted(missing_exclusions)), file=sys.stderr, ) return 1 for test_id in removed: print(f"Artifact contract replaces source-coupled test: {test_id} ({SOURCE_COUPLED_CORE_TESTS[test_id]})") result = unittest.TextTestRunner(verbosity=1).run(suite) return 0 if result.wasSuccessful() else 1 def _parser() -> argparse.ArgumentParser: meta_root = Path(__file__).resolve().parents[2] parser = argparse.ArgumentParser(description="Artifact-aware GovOPlaN release integration checks") subparsers = parser.add_subparsers(dest="command", required=True) artifact_parser = subparsers.add_parser("artifacts", help="Validate installed module artifacts and capabilities") artifact_parser.add_argument( "--requirements", type=Path, default=meta_root / "requirements-release.txt", ) core_parser = subparsers.add_parser("core-tests", help="Run Core tests that are valid for tagged module artifacts") core_parser.add_argument("--core-root", type=Path, required=True) return parser def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) if args.command == "artifacts": return run_installed_artifact_checks(args.requirements) return run_core_release_tests(args.core_root) if __name__ == "__main__": raise SystemExit(main())