"""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")