from __future__ import annotations import importlib.util from pathlib import Path import sys import tempfile import unittest META_ROOT = Path(__file__).resolve().parents[1] SCRIPT = META_ROOT / "tools" / "release" / "release-migration-audit.py" def load_audit_module(): spec = importlib.util.spec_from_file_location("release_migration_audit", SCRIPT) if spec is None or spec.loader is None: raise RuntimeError(f"Could not load {SCRIPT}") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class ReleaseMigrationAuditTests(unittest.TestCase): def test_typed_alembic_variables_are_parsed(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: path = Path(directory) / "1234_example.py" path.write_text( """ from __future__ import annotations from typing import Sequence, Union revision: str = "1234" down_revision: Union[str, None] = "base" depends_on: Union[str, None] = "core" branch_labels: Union[str, Sequence[str], None] = None """, encoding="utf-8", ) migration = audit.parse_migration_file("govoplan-core", path) self.assertIsNotNone(migration) self.assertEqual(migration.revision, "1234") self.assertEqual(migration.down_revisions, ("base",)) self.assertEqual(migration.depends_on, ("core",)) self.assertEqual(migration.branch_labels, ()) def test_development_wrapper_uses_release_revision_metadata(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) versions = root / "versions" development = root / "dev_versions" versions.mkdir() development.mkdir() release = versions / "1234_example.py" release.write_text( 'revision = "1234"\n' 'down_revision = "base"\n' 'depends_on = "core"\n' "branch_labels = None\n", encoding="utf-8", ) wrapper = development / release.name wrapper.write_text( "from importlib import import_module\n" '_migration = import_module("govoplan_core.backend.migrations.versions.1234_example")\n' "revision = _migration.revision\n" "down_revision = _migration.down_revision\n" "depends_on = _migration.depends_on\n" "branch_labels = _migration.branch_labels\n", encoding="utf-8", ) migration = audit.parse_migration_file("govoplan-core", wrapper) self.assertIsNotNone(migration) self.assertEqual("1234", migration.revision) self.assertEqual(("base",), migration.down_revisions) self.assertEqual(("core",), migration.depends_on) def test_literal_wrapper_alias_and_different_filename_are_resolved_without_execution(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) (root / "versions").mkdir() (root / "dev_versions").mkdir() (root / "versions/1234_v019_example.py").write_text( 'revision = "1234"\ndown_revision = "base"\ndepends_on = "core"\nbranch_labels = None\n' 'raise AssertionError("Migration implementation must not execute")\n', encoding="utf-8", ) wrapper = root / "dev_versions/1234_example.py" for alias in ("_migration", "edit_revision", "message_actions"): with self.subTest(alias=alias): wrapper.write_text( "from importlib import import_module as load_migration\n" f'{alias} = load_migration("govoplan_campaign.backend.migrations.versions." "1234_v019_example")\n' f"revision = {alias}.revision\ndown_revision = {alias}.down_revision\n" f"depends_on = {alias}.depends_on\nbranch_labels = {alias}.branch_labels\n" 'raise AssertionError("Wrapper must not execute")\n', encoding="utf-8", ) migration = audit.parse_migration_file("govoplan-campaign", wrapper) self.assertEqual(migration.revision, "1234") self.assertEqual(migration.down_revisions, ("base",)) self.assertEqual(migration.depends_on, ("core",)) def test_core_literal_sibling_file_wrapper_is_resolved_without_execution(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) (root / "versions").mkdir() (root / "dev_versions").mkdir() (root / "versions/1234_example.py").write_text( 'revision = "1234"\ndown_revision = None\ndepends_on = None\nbranch_labels = None\n' 'raise AssertionError("Migration implementation must not execute")\n', encoding="utf-8", ) wrapper = root / "dev_versions/1234_example.py" wrapper.write_text( "from importlib.util import module_from_spec, spec_from_file_location\n" "from pathlib import Path\n" '_path = Path(__file__).resolve().parents[1] / "versions" / "1234_example.py"\n' '_spec = spec_from_file_location("synthetic_migration", _path)\n' "_module = module_from_spec(_spec)\n_spec.loader.exec_module(_module)\n" "revision = _module.revision\ndown_revision = _module.down_revision\n" "depends_on = _module.depends_on\nbranch_labels = _module.branch_labels\n", encoding="utf-8", ) migration = audit.parse_migration_file("govoplan-core", wrapper) self.assertEqual(migration.revision, "1234") self.assertEqual(migration.down_revisions, ()) def test_wrapper_rejects_dynamic_foreign_missing_or_rebound_targets(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) (root / "versions").mkdir() (root / "dev_versions").mkdir() (root / "versions/1234_example.py").write_text('revision = "1234"\n', encoding="utf-8") wrapper = root / "dev_versions/1234_example.py" valid = '_migration = import_module("govoplan_campaign.backend.migrations.versions.1234_example")\n' definitions = ( '_migration = import_module(module_name)\n', '_migration = import_module("govoplan_mail.backend.migrations.versions.1234_example")\n', '_migration = import_module("govoplan_campaign.backend.migrations.versions...other.1234_example")\n', '_migration = import_module("govoplan_campaign.backend.migrations.versions.missing")\n', valid + "_migration = another_module\n", "import_module = another_loader\n" + valid, "def import_module(value):\n return another_module\n" + valid, "import another_loader as import_module\n" + valid, valid + "class _migration:\n revision = 'another'\n", "if condition:\n _migration = another_module\n" + valid, valid + "del _migration\n", ) for definition in definitions: with self.subTest(definition=definition): wrapper.write_text( "from importlib import import_module\n" + definition + "revision = _migration.revision\n", encoding="utf-8", ) with self.assertRaisesRegex(ValueError, "unsupported or ambiguous"): audit.parse_migration_file("govoplan-campaign", wrapper) def test_wrapper_rejects_symlink_and_mixed_release_metadata(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) (root / "versions").mkdir() (root / "dev_versions").mkdir() (root / "versions/1234_example.py").write_text('revision = "1234"\ndown_revision = None\n', encoding="utf-8") (root / "versions/5678_example.py").write_text('revision = "5678"\ndown_revision = None\n', encoding="utf-8") (root / "versions/linked.py").symlink_to(root / "versions/1234_example.py") wrapper = root / "dev_versions/1234_example.py" definitions = ( '_migration = import_module("govoplan_campaign.backend.migrations.versions.linked")\nrevision = _migration.revision\n', '_migration = import_module("govoplan_campaign.backend.migrations.versions.1234_example")\n' '_other = import_module("govoplan_campaign.backend.migrations.versions.5678_example")\n' 'revision = _migration.revision\ndown_revision = _other.down_revision\n', ) for definition in definitions: with self.subTest(definition=definition): wrapper.write_text("from importlib import import_module\n" + definition, encoding="utf-8") with self.assertRaisesRegex(ValueError, "unsupported or ambiguous"): audit.parse_migration_file("govoplan-campaign", wrapper) def test_unresolved_revision_expression_is_not_silently_omitted(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: path = Path(directory) / "1234_example.py" path.write_text("revision = calculate_revision()\n", encoding="utf-8") with self.assertRaisesRegex(ValueError, "Unsupported or ambiguous migration metadata"): audit.parse_migration_file("govoplan-core", path) def test_explicit_metadata_reexport_is_resolved_without_execution(self) -> None: audit = load_audit_module() with tempfile.TemporaryDirectory(prefix="migration-audit-test-") as directory: root = Path(directory) (root / "versions").mkdir() (root / "dev_versions").mkdir() (root / "versions/a234_example.py").write_text( 'revision = "1234"\ndown_revision = "base"\ndepends_on = None\nbranch_labels = None\n' 'raise AssertionError("Migration implementation must not execute")\n', encoding="utf-8", ) wrapper = root / "dev_versions/1234_example.py" source = "govoplan_organizations.backend.migrations.versions.a234_example" wrapper.write_text(f"from {source} import revision, down_revision, depends_on, branch_labels\n", encoding="utf-8") migration = audit.parse_migration_file("govoplan-organizations", wrapper) self.assertEqual(migration.revision, "1234") self.assertEqual(migration.down_revisions, ("base",)) for declaration in ( f"from {source} import *\n", f"from {source} import revision as down_revision\n", f"from {source} import revision\nrevision = 'different'\n", f"from {source} import revision\nimport another as revision\n", f"from {source} import revision\ndef revision():\n pass\n", f"from {source.replace('govoplan_organizations', 'govoplan_mail')} import revision\n", ): with self.subTest(declaration=declaration): wrapper.write_text(declaration, encoding="utf-8") with self.assertRaises(ValueError): audit.parse_migration_file("govoplan-organizations", wrapper) def test_release_baseline_matches_current_heads_in_strict_report(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("base.py"), "base", (), (), ()), audit.Migration("govoplan-core", Path("head.py"), "head", ("base",), (), ()), ] report = audit.build_report( migrations, baseline={ "version": 1, "releases": [ { "release": "0.1.0", "heads": [{"owner": "govoplan-core", "revision": "head"}], } ], }, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) self.assertEqual(report["current_heads"], ["head"]) self.assertEqual(report["graph_errors"], []) self.assertEqual(report["strict_errors"], []) def test_owner_heads_are_accepted_for_subset_installs(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("base.py"), "base", (), (), ()), audit.Migration("govoplan-core", Path("core_head.py"), "core-head", ("base",), (), ()), ] report = audit.build_report( migrations, baseline={ "version": 1, "releases": [ { "release": "0.1.0", "heads": [{"owner": "govoplan-files", "revision": "files-head"}], "owner_heads": [{"owner": "govoplan-core", "revisions": ["core-head"]}], } ], }, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) self.assertEqual(report["current_heads"], ["core-head"]) self.assertEqual(report["strict_errors"], []) def test_records_current_release_baseline(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("base.py"), "base", (), (), ()), audit.Migration("govoplan-core", Path("core_head.py"), "core-head", ("base",), (), ()), audit.Migration("govoplan-files", Path("files_head.py"), "files-head", ("core-head",), (), ()), ] report = audit.build_report( migrations, baseline={"version": 1, "releases": []}, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) baseline = audit.record_release_baseline( {"version": 1, "releases": []}, report, release="0.2.0", replace=False, ) release = baseline["releases"][0] self.assertEqual(release["release"], "0.2.0") self.assertEqual(release["squash_policy"], "reviewed-manual") self.assertEqual(release["heads"], [{"owner": "govoplan-files", "revision": "files-head"}]) self.assertIn({"owner": "govoplan-core", "revisions": ["core-head"]}, release["owner_heads"]) def test_missing_down_revision_is_a_graph_error(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("head.py"), "head", ("missing",), (), ()), ] report = audit.build_report( migrations, baseline={"version": 1, "releases": []}, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) self.assertIn("references missing down_revision", report["graph_errors"][0]) def test_depends_on_participates_in_graph_validation_and_heads(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("core.py"), "core", (), (), ()), audit.Migration("govoplan-files", Path("files.py"), "files", (), ("core",), ()), ] report = audit.build_report( migrations, baseline={"version": 1, "releases": []}, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) self.assertEqual(report["graph_errors"], []) self.assertEqual(report["current_heads"], ["files"]) def test_missing_depends_on_is_a_graph_error(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-files", Path("files.py"), "files", (), ("missing",), ()), ] report = audit.build_report( migrations, baseline={"version": 1, "releases": []}, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), ) self.assertIn("references missing depends_on", report["graph_errors"][0]) def test_dev_track_does_not_emit_release_baseline_warnings(self) -> None: audit = load_audit_module() migrations = [ audit.Migration("govoplan-core", Path("base.py"), "base", (), (), ()), audit.Migration("govoplan-core", Path("head.py"), "head", ("base",), (), ()), ] report = audit.build_report( migrations, baseline={"version": 1, "releases": []}, baseline_file=Path("docs/migration-release-baselines.json"), workspace_root=Path("/workspace"), track="dev", ) self.assertEqual(report["track"], "dev") self.assertEqual(report["strict_errors"], []) if __name__ == "__main__": unittest.main()