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