| 14 | |
| 15 | |
| 16 | class TestCeleryAdapter(unittest.TestCase): |
| 17 | def test_install_kombu_transport_alias_registers_vercel_alias(self) -> None: |
| 18 | from kombu.transport import TRANSPORT_ALIASES # type: ignore[import-untyped] |
| 19 | |
| 20 | with patch.dict(TRANSPORT_ALIASES, {}, clear=True): |
| 21 | vwc_transport.install_kombu_transport_alias() |
| 22 | self.assertIn("vercel", TRANSPORT_ALIASES) |
| 23 | |
| 24 | def test_extract_task_from_kombu_message_supports_common_shapes(self) -> None: |
| 25 | # Publishing relies on converting Kombu/Celery messages into a safe, JSON envelope |
| 26 | |
| 27 | cases = [ |
| 28 | ( |
| 29 | "protocol-v2-minimal", |
| 30 | { |
| 31 | "headers": { |
| 32 | "task": "tasks.add", |
| 33 | "id": "task-123", |
| 34 | "retries": 2, |
| 35 | "eta": "2025-01-01T00:00:10Z", |
| 36 | "expires": "2025-01-01T00:10:00Z", |
| 37 | }, |
| 38 | "body": [[1, 2], {"k": "v"}, None], |
| 39 | "properties": {"correlation_id": "task-123"}, |
| 40 | }, |
| 41 | { |
| 42 | "task": "tasks.add", |
| 43 | "id": "task-123", |
| 44 | "args": [1, 2], |
| 45 | "kwargs": {"k": "v"}, |
| 46 | "retries": 2, |
| 47 | "eta": "2025-01-01T00:00:10Z", |
| 48 | "expires": "2025-01-01T00:10:00Z", |
| 49 | }, |
| 50 | ), |
| 51 | ( |
| 52 | "protocol-v1-like-body-dict", |
| 53 | { |
| 54 | "headers": {}, |
| 55 | "body": { |
| 56 | "task": "tasks.add", |
| 57 | "id": "task-456", |
| 58 | "args": [3], |
| 59 | "kwargs": {"y": 4}, |
| 60 | }, |
| 61 | "properties": {}, |
| 62 | }, |
| 63 | { |
| 64 | "task": "tasks.add", |
| 65 | "id": "task-456", |
| 66 | "args": [3], |
| 67 | "kwargs": {"y": 4}, |
| 68 | "retries": 0, |
| 69 | }, |
| 70 | ), |
| 71 | ] |
| 72 | |
| 73 | for name, message, expected in cases: |
nothing calls this directly
no outgoing calls
no test coverage detected