ComfyUI-OpenClaw/tests/test_sidecar.py

"""
Unit tests for R13 Sidecar Bridge Contract.
"""

import unittest

from services.sidecar.bridge_client import BridgeClient, BridgeClientConfig
from services.sidecar.bridge_contract import (
    BRIDGE_ENDPOINTS,
    REQUIRED_WORKER_SCOPES,
    BridgeDeliveryRequest,
    BridgeHealthResponse,
    BridgeJobRequest,
    BridgeScope,
    DeviceToken,
)


class TestBridgeContract(unittest.TestCase):
    """Tests for R13 sidecar bridge contract."""

    def test_bridge_scopes(self):
        """Test bridge scope enum values."""
        self.assertEqual(BridgeScope.JOB_SUBMIT.value, "job:submit")
        self.assertEqual(BridgeScope.DELIVERY.value, "delivery:send")

    def test_device_token_dataclass(self):
        """Test DeviceToken dataclass."""
        token = DeviceToken(
            device_id="dev123", device_token="secret", scopes=[BridgeScope.JOB_SUBMIT]
        )
        self.assertEqual(token.device_id, "dev123")
        self.assertIn(BridgeScope.JOB_SUBMIT, token.scopes)

    def test_job_request_dataclass(self):
        """Test BridgeJobRequest dataclass."""
        req = BridgeJobRequest(
            template_id="text2img", inputs={"prompt": "test"}, idempotency_key="key123"
        )
        self.assertEqual(req.template_id, "text2img")
        self.assertEqual(req.timeout_sec, 300)  # Default

    def test_delivery_request_dataclass(self):
        """Test BridgeDeliveryRequest dataclass."""
        req = BridgeDeliveryRequest(
            target="discord:123456", text="Hello", idempotency_key="del123"
        )
        self.assertEqual(req.target, "discord:123456")

    def test_health_response_dataclass(self):
        """Test BridgeHealthResponse dataclass."""
        resp = BridgeHealthResponse(ok=True, version="1.0.0", uptime_sec=3600)
        self.assertTrue(resp.ok)
        self.assertEqual(resp.uptime_sec, 3600)

    def test_endpoints_defined(self):
        """Test bridge endpoints are properly defined."""
        self.assertIn("submit", BRIDGE_ENDPOINTS)
        self.assertIn("deliver", BRIDGE_ENDPOINTS)
        self.assertIn("health", BRIDGE_ENDPOINTS)

        # Check submit endpoint structure
        submit = BRIDGE_ENDPOINTS["submit"]
        self.assertEqual(submit["method"], "POST")
        self.assertEqual(submit["path"], "/bridge/submit")
        self.assertEqual(submit["scope"], BridgeScope.JOB_SUBMIT)

    # --- F46 contract alignment ---

    def test_worker_endpoints_defined(self):
        """F46: Worker-facing endpoints must exist in contract."""
        for key in ("worker_poll", "worker_result", "worker_heartbeat"):
            self.assertIn(key, BRIDGE_ENDPOINTS, f"Missing worker endpoint: {key}")

    def test_worker_endpoint_paths(self):
        """F46: Worker endpoints use /bridge/worker/* path convention."""
        self.assertEqual(BRIDGE_ENDPOINTS["worker_poll"]["path"], "/bridge/worker/poll")
        self.assertEqual(
            BRIDGE_ENDPOINTS["worker_result"]["path"], "/bridge/worker/result"
        )
        self.assertEqual(
            BRIDGE_ENDPOINTS["worker_heartbeat"]["path"], "/bridge/worker/heartbeat"
        )

    def test_worker_endpoint_methods(self):
        """F46: Worker endpoints use correct HTTP methods."""
        self.assertEqual(BRIDGE_ENDPOINTS["worker_poll"]["method"], "GET")
        self.assertEqual(BRIDGE_ENDPOINTS["worker_result"]["method"], "POST")
        self.assertEqual(BRIDGE_ENDPOINTS["worker_heartbeat"]["method"], "POST")

    def test_required_worker_scopes(self):
        """F46: Required scopes for sidecar startup are defined."""
        self.assertIn(BridgeScope.JOB_SUBMIT, REQUIRED_WORKER_SCOPES)
        self.assertIn(BridgeScope.JOB_STATUS, REQUIRED_WORKER_SCOPES)


class TestBridgeClient(unittest.TestCase):
    """Tests for R13 sidecar bridge client."""

    def test_client_config_defaults(self):
        """Test client config defaults."""
        config = BridgeClientConfig("http://bridge", "token", "worker1")
        self.assertEqual(config.url, "http://bridge")
        self.assertEqual(config.token, "token")
        self.assertEqual(config.worker_id, "worker1")

    def test_client_not_connected_by_default(self):
        """Test client starts disconnected."""
        client = BridgeClient("http://bridge", "token", "worker1")
        self.assertIsNone(client.session)

    def test_idempotency_key_propagation(self):
        """Test idempotency key is required in requests."""
        req = BridgeJobRequest(template_id="test", inputs={}, idempotency_key="key123")
        # Key should be accessible
        self.assertEqual(req.idempotency_key, "key123")

    def test_device_token_in_request_context(self):
        """Test device_id can be passed in requests."""
        req = BridgeJobRequest(
            template_id="test", inputs={}, idempotency_key="key123", device_id="dev456"
        )
        self.assertEqual(req.device_id, "dev456")

    # --- F46 client endpoint alignment ---

    def test_endpoint_resolver(self):
        """F46: Client resolves contract paths correctly."""
        client = BridgeClient("https://bridge.example.com", "t", "w")
        self.assertEqual(
            client._endpoint("worker_poll"),
            "https://bridge.example.com/bridge/worker/poll",
        )
        self.assertEqual(
            client._endpoint("health"),
            "https://bridge.example.com/bridge/health",
        )

    def test_endpoint_resolver_strips_trailing_slash(self):
        """F46: Client strips trailing slash from base URL."""
        client = BridgeClient("https://bridge.example.com/", "t", "w")
        self.assertEqual(
            client._endpoint("health"),
            "https://bridge.example.com/bridge/health",
        )

    def test_endpoint_resolver_unknown_raises(self):
        """F46: Unknown endpoint name raises ValueError."""
        client = BridgeClient("https://bridge.example.com", "t", "w")
        with self.assertRaises(ValueError):
            client._endpoint("nonexistent")


if __name__ == "__main__":
    unittest.main()