40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.models import CurrentStatus, MonitorCreate, MonitorUpdate, State
|
|
from app.store import MonitorStore
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_creates_are_not_lost() -> None:
|
|
store = MonitorStore()
|
|
await asyncio.gather(*[
|
|
store.create(MonitorCreate(name=f"m-{number}", url=f"https://example.com/{number}"))
|
|
for number in range(100)
|
|
])
|
|
assert len(await store.list()) == 100
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_check_cannot_overwrite_updated_monitor() -> None:
|
|
store = MonitorStore()
|
|
item = await store.create(MonitorCreate(name="old", url="https://example.com"))
|
|
snapshot = await store.snapshot(item.id)
|
|
assert snapshot is not None
|
|
_, revision = snapshot
|
|
await store.update(item.id, MonitorUpdate(name="new"))
|
|
recorded = await store.record_status(item.id, revision, CurrentStatus(state=State.UP))
|
|
assert recorded is None
|
|
current = await store.get(item.id)
|
|
assert current is not None and current.current_status.state == State.UNKNOWN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_defensive_copy_does_not_mutate_store() -> None:
|
|
store = MonitorStore()
|
|
item = await store.create(MonitorCreate(name="safe", url="https://example.com"))
|
|
item.name = "outside"
|
|
stored = await store.get(item.id)
|
|
assert stored is not None and stored.name == "safe"
|