test(release): validate installed module artifacts
This commit is contained in:
84
tests/test_release_integration.py
Normal file
84
tests/test_release_integration.py
Normal file
@@ -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()
|
||||
380
tools/checks/release_integration.py
Normal file
380
tools/checks/release_integration.py
Normal file
@@ -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())
|
||||
Reference in New Issue
Block a user