tegwick
6d2ccc7760
Some checks failed
tamq-ci / test (push) Failing after 5s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a03397-4d51-7fd1-8ff2-946eb22ea2bc
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
import asyncio
|
|
import json
|
|
|
|
from tamq.service import (
|
|
DELIVERY_RELIABILITY_CAPABILITY,
|
|
IDEMPOTENT_SEND_CAPABILITY,
|
|
PROTOCOL_CAPTURE_CAPABILITY,
|
|
PUSHY_FRAMING_CAPABILITY,
|
|
Service,
|
|
)
|
|
from tamq.store import Store
|
|
|
|
|
|
def test_incompatible_protocol_rejected(tmp_path):
|
|
async def run():
|
|
path = tmp_path / "tamq.sock"
|
|
service = Service(path, Store(tmp_path / "queue.sqlite3"))
|
|
server = await asyncio.start_unix_server(service.handle, path=str(path))
|
|
try:
|
|
reader, writer = await asyncio.open_unix_connection(str(path))
|
|
writer.write(b'{"op":"ping","protocol":"2.0"}\n'); await writer.drain()
|
|
response = json.loads(await reader.readline())
|
|
assert response["ok"] is False
|
|
writer.close(); await writer.wait_closed()
|
|
finally:
|
|
server.close(); await server.wait_closed(); service.close()
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_ping_advertises_non_routable_pushy_framing(tmp_path):
|
|
async def run():
|
|
path = tmp_path / "tamq.sock"
|
|
service = Service(path, Store(tmp_path / "queue.sqlite3"))
|
|
server = await asyncio.start_unix_server(service.handle, path=str(path))
|
|
try:
|
|
reader, writer = await asyncio.open_unix_connection(str(path))
|
|
writer.write(b'{"op":"ping"}\n')
|
|
await writer.drain()
|
|
response = json.loads(await reader.readline())
|
|
assert PUSHY_FRAMING_CAPABILITY in response["capabilities"]
|
|
assert PROTOCOL_CAPTURE_CAPABILITY in response["capabilities"]
|
|
assert DELIVERY_RELIABILITY_CAPABILITY in response["capabilities"]
|
|
assert IDEMPOTENT_SEND_CAPABILITY in response["capabilities"]
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
finally:
|
|
server.close()
|
|
await server.wait_closed()
|
|
service.close()
|
|
|
|
asyncio.run(run())
|