32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.models import CurrentStatus, HealthState, MonitorCreate, MonitorUpdate, utc_now
|
|
from app.store import MonitorStore
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_compare_and_set_rejects_stale_revision() -> None:
|
|
store = MonitorStore()
|
|
item = await store.create(MonitorCreate(name="old", url="https://example.com/a"))
|
|
result = CurrentStatus(state=HealthState.up, checked_at=utc_now(), latency_ms=1, http_status=200)
|
|
|
|
await store.update(item.id, MonitorUpdate(name="new", url="https://example.com/b"))
|
|
assert await store.set_status_if_current(item.id, item.revision, result) is False
|
|
current = await store.get(item.id)
|
|
assert current is not None and current.current_status is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_writes_remain_consistent() -> None:
|
|
store = MonitorStore()
|
|
item = await store.create(MonitorCreate(name="start", url="https://example.com"))
|
|
await asyncio.gather(*[
|
|
store.update(item.id, MonitorUpdate(name=f"name-{index}", url="https://example.com"))
|
|
for index in range(20)
|
|
])
|
|
current = await store.get(item.id)
|
|
assert current is not None
|
|
assert current.revision == 21
|