Spaces:
Running
Running
| """Component-only extension data migration invariants.""" | |
| from __future__ import annotations | |
| from core import component_data_domains as domains | |
| from core import component_data_guard as guard | |
| from core import component_data_semantics as semantics | |
| from core import dbmesh_policy_patch as policy | |
| class FakeCursor: | |
| def __init__(self, conn): | |
| self.conn = conn | |
| self.rows = [] | |
| def __enter__(self): return self | |
| def __exit__(self, *args): return False | |
| def execute(self, sql, params=None): | |
| table = sql.split("FROM `", 1)[1].split("`", 1)[0] | |
| select = sql.split("SELECT ", 1)[1].split(" FROM ", 1)[0] | |
| cols = [x.strip().strip("`") for x in select.split(",")] | |
| self.rows = [{c: row.get(c) for c in cols} for row in self.conn.rows.get(table, [])] | |
| return len(self.rows) | |
| def fetchall(self): return list(self.rows) | |
| class FakeConn: | |
| def __init__(self, rows): self.rows = rows | |
| def cursor(self): return FakeCursor(self) | |
| def test_component_prefix_ownership_is_explicit(): | |
| easy = {"element": "com_easyblog", "token": "easyblog"} | |
| dpc = {"element": "com_dpcalendar", "token": "dpcalendar"} | |
| users = {"element": "com_users", "token": "users"} | |
| tours = {"element": "com_guidedtours", "token": "guidedtours"} | |
| assert domains.owns(easy, "easyblog_post") | |
| assert domains.owns(easy, "easyblog_acl_group") | |
| assert not domains.owns(easy, "finder_terms") | |
| assert domains.owns(dpc, "dpcalendar_locations") | |
| assert domains.owns(users, "user_profiles") | |
| assert domains.owns(tours, "guidedtours") | |
| # Joomla names the child table in the singular, so this needs the explicit map. | |
| assert domains.owns(tours, "guidedtour_steps") | |
| def test_unowned_tables_are_quarantined_even_when_target_is_empty(): | |
| old_tables, old_rows = guard.d.tables, guard.d.row_count | |
| old_owner, old_prior = guard.domains.owner, guard._PRIOR_CLASSIFY | |
| try: | |
| guard.d.tables = lambda conn, prefix: {"mystery_cache", "easyblog_post"} | |
| guard.d.row_count = lambda conn, table: 5 if table.endswith("mystery_cache") else 8 | |
| guard.domains.owner = lambda conn, prefix, table: ( | |
| {"element": "com_easyblog", "name": "EasyBlog", "token": "easyblog", | |
| "version": "6.0", "core": False} | |
| if table == "easyblog_post" else None | |
| ) | |
| guard._PRIOR_CLASSIFY = lambda s, d, sp, dp, table, phase="extensions": { | |
| "class": "client_entity", "source_rows": 8, "target_rows": 0 | |
| } | |
| q = guard.classify_table(object(), object(), "s_", "d_", "mystery_cache", phase="extensions") | |
| assert q["class"] == "quarantine" | |
| assert q["component"] == "" | |
| assert "plugins/modules/packages are not data owners" in q["reason"] | |
| ok = guard.classify_table(object(), object(), "s_", "d_", "easyblog_post", phase="extensions") | |
| assert ok["class"] == "client_entity" | |
| assert ok["component"] == "com_easyblog" | |
| finally: | |
| guard.d.tables, guard.d.row_count = old_tables, old_rows | |
| guard.domains.owner, guard._PRIOR_CLASSIFY = old_owner, old_prior | |
| def test_regenerate_state_is_not_mislabeled_as_factory(): | |
| old_tables, old_rows = guard.d.tables, guard.d.row_count | |
| try: | |
| guard.d.tables = lambda conn, prefix: {"finder_terms"} | |
| guard.d.row_count = lambda conn, table: 77 | |
| got = guard.classify_table(object(), object(), "s_", "d_", "finder_terms", phase="extensions") | |
| assert got == {"class": "regenerate", "source_rows": 77, "target_rows": 77} | |
| finally: | |
| guard.d.tables, guard.d.row_count = old_tables, old_rows | |
| def test_known_derived_profiles_and_guided_tour_semantics_are_installed(): | |
| assert "finder_links_terms" in policy.REGENERATE | |
| for suffix in "0123456789abcdef": | |
| shard = f"finder_links_terms{suffix}" | |
| assert shard in policy.REGENERATE | |
| assert shard in guard.d.NEVER_COPY | |
| assert "scheduler_logs" in policy.REGENERATE | |
| assert guard.d.NATURAL_KEY["user_profiles"] == ("user_id", "profile_key") | |
| assert guard.d.FKS["user_profiles"]["user_id"] == "users" | |
| # Custom tours are real client data, not target-factory-only rows. | |
| assert "guidedtours" not in semantics.REUSE_ONLY_KEYS | |
| assert guard.d.NATURAL_KEY["guidedtours"] == ("uid",) | |
| assert guard.d.NATURAL_KEY["guidedtour_steps"] == ("tour_id", "ordering", "title") | |
| assert guard.d.FKS["guidedtour_steps"]["tour_id"] == "guidedtours" | |
| def test_guided_tours_run_before_their_steps(): | |
| old_phase, old_prior = semantics._PRIOR_PHASE_TABLES, semantics._PRIOR_CLASSIFY | |
| try: | |
| semantics._PRIOR_PHASE_TABLES = lambda *a, **k: [ | |
| "z_other", "guidedtour_steps", "guidedtours", "a_other" | |
| ] | |
| semantics._PRIOR_CLASSIFY = lambda *a, **k: { | |
| "class": "semantic_merge", "source_rows": 1, "target_rows": 1 | |
| } | |
| rows = semantics.phase_tables(object(), object(), "s_", "d_", "extensions") | |
| assert rows[:2] == ["guidedtours", "guidedtour_steps"] | |
| assert rows[2:] == ["a_other", "z_other"] | |
| finally: | |
| semantics._PRIOR_PHASE_TABLES, semantics._PRIOR_CLASSIFY = old_phase, old_prior | |
| def test_reuse_only_scheduler_tasks_unblock_only_when_target_owns_every_identity(): | |
| old_tables, old_columns, old_prior = semantics.d.tables, semantics.d.columns, semantics._PRIOR_CLASSIFY | |
| src = FakeConn({"s_scheduler_tasks": [{"type": "a", "title": "A"}, {"type": "b", "title": "B"}]}) | |
| dst = FakeConn({"d_scheduler_tasks": [{"type": "a", "title": "A"}, {"type": "b", "title": "B"}]}) | |
| try: | |
| semantics.d.tables = lambda conn, prefix: { | |
| name[len(prefix):] for name in conn.rows if name.startswith(prefix) | |
| } | |
| semantics.d.columns = lambda conn, table: { | |
| key: {} for row in conn.rows.get(table, []) for key in row | |
| } | |
| semantics._PRIOR_CLASSIFY = lambda *a, **k: { | |
| "class": "quarantine", "source_rows": 2, "target_rows": 2, | |
| "component": "com_scheduler" | |
| } | |
| got = semantics.classify_table(src, dst, "s_", "d_", "scheduler_tasks", phase="extensions") | |
| assert got["class"] == "semantic_merge" | |
| assert got["identity"] == ["type", "title"] | |
| src.rows["s_scheduler_tasks"].append({"type": "custom", "title": "Custom"}) | |
| semantics._PRIOR_CLASSIFY = lambda *a, **k: { | |
| "class": "quarantine", "source_rows": 3, "target_rows": 2, | |
| "component": "com_scheduler" | |
| } | |
| blocked = semantics.classify_table(src, dst, "s_", "d_", "scheduler_tasks", phase="extensions") | |
| assert blocked["class"] == "quarantine" | |
| assert blocked["source_only_identities"] == [["custom", "Custom"]] | |
| finally: | |
| semantics.d.tables, semantics.d.columns = old_tables, old_columns | |
| semantics._PRIOR_CLASSIFY = old_prior | |
| def test_stock_workflow_is_target_factory_but_custom_workflow_is_not_assumed(): | |
| transitions = [ | |
| "UNPUBLISH", "PUBLISH", "TRASH", "ARCHIVE", | |
| "FEATURE", "UNFEATURE", "PUBLISH_AND_FEATURE", | |
| ] | |
| src = FakeConn({ | |
| "s_workflow_stages": [{"title": "COM_WORKFLOW_BASIC_STAGE"}], | |
| "s_workflow_transitions": [{"title": t} for t in transitions], | |
| "s_workflow_associations": [{"item_id": 1}], | |
| }) | |
| dst = FakeConn({ | |
| "d_workflow_stages": [{"title": "COM_WORKFLOW_BASIC_STAGE"}], | |
| "d_workflow_transitions": [{"title": t} for t in transitions], | |
| "d_workflow_associations": [{"item_id": 99}], | |
| }) | |
| old_tables, old_columns, old_prior = semantics.d.tables, semantics.d.columns, semantics._PRIOR_CLASSIFY | |
| try: | |
| semantics.d.tables = lambda conn, prefix: { | |
| name[len(prefix):] for name in conn.rows if name.startswith(prefix) | |
| } | |
| semantics.d.columns = lambda conn, table: { | |
| key: {} for row in conn.rows.get(table, []) for key in row | |
| } | |
| semantics._PRIOR_CLASSIFY = lambda *a, **k: { | |
| "class": "quarantine", "source_rows": 1, "target_rows": 1, | |
| "component": "com_workflow" | |
| } | |
| stage = semantics.classify_table(src, dst, "s_", "d_", "workflow_stages", phase="extensions") | |
| assoc = semantics.classify_table(src, dst, "s_", "d_", "workflow_associations", phase="extensions") | |
| assert stage["class"] == "target_factory" | |
| assert assoc["class"] == "regenerate" | |
| src.rows["s_workflow_stages"].append({"title": "Custom editorial stage"}) | |
| custom = semantics.classify_table(src, dst, "s_", "d_", "workflow_stages", phase="extensions") | |
| assert custom["class"] == "quarantine" | |
| finally: | |
| semantics.d.tables, semantics.d.columns = old_tables, old_columns | |
| semantics._PRIOR_CLASSIFY = old_prior | |
| if __name__ == "__main__": | |
| test_component_prefix_ownership_is_explicit() | |
| test_unowned_tables_are_quarantined_even_when_target_is_empty() | |
| test_regenerate_state_is_not_mislabeled_as_factory() | |
| test_known_derived_profiles_and_guided_tour_semantics_are_installed() | |
| test_guided_tours_run_before_their_steps() | |
| test_reuse_only_scheduler_tasks_unblock_only_when_target_owns_every_identity() | |
| test_stock_workflow_is_target_factory_but_custom_workflow_is_not_assumed() | |
| print("component data guard: OK") | |