strepsiades
528 lines
20 KiB
Python
528 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import logging
|
|
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,
|
|
)
|
|
|
|
|
|
def _configure_strategy(strategy, **kwargs):
|
|
for key, value in kwargs.items():
|
|
setattr(strategy.config, key, value)
|
|
strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logging.getLogger(__name__))
|
|
|
|
|
|
@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)
|
|
_configure_strategy(
|
|
project_strategy,
|
|
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)
|
|
_configure_strategy(
|
|
contract_strategy,
|
|
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 local_collection.get("lp_auto").context.get("bind_data_id") == remote_collection.get("rp_auto").data_id
|
|
assert remote_collection.get("rp_auto").context.get("bind_data_id") == local_collection.get("lp_auto").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 local_collection.get("lp_create_ok").context.get("bind_data_id") == create_ok_nodes[0].data_id
|
|
assert create_ok_nodes[0].context.get("bind_data_id") == local_collection.get("lp_create_ok").data_id
|
|
assert local_collection.get("lp_create_async").context.get("bind_data_id") == create_async_nodes[0].data_id
|
|
assert create_async_nodes[0].context.get("bind_data_id") == local_collection.get("lp_create_async").data_id
|
|
assert "bind_data_id" not in local_collection.get("lp_create_fail").context
|
|
assert "bind_data_id" not in create_fail_nodes[0].context
|
|
|
|
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)
|
|
_configure_strategy(
|
|
project_strategy,
|
|
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
|