From d9819c4aadaf94eadde072a2f14a77c8422f3fc7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 13:24:02 +0200 Subject: [PATCH] test(release): validate installed module artifacts --- tests/test_release_integration.py | 84 ++++++ tools/checks/release_integration.py | 380 ++++++++++++++++++++++++++++ 2 files changed, 464 insertions(+) create mode 100644 tests/test_release_integration.py create mode 100644 tools/checks/release_integration.py diff --git a/tests/test_release_integration.py b/tests/test_release_integration.py new file mode 100644 index 0000000..4caf9c8 --- /dev/null +++ b/tests/test_release_integration.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from tools.checks.release_integration import ( + SOURCE_COUPLED_CORE_TESTS, + InstalledModuleArtifact, + artifact_contract_issues, + filter_test_suite, + release_package_names, +) + + +def _named_test(test_id: str) -> unittest.TestCase: + class NamedTest(unittest.TestCase): + def id(self) -> str: + return test_id + + def runTest(self) -> None: + return None + + return NamedTest() + + +class ReleaseIntegrationTests(unittest.TestCase): + def test_release_requirement_parser_selects_only_immutable_module_refs(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + requirements = Path(temp_dir) / "requirements-release.txt" + requirements.write_text( + "\n".join( + ( + "../govoplan-core[server]", + "govoplan-idm @ git+ssh://git@example.test/add-ideas/govoplan-idm.git@v0.1.8", + "govoplan-campaign @ git+ssh://git@example.test/add-ideas/govoplan-campaign.git@v0.1.8", + "govoplan-idm @ git+ssh://git@example.test/add-ideas/govoplan-idm.git@v0.1.8", + "other-package==1.0", + ) + ), + encoding="utf-8", + ) + + self.assertEqual( + release_package_names(requirements), + ("govoplan-idm", "govoplan-campaign"), + ) + + def test_artifact_contracts_compare_distribution_entry_point_and_discovery_versions(self) -> None: + artifacts = ( + InstalledModuleArtifact("govoplan-access", "0.1.8", "access", "0.1.8"), + InstalledModuleArtifact("govoplan-idm", "0.1.8", "idm", "0.1.7"), + InstalledModuleArtifact("govoplan-shadow-idm", "0.1.8", "idm", "0.1.8"), + ) + + issues = artifact_contract_issues( + artifacts, + release_packages=("govoplan-access", "govoplan-idm", "govoplan-shadow-idm", "govoplan-campaign"), + available_manifest_versions={"access": "0.1.8", "idm": "0.1.6"}, + ) + + self.assertEqual(len(issues), 5) + self.assertTrue(any("metadata version '0.1.8' does not match" in issue for issue in issues)) + self.assertTrue(any("does not publish a govoplan.modules entry point" in issue for issue in issues)) + self.assertTrue(any("multiple release packages" in issue for issue in issues)) + + def test_release_suite_excludes_only_named_source_coupled_tests(self) -> None: + retained_id = "tests.test_core_events.CoreEventTests.test_event_delivery" + nested = unittest.TestSuite( + ( + unittest.TestSuite(_named_test(test_id) for test_id in SOURCE_COUPLED_CORE_TESTS), + unittest.TestSuite((_named_test(retained_id),)), + ) + ) + + filtered, removed = filter_test_suite(nested, tuple(SOURCE_COUPLED_CORE_TESTS)) + + self.assertEqual(set(removed), set(SOURCE_COUPLED_CORE_TESTS)) + self.assertEqual([test.id() for test in filtered], [retained_id]) + self.assertEqual(len(SOURCE_COUPLED_CORE_TESTS), 4) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/checks/release_integration.py b/tools/checks/release_integration.py new file mode 100644 index 0000000..51e427a --- /dev/null +++ b/tools/checks/release_integration.py @@ -0,0 +1,380 @@ +#!/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())