Class TestQueueClients

python/vercel-workers/tests/test_client_and_callback.py:616–777

class TestQueueClients(unittest.TestCase):
    def setUp(self) -> None:
        _FakeHttpxClient.captured_bodies.clear()
        _FakeHttpxClient.captured_headers.clear()
        _FakeHttpxClient.captured_urls.clear()
        _FakeAsyncHttpxClient.captured_bodies.clear()
        _FakeAsyncHttpxClient.captured_headers.clear()
        _FakeAsyncHttpxClient.captured_urls.clear()

    def test_queue_client_uses_constructor_configuration(self) -> None:
        client = queue_client_impl.QueueClient(
            region="sfo1",
            token="tok",
            deployment_id="dpl_123",
            headers={"X-Client": "client"},
        )

        with (
            patch.dict(queue_service.os.environ, {}, clear=True),
            patch.object(queue_service.httpx, "Client", _FakeHttpxClient),
        ):
            client.send(
                "orders",
                {"ok": True},
                idempotency_key="key-1",
                headers={"X-Call": "call"},
            )

        self.assertEqual(
            _FakeHttpxClient.captured_urls[-1],
            "https://sfo1.vercel-queue.com/api/v3/topic/orders",
        )
        self.assertEqual(json.loads(_FakeHttpxClient.captured_bodies[-1]), {"ok": True})
        headers = _FakeHttpxClient.captured_headers[-1]
        self.assertEqual(headers["Authorization"], "Bearer tok")
        self.assertEqual(headers["Vqs-Deployment-Id"], "dpl_123")
        self.assertEqual(headers["Vqs-Idempotency-Key"], "key-1")
        self.assertEqual(headers["X-Client"], "client")
        self.assertEqual(headers["X-Call"], "call")

    def test_queue_client_preserves_dev_proxy_base_url(self) -> None:
        client = queue_client_impl.QueueClient(region="sfo1", token="tok", deployment_id=None)

        with (
            patch.dict(
                queue_service.os.environ,
                {"VERCEL_QUEUE_BASE_URL": "http://localhost:3000/_svc/_queues"},
                clear=True,
            ),
            patch.object(queue_service.httpx, "Client", _FakeHttpxClient),
        ):
            client.send("orders", {"ok": True})

        self.assertEqual(
            _FakeHttpxClient.captured_urls[-1],
            "http://localhost:3000/_svc/_queues/api/v3/topic/orders",
        )