73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
import json
|
|
import asyncio
|
|
|
|
import pytest
|
|
from websockets.asyncio.server import serve
|
|
|
|
from property_bot.config import PropertyApiConfig
|
|
from property_bot.consumer import ListingEventConsumer
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumer_ignores_service_ready_and_forwards_listing() -> None:
|
|
received = []
|
|
|
|
async def sink(listing: object) -> None:
|
|
received.append(listing)
|
|
|
|
consumer = ListingEventConsumer(
|
|
PropertyApiConfig("ws://localhost:3000/ws", 1, 1, 2), sink # type: ignore[arg-type]
|
|
)
|
|
assert await consumer.handle_message(json.dumps({"type": "service.ready"})) is False
|
|
assert (
|
|
await consumer.handle_message(
|
|
json.dumps(
|
|
{
|
|
"type": "listing.new",
|
|
"listing": {
|
|
"externalId": "abc",
|
|
"url": "https://example.test/a",
|
|
"title": "A",
|
|
"images": [],
|
|
},
|
|
}
|
|
)
|
|
)
|
|
) is True
|
|
assert len(received) == 1
|
|
assert consumer.status.last_event_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consumer_reads_listing_from_websocket_server(unused_tcp_port: int) -> None:
|
|
received = asyncio.Event()
|
|
|
|
async def sink(_: object) -> None:
|
|
received.set()
|
|
|
|
async def handler(websocket: object) -> None:
|
|
await websocket.send( # type: ignore[attr-defined]
|
|
json.dumps(
|
|
{
|
|
"type": "listing.new",
|
|
"listing": {
|
|
"externalId": "from-socket",
|
|
"url": "https://example.test/a",
|
|
"title": "From socket",
|
|
},
|
|
}
|
|
)
|
|
)
|
|
|
|
async with serve(handler, "127.0.0.1", unused_tcp_port):
|
|
consumer = ListingEventConsumer(
|
|
PropertyApiConfig(f"ws://127.0.0.1:{unused_tcp_port}", 1, 0.01, 0.02), sink # type: ignore[arg-type]
|
|
)
|
|
task = asyncio.create_task(consumer.run())
|
|
try:
|
|
await asyncio.wait_for(received.wait(), timeout=2)
|
|
finally:
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|