import gzip import json import os import tempfile import unittest from datetime import datetime from unittest.mock import patch from app import create_app, db from app.models import Component, SyncClient, SyncDelivery, SyncQueue from config import TestingConfig class SyncRoutesTestConfig(TestingConfig): _TMP_DIR = tempfile.mkdtemp(prefix="wesp-sync-routes-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 SyncRoutesIntegrationTests(unittest.TestCase): def setUp(self) -> None: self.app = create_app(SyncRoutesTestConfig) self.client = self.app.test_client() self.ctx = self.app.app_context() self.ctx.push() db.create_all() def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() def test_push_server_master_skips_stale_version_via_http(self) -> None: db.session.add(Component(id="cmp-1", name="server-name", version=2)) db.session.commit() payload = { "client_id": "http-client-a", "changes": [ { "table_name": "component", "record_id": "cmp-1", "action": "update", "data": {"id": "cmp-1", "name": "client-name", "version": 1}, }, { "table_name": "component", "record_id": "cmp-2", "action": "create", "data": {"id": "cmp-2", "name": "new-component"}, }, ], } resp = self.client.post("/api/sync/push", json=payload) body = resp.get_json() self.assertEqual(resp.status_code, 200) self.assertIsInstance(body, dict) self.assertTrue(body.get("success")) self.assertEqual(body.get("total_applied"), 1) cmp1 = db.session.get(Component, "cmp-1") cmp2 = db.session.get(Component, "cmp-2") self.assertIsNotNone(cmp1) self.assertEqual(cmp1.name, "server-name") self.assertIsNotNone(cmp2) self.assertEqual(cmp2.name, "new-component") def test_push_accepts_gzip_json_body(self) -> None: raw = json.dumps({"client_id": "http-client-gzip", "changes": []}).encode("utf-8") compressed = gzip.compress(raw) resp = self.client.post( "/api/sync/push", data=compressed, headers={"Content-Encoding": "gzip"}, ) body = resp.get_json() self.assertEqual(resp.status_code, 200) self.assertIsInstance(body, dict) self.assertTrue(body.get("success")) self.assertEqual(body.get("total_applied"), 0) self.assertEqual(body.get("total_conflicts"), 0) def test_confirm_universal_task_requires_delivery_to_all_active_clients(self) -> None: now = datetime.now() c1 = SyncClient(node_id="node-1", client_name="client-1", status="active", is_enabled=True) c2 = SyncClient(node_id="node-2", client_name="client-2", status="active", is_enabled=True) db.session.add_all([c1, c2]) db.session.flush() task = SyncQueue( id="task-http-1", table_name="component", record_id="cmp-http-x", action="update", status="processing", target_node_id=None, created_at=now, processed_at=now, ) db.session.add(task) db.session.add(SyncDelivery(client_id=c1.id, task_id=task.id, delivered_at=now)) db.session.commit() resp1 = self.client.post( "/api/sync/confirm", json={"client_id": "node-1", "task_ids": [task.id]}, ) body1 = resp1.get_json() self.assertEqual(resp1.status_code, 200) self.assertEqual(body1.get("updated"), 0) self.assertEqual(db.session.get(SyncQueue, task.id).status, "processing") db.session.add(SyncDelivery(client_id=c2.id, task_id=task.id, delivered_at=now)) db.session.commit() resp2 = self.client.post( "/api/sync/confirm", json={"client_id": "node-1", "task_ids": [task.id]}, ) body2 = resp2.get_json() self.assertEqual(resp2.status_code, 200) self.assertEqual(body2.get("updated"), 1) self.assertEqual(db.session.get(SyncQueue, task.id).status, "completed") def test_client_log_upload_gzip_binary(self) -> None: raw = b"wesp log line\n" resp = self.client.post( "/api/sync/client-log", data=gzip.compress(raw), headers={ "X-WESP-Client-Id": "client-log-node", "X-WESP-Log-Name": "wesp.log", "Content-Encoding": "gzip", }, ) self.assertEqual(resp.status_code, 200, resp.get_data(as_text=True)) body = resp.get_json() self.assertTrue(body.get("success")) self.assertGreater(body.get("bytes_written", 0), 0) def test_client_log_upload_json_text(self) -> None: resp = self.client.post( "/api/sync/client-log", json={ "client_id": "client-json", "log_name": "mini.log", "text": "hello", }, ) self.assertEqual(resp.status_code, 200) self.assertTrue(resp.get_json().get("success")) def test_client_log_upload_rejects_bad_secret(self) -> None: self.app.config["WESP_CLIENT_LOG_UPLOAD_SECRET"] = "s3cret" resp = self.client.post( "/api/sync/client-log", data=gzip.compress(b"x"), headers={ "X-WESP-Client-Id": "n1", "Content-Encoding": "gzip", "X-WESP-Client-Log-Secret": "wrong", }, ) self.assertEqual(resp.status_code, 401) def test_client_log_upload_disabled(self) -> None: self.app.config["WESP_CLIENT_LOG_UPLOAD_ENABLED"] = False resp = self.client.post( "/api/sync/client-log", json={"client_id": "x", "text": "a"}, ) self.assertEqual(resp.status_code, 403) @patch("app.routes.sync.SyncManager.process_pull") def test_pull_202_sets_retry_after_header(self, mock_pull) -> None: mock_pull.return_value = { "status_code": 202, "payload": { "success": True, "changes": [], "total": 0, "initial_sync_active": True, "retry_after_sec": 3, }, } resp = self.client.post("/api/sync/pull", json={"client_id": "node-202", "limit": 1}) self.assertEqual(resp.status_code, 202) self.assertEqual(resp.headers.get("Retry-After"), "3") if __name__ == "__main__": unittest.main()