Spaces:
Running
Running
| """Component-data adaptation invariants.""" | |
| from __future__ import annotations | |
| import importlib.util | |
| import pathlib | |
| import sys | |
| import types | |
| ROOT = pathlib.Path(__file__).resolve().parent | |
| MODULE = ROOT / "core" / "component_data_blocker_guard.py" | |
| def load(): | |
| core = types.ModuleType("core"); core.__path__ = [str(ROOT / "core")] | |
| um = types.ModuleType("core.unified_migration") | |
| um.plan = lambda source, target: { | |
| "database": { | |
| "component_data": {"domains": [ | |
| {"component": "com_easyblog", "tables": [ | |
| {"table": "easyblog_post", "class": "missing_target", "source_rows": 42, | |
| "reason": "target component table missing"}, | |
| {"table": "easyblog_acl", "class": "client_entity", "source_rows": 9}, | |
| ]}, | |
| {"component": "com_docman", "tables": [ | |
| {"table": "docman_documents", "class": "quarantine", "source_rows": 3, | |
| "reason": "no semantic merge key"}, | |
| ]}, | |
| ]}, | |
| "unowned_tables": [ | |
| {"table": "plugin_cache", "source_rows": 100, "reason": "not component-owned"}, | |
| ], | |
| }, | |
| "blockers": [], | |
| "ready_for_live": True, | |
| } | |
| semantic = types.ModuleType("core.component_data_semantics") | |
| semantic.installed = False | |
| def install_semantics(): | |
| semantic.installed = True | |
| semantic.install = install_semantics | |
| sys.modules["core"] = core | |
| sys.modules["core.unified_migration"] = um | |
| sys.modules["core.component_data_semantics"] = semantic | |
| spec = importlib.util.spec_from_file_location("core.component_data_blocker_guard", MODULE) | |
| mod = importlib.util.module_from_spec(spec); sys.modules["core.component_data_blocker_guard"] = mod | |
| spec.loader.exec_module(mod) | |
| mod.install() | |
| assert semantic.installed, "Joomla table semantics must install before adaptation evaluation" | |
| return um | |
| def test_component_data_gaps_are_deferred_or_quarantined_without_stopping_safe_rows(): | |
| um = load() | |
| got = um.plan("source", "target") | |
| assert got["blockers"] == [] | |
| assert got["ready_for_live"] is True | |
| kinds = [b["kind"] for b in got["adaptation_tasks"]] | |
| assert kinds == ["component_target_schema", "component_data_merge"] | |
| assert got["adaptation_tasks"][0]["disposition"] == "defer_until_target_schema_exists" | |
| assert got["adaptation_tasks"][1]["disposition"] == "quarantine_until_semantic_identity_is_proven" | |
| assert all(b["blocking"] is False for b in got["adaptation_tasks"]) | |
| assert got["runtime_table_warnings"][0]["table"] == "plugin_cache" | |
| assert not any(b.get("table") == "plugin_cache" for b in got["adaptation_tasks"]) | |
| if __name__ == "__main__": | |
| test_component_data_gaps_are_deferred_or_quarantined_without_stopping_safe_rows() | |
| print("component data adaptation guard: OK") | |