strepsiades
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from sync_state_machine.sync_system.config import UpdateDirection
|
|
from sync_state_machine.sync_system.strategy_ops.update_ops import run_update
|
|
from sync_state_machine.domain.construction.sync_strategy import ConstructionTaskSyncStrategy
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_update_with_direction_overrides():
|
|
class FakeNode:
|
|
def __init__(self, data):
|
|
self._data = data
|
|
|
|
def get_data(self):
|
|
return self._data
|
|
|
|
# Setup mock strategy
|
|
strategy = MagicMock()
|
|
strategy.node_type = "TestNode"
|
|
strategy.config.field_direction_key = "sync_direction"
|
|
strategy.config.field_direction_overrides = {
|
|
"local_only": UpdateDirection.PUSH,
|
|
"remote_only": UpdateDirection.PULL,
|
|
"ignore": UpdateDirection.BIDIRECTIONAL # Just to check filtering
|
|
}
|
|
strategy.config.update_direction = UpdateDirection.PUSH
|
|
strategy.local_collection = MagicMock()
|
|
strategy.remote_collection = MagicMock()
|
|
|
|
with patch('sync_state_machine.sync_system.strategy_ops.update_ops.add_update_action', new_callable=AsyncMock) as mock_add_action:
|
|
mock_add_action.return_value = []
|
|
|
|
await run_update(strategy)
|
|
|
|
# In the active directions, PUSH and PULL should be handled because of overrides.
|
|
# Check PUSH call
|
|
assert mock_add_action.call_count == 2
|
|
|
|
push_call = mock_add_action.call_args_list[0]
|
|
assert push_call.kwargs['source_is_local'] is True
|
|
assert push_call.kwargs['from_collection'] == strategy.local_collection
|
|
|
|
pull_call = mock_add_action.call_args_list[1]
|
|
assert pull_call.kwargs['source_is_local'] is False
|
|
assert pull_call.kwargs['from_collection'] == strategy.remote_collection
|
|
|
|
# Test the node filters
|
|
push_filter = push_call.kwargs['node_filter']
|
|
pull_filter = pull_call.kwargs['node_filter']
|
|
|
|
# Node that has 'local_only' -> PUSH
|
|
node_push = FakeNode({"sync_direction": "local_only"})
|
|
assert push_filter(node_push) is True
|
|
assert pull_filter(node_push) is False
|
|
|
|
# Node that has 'remote_only' -> PULL
|
|
node_pull = FakeNode({"sync_direction": "remote_only"})
|
|
assert push_filter(node_pull) is False
|
|
assert pull_filter(node_pull) is True
|
|
|
|
# Node with unknown key -> default direction (PUSH)
|
|
node_default = FakeNode({"sync_direction": "unknown"})
|
|
assert push_filter(node_default) is True
|
|
assert pull_filter(node_default) is False
|
|
|
|
# Node with empty data -> default direction (PUSH)
|
|
node_empty = FakeNode(None)
|
|
assert push_filter(node_empty) is True
|
|
assert pull_filter(node_empty) is False
|
|
|
|
|
|
def test_construction_task_compare_payload_ignores_plan_node_is_audit():
|
|
strategy = ConstructionTaskSyncStrategy(
|
|
"construction_task",
|
|
MagicMock(),
|
|
MagicMock(),
|
|
MagicMock(),
|
|
)
|
|
|
|
local_payload = {
|
|
"plan_node": [{"name": "A", "is_audit": True}],
|
|
"task_type": "demo",
|
|
}
|
|
remote_payload = {
|
|
"plan_node": [{"name": "A", "is_audit": False}],
|
|
"task_type": "demo",
|
|
}
|
|
|
|
assert strategy.normalize_compare_payload(local_payload) == strategy.normalize_compare_payload(remote_payload)
|