site/WESP_REL/tests/test_sync_routes.py

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