decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project.
This commit is contained in:
39
tests/test_store.py
Normal file
39
tests/test_store.py
Normal file
@@ -0,0 +1,39 @@
|
|||||||
|
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"
|
||||||
Reference in New Issue
Block a user