from __future__ import annotations import tempfile from pathlib import Path import copy import json import pytest import pytest_asyncio from sync_state_machine.common.binding import BindingManager from sync_state_machine.common.collection import DataCollection from sync_state_machine.common.persistence import PersistenceBackend from sync_state_machine.common.sqlite_backend import SQLitePersistenceBackend from sync_state_machine.common.types import BindingStatus, SyncAction, SyncStatus from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline from sync_state_machine.sync_system.strategy_ops.compare_ops import ( build_data_id_normalization_map, normalized_data_for_compare, collect_payload_diffs, ) from sync_state_machine.sync_system.config import OrphanAction, UpdateDirection from tests.fixtures.mock_domain import ( InMemoryDataSource, InMemoryNodeHandler, TestContractNode, TestContractSchema, TestContractStrategy, TestProjectNode, TestProjectSchema, TestProjectStrategy, build_contract_node, build_project_node, register_test_domain, ) @pytest_asyncio.fixture async def pipeline_bundle(): register_test_domain() db = Path(tempfile.mkdtemp(prefix="sync_state_machine_pipeline_") ) / "state.db" persistence = SQLitePersistenceBackend(str(db)) await persistence.initialize() local_collection = DataCollection("local", persistence=persistence, auto_persist=False) remote_collection = DataCollection("remote", persistence=persistence, auto_persist=False) binding_manager = BindingManager(persistence, auto_persist=False) local_project_nodes = [ build_project_node("lp_norm", "P-NORM", name="Project Normal", code="NORM"), build_project_node("lp_warn", "P-WARN", name="Project Warning", code="WARN"), build_project_node("lp_auto", "P-AUTO", name="Project Auto", code="AUTO-1"), build_project_node("lp_create_ok", "P-COK", name="Project Create OK", code="CREATE_OK"), build_project_node("lp_create_fail", "P-CFAIL", name="Project Create Fail", code="CREATE_FAIL"), build_project_node("lp_create_async", "P-CASYNC", name="Project Create Async", code="CREATE_ASYNC"), build_project_node("lp_update", "P-UPD", name="Project Update New", code="UPD-1"), ] remote_project_nodes = [ build_project_node("rp_norm", "RP-NORM", name="Project Normal", code="NORM"), build_project_node("rp_warn", "RP-WARN", name="Project Warning", code="WARN", data_present=False), build_project_node("rp_auto", "RP-AUTO", name="Project Auto", code="AUTO-1"), build_project_node("rp_update", "RP-UPD", name="Project Update Old", code="UPD-1"), ] local_contract_nodes = [ build_contract_node( "lc_dep_ok", "C-DEP-OK", name="Contract Dep OK", code="C-DEP-OK", project_ref="P-NORM", ), build_contract_node( "lc_dep_err", "C-DEP-ERR", name="Contract Dep Err", code="C-DEP-ERR", project_ref="P-MISSING", ), build_contract_node( "lc_upd_skip", "C-UPD-SKIP", name="Contract Update Skip New", code="C-UPD-SKIP", project_ref="P-NORM", ), build_contract_node( "lc_upd_fail", "C-UPD-FAIL", name="Contract Update Fail New", code="C-UPD-FAIL", project_ref="P-NORM", ), ] remote_contract_nodes = [ build_contract_node( "rc_upd_skip", "RC-UPD-SKIP", name="Contract Update Skip Old", code="C-UPD-SKIP", project_ref="RP-NORM", ), build_contract_node( "rc_upd_fail", "RC-UPD-FAIL", name="Contract Update Fail Old", code="C-UPD-FAIL", project_ref="RP-NORM", ), ] local_ds = InMemoryDataSource(poll_max_retries=2, poll_interval=0) remote_ds = InMemoryDataSource(poll_max_retries=2, poll_interval=0) local_ds.add_handler( InMemoryNodeHandler( node_type="test_project", node_class=TestProjectNode, schema=TestProjectSchema, load_nodes=local_project_nodes, ) ) local_ds.add_handler( InMemoryNodeHandler( node_type="test_contract", node_class=TestContractNode, schema=TestContractSchema, load_nodes=local_contract_nodes, ) ) remote_ds.add_handler( InMemoryNodeHandler( node_type="test_project", node_class=TestProjectNode, schema=TestProjectSchema, load_nodes=remote_project_nodes, create_mode_by_code={ "CREATE_OK": "success", "CREATE_FAIL": "failed", "CREATE_ASYNC": "in_progress_success", }, update_mode_by_code={"UPD-1": "success"}, ) ) remote_ds.add_handler( InMemoryNodeHandler( node_type="test_contract", node_class=TestContractNode, schema=TestContractSchema, load_nodes=remote_contract_nodes, create_mode_by_code={"C-DEP-OK": "success"}, update_mode_by_code={ "C-UPD-SKIP": "skipped", "C-UPD-FAIL": "failed", }, ) ) await binding_manager.bind("test_project", "lp_norm", "rp_norm") await binding_manager.bind("test_project", "lp_warn", "rp_warn") await binding_manager.bind("test_project", "lp_update", "rp_update") await binding_manager.bind("test_contract", "lc_upd_skip", "rc_upd_skip") await binding_manager.bind("test_contract", "lc_upd_fail", "rc_upd_fail") await binding_manager.persist() project_strategy = TestProjectStrategy("test_project", local_collection, remote_collection, binding_manager) project_strategy.update_config( auto_bind=True, auto_bind_fields=["code"], depend_fields={}, local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, ) contract_strategy = TestContractStrategy("test_contract", local_collection, remote_collection, binding_manager) contract_strategy.update_config( auto_bind=False, depend_fields={"project_ref": "test_project"}, local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, ) pipeline = FullSyncPipeline( local_datasource=local_ds, remote_datasource=remote_ds, strategies=[project_strategy, contract_strategy], node_types=["test_project", "test_contract"], load_order=["test_project", "test_contract"], sync_order=["test_project", "test_contract"], persistence=persistence, local_collection=local_collection, remote_collection=remote_collection, binding_manager=binding_manager, enable_persistence=True, ) yield pipeline, local_collection, remote_collection, binding_manager @pytest.mark.asyncio async def test_full_pipeline_multinode_mock_covers_business_points(pipeline_bundle): pipeline, local_collection, remote_collection, binding_manager = pipeline_bundle stats = await pipeline.run() assert stats["post_check_passed"] is True assert local_collection.get("lp_norm").binding_status == BindingStatus.NORMAL assert local_collection.get("lp_warn").binding_status == BindingStatus.WARNING assert remote_collection.get("rp_warn").binding_status == BindingStatus.ABNORMAL assert local_collection.get("lc_dep_err").binding_status == BindingStatus.DEPENDENCY_ERROR assert local_collection.get("lp_norm").context.get("bind_data_id") == remote_collection.get("rp_norm").data_id assert remote_collection.get("rp_norm").context.get("bind_data_id") == local_collection.get("lp_norm").data_id assert await binding_manager.get_remote_id("test_project", "lp_auto") == "rp_auto" project_nodes = remote_collection.filter(node_type="test_project") create_ok_nodes = [n for n in project_nodes if (n.get_data() or {}).get("code") == "CREATE_OK"] create_fail_nodes = [n for n in project_nodes if (n.get_data() or {}).get("code") == "CREATE_FAIL"] create_async_nodes = [n for n in project_nodes if (n.get_data() or {}).get("code") == "CREATE_ASYNC"] assert len(create_ok_nodes) == 1 assert len(create_fail_nodes) == 1 assert len(create_async_nodes) == 1 assert create_ok_nodes[0].status == SyncStatus.PENDING assert create_ok_nodes[0].action == SyncAction.NONE assert create_fail_nodes[0].status == SyncStatus.FAILED assert create_async_nodes[0].status == SyncStatus.PENDING assert create_async_nodes[0].action == SyncAction.NONE assert remote_collection.get("rc_upd_skip").status == SyncStatus.SKIPPED assert remote_collection.get("rc_upd_skip").action == SyncAction.NONE assert remote_collection.get("rc_upd_fail").status == SyncStatus.FAILED assert local_collection.get("lc_dep_ok").binding_status == BindingStatus.WARNING assert await binding_manager.get_remote_id("test_contract", "lc_dep_ok") is None project_bindings = await binding_manager.get_all_records("test_project") contract_bindings = await binding_manager.get_all_records("test_contract") assert len(project_bindings) >= 7 assert len(contract_bindings) >= 2 @pytest.mark.asyncio async def test_pipeline_reloads_after_create_update_writes(pipeline_bundle): pipeline, _, _, _ = pipeline_bundle load_counts: dict[tuple[str, str], int] = {} def attach_counter(datasource, side: str, node_type: str): handler = datasource.get_handler(node_type) original_load = handler.load async def _wrapped_load(): key = (side, node_type) load_counts[key] = load_counts.get(key, 0) + 1 return await original_load() handler.load = _wrapped_load # type: ignore[method-assign] for side_name, ds in (("local", pipeline.local_datasource), ("remote", pipeline.remote_datasource)): attach_counter(ds, side_name, "test_project") attach_counter(ds, side_name, "test_contract") await pipeline.run() assert load_counts[("local", "test_project")] >= 3 assert load_counts[("remote", "test_project")] >= 3 assert load_counts[("local", "test_contract")] >= 1 assert load_counts[("remote", "test_contract")] >= 1 @pytest.mark.asyncio async def test_pipeline_emits_consistency_stats(pipeline_bundle): pipeline, _, _, _ = pipeline_bundle stats = await pipeline.run() assert "consistency" in stats assert "test_project" in stats["consistency"] assert "test_contract" in stats["consistency"] project_consistency = stats["consistency"]["test_project"] contract_consistency = stats["consistency"]["test_contract"] assert isinstance(project_consistency["ok"], bool) assert isinstance(contract_consistency["ok"], bool) assert isinstance(project_consistency["checked_pairs"], int) assert isinstance(contract_consistency["checked_pairs"], int) assert isinstance(project_consistency["mismatch_count"], int) assert isinstance(contract_consistency["mismatch_count"], int) @pytest.mark.asyncio async def test_load_overrides_bind_data_id_from_bindings(pipeline_bundle): pipeline, local_collection, remote_collection, _ = pipeline_bundle await pipeline.run() existing_persistence = local_collection.persistence assert existing_persistence is not None persistence = SQLitePersistenceBackend(existing_persistence.db_path) await persistence.initialize() await persistence.conn.execute( "UPDATE nodes SET bind_data_id = ?, context = ? WHERE collection_id = ? AND node_id = ?", ("STALE", json.dumps({"bind_data_id": "STALE"}), "local", "lp_norm"), ) await persistence.conn.execute( "UPDATE nodes SET bind_data_id = ?, context = ? WHERE collection_id = ? AND node_id = ?", ("STALE", json.dumps({"bind_data_id": "STALE"}), "local", "lp_create_fail"), ) await persistence.conn.commit() local_reloaded = DataCollection("local", persistence=persistence, auto_persist=False) remote_reloaded = DataCollection("remote", persistence=persistence, auto_persist=False) await local_reloaded.load_from_persistence() await remote_reloaded.load_from_persistence() assert local_reloaded.get("lp_norm").context.get("bind_data_id") == remote_reloaded.get("rp_norm").data_id assert remote_reloaded.get("rp_norm").context.get("bind_data_id") == local_reloaded.get("lp_norm").data_id assert "bind_data_id" not in local_reloaded.get("lp_create_fail").context await persistence.close() @pytest.mark.asyncio async def test_post_check_normalizes_id_fields_and_none_empty_equivalence(pipeline_bundle): pipeline, local_collection, remote_collection, _ = pipeline_bundle await pipeline.run() local_node = local_collection.get("lp_norm") remote_node = remote_collection.get("rp_norm") local_ref = local_collection.get("lp_warn") remote_ref = remote_collection.get("rp_warn") assert local_node is not None and remote_node is not None assert local_ref is not None and remote_ref is not None assert local_node.get_data() is not None and remote_node.get_data() is not None local_payload_seed = local_node.get_data() or {} remote_payload_seed = remote_node.get_data() or {} local_payload_seed["contract_id"] = local_ref.data_id remote_payload_seed["contract_id"] = remote_ref.data_id local_payload_seed["contract_type"] = None remote_payload_seed["contract_type"] = "GP" local_payload_seed["plan_start_date"] = None remote_payload_seed["plan_start_date"] = "" local_node.set_data(local_payload_seed, validate=False) remote_node.set_data(remote_payload_seed, validate=False) data_id_map = await build_data_id_normalization_map( node_type="test_project", local_collection=local_collection, remote_collection=remote_collection, binding_manager=pipeline.binding_manager, ) local_payload = normalized_data_for_compare(local_node.get_data(), data_id_map) remote_payload = normalized_data_for_compare(remote_node.get_data(), data_id_map) diff_map = collect_payload_diffs(local_payload, remote_payload) assert "contract_id" not in diff_map assert "contract_type" in diff_map assert "plan_start_date" not in diff_map class _MutableProjectRemoteHandler(InMemoryNodeHandler): def __init__(self): super().__init__( node_type="test_project", node_class=TestProjectNode, schema=TestProjectSchema, load_nodes=[], ) self._store_by_node_id: dict[str, dict] = {} self.update_calls = 0 async def load(self): nodes = [] for row in self._store_by_node_id.values(): nodes.append( TestProjectNode( node_id=row["node_id"], data_id=row["data_id"], data=copy.deepcopy(row["data"]), origin_data=copy.deepcopy(row["data"]), ) ) return nodes async def create_all(self, nodes): for node in nodes: node_data = node.get_data() or {} new_data_id = f"RID-{node.node_id}" self._store_by_node_id[node.node_id] = { "node_id": node.node_id, "data_id": new_data_id, "data": { "id": new_data_id, "name": "REMOTE-STUB", "code": node_data.get("code", ""), }, } from sync_state_machine.datasource.task_result import TaskResult return [ TaskResult.success(node_id=node.node_id, data_id=f"RID-{node.node_id}") for node in nodes ] async def update_all(self, nodes): self.update_calls += len(nodes) from sync_state_machine.datasource.task_result import TaskResult results = [] for node in nodes: node_data = node.get_data() or {} existing = self._store_by_node_id.get(node.node_id) if existing is None: existing_data_id = node.data_id or f"RID-{node.node_id}" else: existing_data_id = existing["data_id"] self._store_by_node_id[node.node_id] = { "node_id": node.node_id, "data_id": existing_data_id, "data": { "id": existing_data_id, "name": node_data.get("name", ""), "code": node_data.get("code", ""), }, } results.append(TaskResult.success(node_id=node.node_id, data_id=existing_data_id)) return results @pytest.mark.asyncio async def test_pipeline_create_then_update_completes_in_single_run_with_reload(): register_test_domain() db = Path(tempfile.mkdtemp(prefix="sync_state_machine_pipeline_reload_") ) / "state.db" persistence = SQLitePersistenceBackend(str(db)) await persistence.initialize() local_collection = DataCollection("local", persistence=persistence, auto_persist=False) remote_collection = DataCollection("remote", persistence=persistence, auto_persist=False) binding_manager = BindingManager(persistence, auto_persist=False) local_ds = InMemoryDataSource(poll_max_retries=2, poll_interval=0) remote_ds = InMemoryDataSource(poll_max_retries=2, poll_interval=0) local_ds.add_handler( InMemoryNodeHandler( node_type="test_project", node_class=TestProjectNode, schema=TestProjectSchema, load_nodes=[ build_project_node("lp_create_then_update", "P-CREATE-UPDATE", name="Project Final", code="P-CU"), ], ) ) remote_handler = _MutableProjectRemoteHandler() remote_ds.add_handler(remote_handler) project_strategy = TestProjectStrategy("test_project", local_collection, remote_collection, binding_manager) project_strategy.update_config( auto_bind=True, auto_bind_fields=["code"], depend_fields={}, local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, ) pipeline = FullSyncPipeline( local_datasource=local_ds, remote_datasource=remote_ds, strategies=[project_strategy], node_types=["test_project"], load_order=["test_project"], sync_order=["test_project"], persistence=persistence, local_collection=local_collection, remote_collection=remote_collection, binding_manager=binding_manager, enable_persistence=True, ) stats = await pipeline.run() records = await binding_manager.get_all_records("test_project") assert len(records) == 1 local_id = records[0].local_id remote_id = records[0].remote_id assert remote_id is not None local_node = local_collection.get_by_node_id(local_id) remote_node = remote_collection.get_by_node_id(remote_id) assert local_node is not None assert remote_node is not None assert remote_handler.update_calls >= 1 assert (remote_node.get_data() or {}).get("name") == "Project Final" assert stats["consistency"]["test_project"]["ok"] is True