647 lines
24 KiB
Python
647 lines
24 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import Mock, patch
|
|
|
|
import requests
|
|
from sqlalchemy import select
|
|
|
|
from app import create_app, db
|
|
from app.models import Component, LoadingReport, LoadingReportComponent, Recipe, SyncQueue
|
|
from config import TestingConfig
|
|
from sync_client import SyncClient, _attach_local_db_apply, _attach_local_db_push, _push_task_sort_key
|
|
|
|
|
|
class SyncClientApiTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.client = SyncClient()
|
|
self.client.server_url = "http://example.local"
|
|
self.client.client_id = "client-test-1"
|
|
self.client.config["sync_compression"]["enabled"] = True
|
|
self.client.log_auto_upload = False
|
|
|
|
@patch("sync_client.requests.post")
|
|
def test_pull_changes_returns_list(self, mock_post: Mock) -> None:
|
|
response = Mock()
|
|
response.status_code = 200
|
|
response.content = b"{}"
|
|
response.json.return_value = {
|
|
"changes": [{"id": "t1", "table_name": "component", "record_id": "c1", "action": "update"}]
|
|
}
|
|
mock_post.return_value = response
|
|
|
|
pulled = self.client.pull_changes()
|
|
|
|
self.assertEqual(len(pulled), 1)
|
|
self.assertEqual(pulled[0]["id"], "t1")
|
|
|
|
@patch("sync_client.requests.post")
|
|
def test_push_changes_conflict_response(self, mock_post: Mock) -> None:
|
|
response = Mock()
|
|
response.status_code = 409
|
|
response.content = b"{}"
|
|
response.json.return_value = {
|
|
"error": True,
|
|
"message": "conflict",
|
|
"total_conflicts": 2,
|
|
}
|
|
mock_post.return_value = response
|
|
|
|
result = self.client.push_changes(
|
|
[{"table_name": "component", "record_id": "c1", "action": "update", "data": {"version": 1}}]
|
|
)
|
|
|
|
self.assertFalse(result.get("success"))
|
|
self.assertTrue(result.get("error"))
|
|
self.assertEqual(result.get("total_conflicts"), 2)
|
|
|
|
@patch("sync_client.requests.post")
|
|
def test_push_changes_uses_gzip_when_enabled(self, mock_post: Mock) -> None:
|
|
response = Mock()
|
|
response.status_code = 200
|
|
response.content = b"{}"
|
|
response.json.return_value = {"success": True}
|
|
mock_post.return_value = response
|
|
|
|
self.client.push_changes([])
|
|
|
|
_, kwargs = mock_post.call_args
|
|
headers = kwargs.get("headers", {})
|
|
self.assertEqual(headers.get("Content-Encoding"), "gzip")
|
|
self.assertIn("data", kwargs)
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
def test_sync_cycle_transport_flow(
|
|
self,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
mock_collect.return_value = [
|
|
{"table_name": "component", "record_id": "c1", "action": "update", "data": {"id": "c1"}}
|
|
]
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
mock_pull.return_value = [
|
|
{"id": "task-1", "table_name": "component", "record_id": "c2", "action": "update"}
|
|
]
|
|
mock_confirm.return_value = True
|
|
|
|
result = self.client.sync_cycle()
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["pushed"], 1)
|
|
self.assertEqual(result["pulled"], 1)
|
|
self.assertEqual(result["confirmed"], 1)
|
|
self.assertEqual(result["conflicts"], 0)
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
def test_sync_cycle_confirms_only_successful_apply_tasks(
|
|
self,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
"""Ошибка apply: не вызывать confirm для проваленных task id — иначе сервер больше не отдаёт строки."""
|
|
mock_collect.return_value = []
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
mock_pull.return_value = [
|
|
{"id": "ok-1", "table_name": "component", "record_id": "a", "action": "update"},
|
|
{"id": "bad-1", "table_name": "component", "record_id": "b", "action": "update"},
|
|
]
|
|
|
|
def on_apply(changes):
|
|
failed = []
|
|
for ch in changes:
|
|
tid = ch.get("id")
|
|
if tid == "bad-1":
|
|
failed.append(str(tid))
|
|
return failed
|
|
|
|
self.client.on_remote_changes = on_apply
|
|
mock_confirm.return_value = True
|
|
|
|
result = self.client.sync_cycle()
|
|
|
|
self.assertEqual(result["pulled"], 2)
|
|
self.assertEqual(result["confirmed"], 1)
|
|
mock_confirm.assert_called_once_with(["ok-1"])
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
def test_initial_sync_progress_bumps_only_successful_applies(
|
|
self,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
"""Прогресс первой синхронизации — по успешному apply, не по числу строк в ответе pull."""
|
|
mock_collect.return_value = []
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
mock_pull.return_value = [
|
|
{"id": "ok-1", "table_name": "component", "record_id": "a", "action": "update", "data": {}},
|
|
{"id": "bad-1", "table_name": "component", "record_id": "b", "action": "update", "data": {}},
|
|
]
|
|
mock_confirm.return_value = True
|
|
|
|
def on_apply(changes):
|
|
return ["bad-1"]
|
|
|
|
self.client.on_remote_changes = on_apply
|
|
self.client._initial_sync_active = True
|
|
self.client._last_pull_meta = {"remaining_hint": 10}
|
|
|
|
self.client.sync_cycle()
|
|
|
|
self.assertEqual(self.client._initial_sync_applied_tasks, 1)
|
|
|
|
@patch("sync_client.time.sleep")
|
|
@patch.object(SyncClient, "_request_json")
|
|
def test_pull_busy_retry_handles_202_with_retry_after(
|
|
self, mock_request_json: Mock, mock_sleep: Mock
|
|
) -> None:
|
|
mock_request_json.side_effect = [
|
|
{"status_code": 202, "body": {"retry_after_sec": 2}},
|
|
{"status_code": 200, "body": {"changes": []}},
|
|
]
|
|
out = self.client._request_pull_with_busy_retry({"client_id": "client-test-1", "limit": 10})
|
|
self.assertEqual(out["status_code"], 200)
|
|
mock_sleep.assert_called_once_with(2)
|
|
|
|
@patch.object(SyncClient, "_request_pull_with_busy_retry")
|
|
def test_pull_changes_rejects_batch_when_total_mismatch(self, mock_pull: Mock) -> None:
|
|
mock_pull.return_value = {
|
|
"status_code": 200,
|
|
"body": {
|
|
"changes": [{"id": "task-1"}],
|
|
"total": 2,
|
|
"has_more": False,
|
|
"batch_id": "b-1",
|
|
},
|
|
}
|
|
pulled = self.client.pull_changes()
|
|
self.assertEqual(pulled, [])
|
|
self.assertFalse(self.client._last_pull_integrity_ok)
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
def test_sync_cycle_skips_confirm_for_failed_integrity_batch(
|
|
self,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
mock_collect.return_value = []
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
|
|
def pull_side_effect(*_args, **_kwargs):
|
|
self.client._last_pull_meta = {"has_more": False, "batch_id": "bad-batch"}
|
|
self.client._last_pull_integrity_ok = False
|
|
return [{"id": "task-1"}]
|
|
|
|
mock_pull.side_effect = pull_side_effect
|
|
out = self.client.sync_cycle()
|
|
self.assertTrue(out["success"])
|
|
self.assertEqual(out["pulled"], 1)
|
|
self.assertEqual(out["confirmed"], 0)
|
|
mock_confirm.assert_not_called()
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
@patch("sync_client.time.sleep")
|
|
def test_sync_cycle_reads_until_has_more_false(
|
|
self,
|
|
mock_sleep: Mock,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
mock_collect.return_value = []
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
|
|
def pull_side_effect(*_args, **_kwargs):
|
|
if not getattr(self.client, "_test_pull_called", False):
|
|
self.client._test_pull_called = True
|
|
self.client._last_pull_meta = {"has_more": True}
|
|
return [{"id": "task-1"}]
|
|
self.client._last_pull_meta = {"has_more": False}
|
|
return [{"id": "task-2"}]
|
|
|
|
mock_pull.side_effect = pull_side_effect
|
|
mock_confirm.return_value = True
|
|
|
|
result = self.client.sync_cycle()
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["pulled"], 2)
|
|
self.assertEqual(result["confirmed"], 2)
|
|
mock_sleep.assert_called()
|
|
|
|
@patch("sync_client.requests.post")
|
|
def test_upload_client_log_bytes_posts_gzip(self, mock_post: Mock) -> None:
|
|
mock_resp = Mock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.content = b'{"success": true}'
|
|
mock_resp.json.return_value = {"success": True}
|
|
mock_post.return_value = mock_resp
|
|
|
|
out = self.client.upload_client_log_bytes(b"local log", remote_name="wesp.log")
|
|
|
|
self.assertTrue(out.get("success"))
|
|
args, kwargs = mock_post.call_args
|
|
self.assertIn("/api/sync/client-log", args[0])
|
|
self.assertEqual(kwargs["headers"].get("Content-Encoding"), "gzip")
|
|
self.assertEqual(kwargs["headers"].get("X-WESP-Client-Id"), "client-test-1")
|
|
|
|
def test_client_log_auto_upload_due_respects_interval(self) -> None:
|
|
c = SyncClient()
|
|
c.server_url = "http://example.local"
|
|
c.log_auto_upload = True
|
|
c.role = "client"
|
|
c.log_auto_interval_sec = 86400
|
|
c._last_log_upload_at = ""
|
|
with patch.object(SyncClient, "default_local_log_path", return_value=""):
|
|
self.assertFalse(c._is_client_log_auto_upload_due())
|
|
|
|
fd, path = tempfile.mkstemp(suffix=".log")
|
|
os.close(fd)
|
|
try:
|
|
with patch.object(SyncClient, "default_local_log_path", return_value=path):
|
|
self.assertTrue(c._is_client_log_auto_upload_due())
|
|
c._last_log_upload_at = datetime.now(timezone.utc).isoformat()
|
|
with patch.object(SyncClient, "default_local_log_path", return_value=path):
|
|
self.assertFalse(c._is_client_log_auto_upload_due())
|
|
finally:
|
|
os.remove(path)
|
|
|
|
def test_resolve_sync_server_url_empty_without_sources(self) -> None:
|
|
from config import resolve_sync_server_url
|
|
|
|
class _Cfg:
|
|
SYNC_CLIENT_SERVER_URL = ""
|
|
|
|
with patch.dict(os.environ, {"WESP_SYNC_SERVER_URL": ""}):
|
|
u, ex = resolve_sync_server_url(state={}, config_defaults=_Cfg())
|
|
self.assertEqual(u, "")
|
|
self.assertFalse(ex)
|
|
|
|
def test_resolve_sync_server_url_state_wins(self) -> None:
|
|
from config import resolve_sync_server_url
|
|
|
|
class _Cfg:
|
|
SYNC_CLIENT_SERVER_URL = "http://cfg:5000"
|
|
|
|
with patch.dict(os.environ, {"WESP_SYNC_SERVER_URL": ""}):
|
|
u, ex = resolve_sync_server_url(
|
|
state={"server_url": "http://central:9/"},
|
|
config_defaults=_Cfg(),
|
|
)
|
|
self.assertEqual(u, "http://central:9")
|
|
self.assertTrue(ex)
|
|
|
|
def test_resolve_sync_server_url_empty_key_in_state_skips_config(self) -> None:
|
|
"""Ключ server_url в JSON есть, но пустой — Config не подставляет URL."""
|
|
from config import resolve_sync_server_url
|
|
|
|
class _Cfg:
|
|
SYNC_CLIENT_SERVER_URL = "http://cfg:5000"
|
|
|
|
with patch.dict(os.environ, {"WESP_SYNC_SERVER_URL": ""}):
|
|
u, ex = resolve_sync_server_url(
|
|
state={"server_url": ""},
|
|
config_defaults=_Cfg(),
|
|
)
|
|
self.assertEqual(u, "")
|
|
self.assertTrue(ex)
|
|
|
|
def test_confirm_clears_has_more_when_initial_sync_finishes(self) -> None:
|
|
self.client._initial_sync_active = True
|
|
self.client._last_pull_meta = {"has_more": True, "initial_sync_active_server": True}
|
|
with patch.object(
|
|
self.client,
|
|
"_request_json",
|
|
return_value={
|
|
"status_code": 200,
|
|
"body": {
|
|
"initial_sync_active": False,
|
|
"server_now": "2026-01-01T00:00:00+00:00",
|
|
},
|
|
},
|
|
):
|
|
self.client.confirm_tasks(["task-1"])
|
|
self.assertFalse(self.client._last_pull_meta.get("has_more"))
|
|
self.assertFalse(self.client._last_pull_meta.get("initial_sync_active_server"))
|
|
|
|
@patch.object(SyncClient, "confirm_tasks")
|
|
@patch.object(SyncClient, "pull_changes")
|
|
@patch.object(SyncClient, "push_changes")
|
|
@patch.object(SyncClient, "collect_local_changes")
|
|
def test_sync_cycle_skips_confirm_when_apply_raises(
|
|
self,
|
|
mock_collect: Mock,
|
|
mock_push: Mock,
|
|
mock_pull: Mock,
|
|
mock_confirm: Mock,
|
|
) -> None:
|
|
mock_collect.return_value = []
|
|
mock_push.return_value = {"success": True, "total_conflicts": 0}
|
|
mock_pull.return_value = [
|
|
{
|
|
"id": "t1",
|
|
"table_name": "component",
|
|
"record_id": "c1",
|
|
"action": "update",
|
|
"data": {"id": "c1"},
|
|
},
|
|
]
|
|
|
|
def fail_apply(changes):
|
|
raise RuntimeError("simulated apply failure")
|
|
|
|
self.client.on_remote_changes = fail_apply
|
|
self.client._last_pull_integrity_ok = True
|
|
self.client.sync_cycle()
|
|
mock_confirm.assert_not_called()
|
|
|
|
|
|
class SyncClientLocalQueueConfig(TestingConfig):
|
|
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-sync-client-queue-tests-")
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}"
|
|
SQLALCHEMY_BINDS = {
|
|
"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}",
|
|
}
|
|
|
|
|
|
class SyncClientLocalQueueTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app(SyncClientLocalQueueConfig)
|
|
self.ctx = self.app.app_context()
|
|
self.ctx.push()
|
|
db.create_all()
|
|
self.client = SyncClient()
|
|
self.client.client_id = "client-local-1"
|
|
_attach_local_db_push(self.client, self.app)
|
|
|
|
def tearDown(self) -> None:
|
|
db.session.remove()
|
|
db.drop_all()
|
|
self.ctx.pop()
|
|
|
|
def test_collect_local_changes_includes_reports_and_master_tables(self) -> None:
|
|
report = LoadingReport(
|
|
id="lr-local-1",
|
|
recipe_id="recipe-1",
|
|
recipe_name="R",
|
|
start_time=datetime.now(),
|
|
dispenser_type="dispenser",
|
|
)
|
|
db.session.add(report)
|
|
db.session.flush()
|
|
db.session.add(
|
|
LoadingReportComponent(
|
|
id="lrc-local-1",
|
|
report_id=report.id,
|
|
component_name="C1",
|
|
target_weight=1.0,
|
|
actual_weight=1.0,
|
|
loading_order=1,
|
|
)
|
|
)
|
|
db.session.add(
|
|
Component(
|
|
id="cmp-local-1",
|
|
name="MillComponent",
|
|
type="grain",
|
|
is_active=True,
|
|
dry_matter=1.0,
|
|
protein=0.0,
|
|
energy=0.0,
|
|
price=0.0,
|
|
)
|
|
)
|
|
db.session.add(
|
|
Recipe(
|
|
id="recipe-mill-1",
|
|
name="Mill recipe",
|
|
heads_per_trip=10,
|
|
mixing_time=5,
|
|
trip_percent=100.0,
|
|
)
|
|
)
|
|
db.session.commit()
|
|
|
|
changes = self.client.collect_local_changes()
|
|
|
|
self.assertEqual(
|
|
{(c["table_name"], c["record_id"]) for c in changes},
|
|
{
|
|
("component", "cmp-local-1"),
|
|
("recipe", "recipe-mill-1"),
|
|
("loading_report", "lr-local-1"),
|
|
("loading_report_component", "lrc-local-1"),
|
|
},
|
|
)
|
|
table_order = [c["table_name"] for c in changes]
|
|
self.assertLess(
|
|
table_order.index("component"),
|
|
table_order.index("recipe"),
|
|
)
|
|
for change in changes:
|
|
row = db.session.get(SyncQueue, change["task_id"])
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row.status, "processing")
|
|
|
|
def test_push_task_sort_key_orders_master_before_reports(self) -> None:
|
|
from app.models import SyncQueue
|
|
from app.timeutil import utc_now_naive
|
|
|
|
now = utc_now_naive()
|
|
report_task = SyncQueue(
|
|
id="t-report",
|
|
table_name="loading_report",
|
|
record_id="lr-1",
|
|
action="create",
|
|
status="pending",
|
|
priority=9,
|
|
created_at=now,
|
|
)
|
|
recipe_task = SyncQueue(
|
|
id="t-recipe",
|
|
table_name="recipe",
|
|
record_id="r-1",
|
|
action="create",
|
|
status="pending",
|
|
priority=1,
|
|
created_at=now,
|
|
)
|
|
ordered = sorted([report_task, recipe_task], key=_push_task_sort_key)
|
|
self.assertEqual(ordered[0].table_name, "recipe")
|
|
self.assertEqual(ordered[1].table_name, "loading_report")
|
|
|
|
def test_finalize_local_push_updates_task_statuses(self) -> None:
|
|
report = LoadingReport(
|
|
id="lr-local-2",
|
|
recipe_id="recipe-2",
|
|
recipe_name="R2",
|
|
start_time=datetime.now(),
|
|
dispenser_type="dispenser",
|
|
)
|
|
db.session.add(report)
|
|
db.session.commit()
|
|
|
|
changes = self.client.collect_local_changes()
|
|
self.assertEqual(len(changes), 1)
|
|
task_id = changes[0]["task_id"]
|
|
|
|
self.client.on_local_push_result(changes, {"success": False, "message": "network down"})
|
|
db.session.remove()
|
|
task = db.session.get(SyncQueue, task_id)
|
|
self.assertIsNotNone(task)
|
|
self.assertEqual(task.status, "pending")
|
|
self.assertEqual(task.retry_count, 1)
|
|
self.assertEqual(task.error_message, "network down")
|
|
|
|
changes = self.client.collect_local_changes()
|
|
self.client.on_local_push_result(changes, {"success": True})
|
|
db.session.remove()
|
|
task = db.session.get(SyncQueue, task_id)
|
|
self.assertIsNotNone(task)
|
|
self.assertEqual(task.status, "completed")
|
|
self.assertIsNotNone(task.completed_at)
|
|
|
|
def test_first_bootstrap_not_set_when_has_more(self) -> None:
|
|
self.client._initial_sync_active = True
|
|
self.client._last_pull_meta = {
|
|
"has_more": True,
|
|
"initial_sync_active_server": False,
|
|
}
|
|
self.client._apply_initial_sync_from_body({"initial_sync_active": False})
|
|
self.assertTrue(self.client._initial_sync_active)
|
|
|
|
@patch("sync_client.write_sync_client_state")
|
|
def test_first_bootstrap_set_when_server_done_and_no_has_more(
|
|
self, mock_write: Mock
|
|
) -> None:
|
|
self.client._initial_sync_active = True
|
|
self.client._last_pull_meta = {
|
|
"has_more": False,
|
|
"initial_sync_active_server": False,
|
|
}
|
|
self.client._last_cycle_apply_failed = False
|
|
self.client._apply_initial_sync_from_body(
|
|
{"initial_sync_active": False, "server_now": "2026-01-01T00:00:00+00:00"}
|
|
)
|
|
self.assertFalse(self.client._initial_sync_active)
|
|
mock_write.assert_called()
|
|
|
|
@patch("sync_client.requests.post", side_effect=requests.ConnectionError("offline"))
|
|
def test_pull_changes_handles_transport_exception(self, _mock_post: Mock) -> None:
|
|
pulled = self.client.pull_changes()
|
|
self.assertEqual(pulled, [])
|
|
|
|
@patch.object(SyncClient, "sync_cycle")
|
|
@patch.object(SyncClient, "register", return_value=False)
|
|
def test_start_sync_with_retry_does_not_start_when_register_fails(
|
|
self, _mock_register: Mock, _mock_cycle: Mock
|
|
) -> None:
|
|
self.client.config["role"] = "client"
|
|
self.client.server_url = "http://example.local"
|
|
ok = self.client.start_sync_with_retry()
|
|
self.assertFalse(ok)
|
|
self.assertFalse(self.client.is_running)
|
|
_mock_cycle.assert_not_called()
|
|
|
|
@patch("sync_client.write_sync_client_state")
|
|
@patch("sync_client.read_sync_client_state")
|
|
@patch("sync_client.resolve_effective_sync_role", return_value=("server", False))
|
|
def test_constructor_promotes_to_client_when_url_and_id_present(
|
|
self, _mock_role: Mock, mock_read: Mock, _mock_write: Mock
|
|
) -> None:
|
|
mock_read.return_value = {
|
|
"role": "server",
|
|
"server_url": "http://komton.local",
|
|
"client_id": "term1-id",
|
|
}
|
|
sc = SyncClient()
|
|
self.assertEqual(sc.role, "client")
|
|
|
|
|
|
class SyncClientRealApplyConfig(TestingConfig):
|
|
_TMP_DIR = tempfile.mkdtemp(prefix="wesp-sync-real-apply-")
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'server.db')}"
|
|
SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports.db')}"}
|
|
AUTH_LOGIN = "sync-apply-admin"
|
|
AUTH_PASSWORD = "sync-apply-secret"
|
|
|
|
|
|
class SyncClientRealApplyTests(unittest.TestCase):
|
|
"""SyncDualInstanceHarness + _attach_local_db_apply: pull → apply → confirm."""
|
|
|
|
def setUp(self) -> None:
|
|
from tests.helpers.sync_dual_harness import SyncDualInstanceHarness
|
|
|
|
self.harness = SyncDualInstanceHarness()
|
|
self.harness.start()
|
|
self.component_id = str(uuid.uuid4())
|
|
with self.harness.server_ctx():
|
|
db.session.add(
|
|
Component(
|
|
id=self.component_id,
|
|
name="PullMe",
|
|
type="grain",
|
|
dry_matter=50.0,
|
|
protein=0.0,
|
|
energy=0.0,
|
|
price=0.0,
|
|
)
|
|
)
|
|
db.session.commit()
|
|
from app.services.sync_manager import enqueue_sync_queue_task
|
|
|
|
enqueue_sync_queue_task(
|
|
"component", self.component_id, "create", priority=2, target_node_id=None
|
|
)
|
|
db.session.commit()
|
|
|
|
self.sync_client = SyncClient()
|
|
self.sync_client.server_url = self.harness.server_url
|
|
self.sync_client.client_id = self.harness.NODE_A
|
|
self.sync_client.role = "client"
|
|
self.sync_client.config["role"] = "client"
|
|
self.sync_client._initial_sync_active = False
|
|
_attach_local_db_apply(self.sync_client, self.harness.term_a_app)
|
|
|
|
def tearDown(self) -> None:
|
|
self.harness.stop()
|
|
|
|
def test_pull_apply_confirm_brings_component_to_client_db(self) -> None:
|
|
result = self.sync_client.sync_cycle()
|
|
self.assertTrue(result.get("success", True))
|
|
self.assertGreater(int(result.get("pulled") or 0), 0)
|
|
|
|
with self.harness.terminal_ctx("a"):
|
|
row = db.session.get(Component, self.component_id)
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row.name, "PullMe")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|