patrickandClaude Sonnet 5
a1ad2d6cef
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
|
|
from app.models.readiness_snapshot import ReadinessSnapshot, ReadinessZustand
|
|
from app.services.readiness_snapshot import erstelle_snapshot, verlauf_fuer_objekt, verlauf_global
|
|
from tests.conftest import login
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_erstellt_zeile_je_objekt(db_session, standort_factory, objekt_factory):
|
|
standort = await standort_factory("Wache-Ready")
|
|
objekt = await objekt_factory(name="Rucksack Ready", code="RS-READY-1", standort=standort)
|
|
|
|
anzahl = await erstelle_snapshot(db_session)
|
|
assert anzahl >= 1
|
|
|
|
treffer = (
|
|
await db_session.execute(select(ReadinessSnapshot).where(ReadinessSnapshot.objekt_id == objekt.id))
|
|
).scalar_one()
|
|
# noch nie kontrolliert -> unbekannt (READY-002)
|
|
assert treffer.zustand == ReadinessZustand.unbekannt
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_erneuter_lauf_am_selben_tag_ueberschreibt(db_session, standort_factory, objekt_factory):
|
|
standort = await standort_factory("Wache-Ready2")
|
|
await objekt_factory(name="Rucksack Ready2", code="RS-READY-2", standort=standort)
|
|
|
|
erste = await erstelle_snapshot(db_session)
|
|
zweite = await erstelle_snapshot(db_session)
|
|
assert erste == zweite
|
|
|
|
anzahl_zeilen = (
|
|
await db_session.execute(select(func.count()).select_from(ReadinessSnapshot))
|
|
).scalar_one()
|
|
assert anzahl_zeilen == erste
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verlauf_global_und_fuer_objekt(db_session, standort_factory, objekt_factory):
|
|
standort = await standort_factory("Wache-Ready3")
|
|
objekt = await objekt_factory(name="Rucksack Ready3", code="RS-READY-3", standort=standort)
|
|
|
|
gestern = date.today() - timedelta(days=1)
|
|
await erstelle_snapshot(db_session, zeitpunkt=gestern)
|
|
await erstelle_snapshot(db_session)
|
|
|
|
global_verlauf = await verlauf_global(db_session, von=gestern, bis=date.today())
|
|
zeitpunkte = {v["zeitpunkt"] for v in global_verlauf}
|
|
assert gestern in zeitpunkte and date.today() in zeitpunkte
|
|
|
|
objekt_verlauf = await verlauf_fuer_objekt(db_session, objekt.id)
|
|
assert len(objekt_verlauf) == 2
|
|
assert all(p["zustand"] == "unbekannt" for p in objekt_verlauf)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readiness_verlauf_endpunkt_erfordert_leitung(client, admin_user, mitarbeiter_user):
|
|
token_mitarbeiter = await login(client, "mitarbeiter1")
|
|
verboten = await client.get(
|
|
"/api/v1/readiness/verlauf", headers={"Authorization": f"Bearer {token_mitarbeiter}"}
|
|
)
|
|
assert verboten.status_code == 403
|
|
|
|
token_admin = await login(client, "admin1")
|
|
erlaubt = await client.get(
|
|
"/api/v1/readiness/verlauf", headers={"Authorization": f"Bearer {token_admin}"}
|
|
)
|
|
assert erlaubt.status_code == 200
|