23 lines
700 B
Python
23 lines
700 B
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.models import CurrentStatus, MonitorCreate, MonitorState
|
|
from app.store import MonitorStore
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_concurrent_status_updates_are_valid() -> None:
|
|
store = MonitorStore()
|
|
monitor = await store.create(MonitorCreate(name="x", target_url="https://example.com"))
|
|
|
|
async def set_one(code: int) -> None:
|
|
await store.set_status(
|
|
monitor.id,
|
|
CurrentStatus(state=MonitorState.UP, latency_ms=1, http_status=code),
|
|
)
|
|
|
|
await asyncio.gather(*(set_one(code) for code in range(200, 210)))
|
|
saved = await store.get(monitor.id)
|
|
assert saved.current_status.http_status in range(200, 210)
|