linusandClaude Sonnet 5
03f6fdb0d9
* Datenmodell + Alembic-Initialmigration (Plan §4), Seed Saison 2026 * Wertungs-Engine (Plan §3.3), regelbasiert je Kategorie/Saison - Abnahmetest gegen die Ergebnisliste 2026-05-31 * Meisterschaftsberechnung mit Streichresultaten + Tie-Break (PLATZHALTER-Punktetabelle, O-1 offen) * Ingest-API (idempotent, dry_run) inkl. Excel-Leser und Fahrer-Matching/Merge * Zeitmessungs-Endpoint (Batch, idempotent, unzugeordnet/unplausibel) + SSE-Anzeige - Mess-Pi-Client bewusst NICHT enthalten (docs/zeitmessung-endpoint.md) * Frontend: öffentliche Seiten + internes Backend (Jinja2/HTMX, kein Build) * Auslieferung: Docker (multi-arch) + docker-compose sowie Debian-Paket/APT-Repo * 57 Tests grün gegen echte PostgreSQL-Testdatenbank Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
"""Geraete-API / Zeitmessung (Plan §3.6, §6.2, AP 5).
|
|
|
|
Abnahmetests: Batch doppelt senden -> keine Dubletten; Messung ohne zuordenbare
|
|
Startnummer -> gespeichert als 'unzugeordnet', nie verworfen.
|
|
"""
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from oamc.models import Geraet, Zeitmessung
|
|
from oamc.models.enums import GeraetTyp, ZeitmessungStatus
|
|
from oamc.seeds import seed_saison_2026
|
|
from oamc.security.tokens import erzeuge_geraete_token
|
|
from tests import factories as f
|
|
|
|
HEAD_ADMIN = None # ueber Cookie unten
|
|
|
|
|
|
def _setup(client):
|
|
db = client.db
|
|
seed_saison_2026(db)
|
|
v = f.veranstaltung(db)
|
|
t = f.turnier(db, v, aktiv=True)
|
|
k = f.klasse(db, "J 3")
|
|
fahrer = f.fahrer(db, "Freund", "Philipp", 2014)
|
|
f.start(db, t, fahrer, k, startnummer=7)
|
|
roh, token_hash = erzeuge_geraete_token()
|
|
g = Geraet(bezeichnung="lichtschranke-ziel-01", typ=GeraetTyp.LICHTSCHRANKE, token_hash=token_hash)
|
|
db.add(g)
|
|
db.flush()
|
|
return t.id, roh
|
|
|
|
|
|
def _msg(turnier_id, startnummer, dauer="159.16", mid=None):
|
|
return {
|
|
"id": str(mid or uuid.uuid4()),
|
|
"turnier_id": turnier_id,
|
|
"startnummer_gemeldet": startnummer,
|
|
"messpunkt": "durchgang-1",
|
|
"dauer_sekunden": dauer,
|
|
"gemessen_am": datetime.now(UTC).isoformat(),
|
|
"monotonic_ns": 884213771004,
|
|
"roh": {"trigger_start_ns": 884054611004, "trigger_ziel_ns": 884213771004},
|
|
}
|
|
|
|
|
|
def test_ohne_token_401(client):
|
|
r = client.post("/api/v1/zeitmessungen", json={"messungen": [_msg(1, 7)]})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_messung_wird_angenommen_und_zugeordnet(client):
|
|
tid, token = _setup(client)
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = client.post("/api/v1/zeitmessungen", json={"messungen": [_msg(tid, 7)]}, headers=h)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["angenommen"] == 1
|
|
assert body["quittungen"][0]["status"] == "angenommen"
|
|
assert body["quittungen"][0]["start_id"] is not None
|
|
|
|
|
|
def test_batch_doppelt_senden_keine_dubletten(client):
|
|
tid, token = _setup(client)
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
mid = uuid.uuid4()
|
|
batch = {"messungen": [_msg(tid, 7, mid=mid)]}
|
|
client.post("/api/v1/zeitmessungen", json=batch, headers=h)
|
|
r2 = client.post("/api/v1/zeitmessungen", json=batch, headers=h)
|
|
assert r2.json()["duplikate"] == 1
|
|
anzahl = client.db.scalar(select(func.count()).select_from(Zeitmessung))
|
|
assert anzahl == 1
|
|
|
|
|
|
def test_unbekannte_startnummer_wird_unzugeordnet_gespeichert(client):
|
|
tid, token = _setup(client)
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = client.post("/api/v1/zeitmessungen", json={"messungen": [_msg(tid, 999)]}, headers=h)
|
|
body = r.json()
|
|
assert body["unzugeordnet"] == 1
|
|
m = client.db.scalars(select(Zeitmessung)).one()
|
|
assert m.status == ZeitmessungStatus.UNZUGEORDNET
|
|
assert m.start_id is None # nie verworfen, nur ohne Zuordnung
|
|
|
|
|
|
def test_unplausible_dauer_wird_markiert_nicht_abgelehnt(client):
|
|
tid, token = _setup(client)
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
r = client.post("/api/v1/zeitmessungen", json={"messungen": [_msg(tid, 7, dauer="3.00")]}, headers=h)
|
|
body = r.json()
|
|
assert body["unplausibel"] == 1
|
|
m = client.db.scalars(select(Zeitmessung)).one()
|
|
assert m.status == ZeitmessungStatus.UNPLAUSIBEL
|
|
|
|
|
|
def test_summe_zeit_wird_aus_messungen_gebildet(client):
|
|
tid, token = _setup(client)
|
|
h = {"Authorization": f"Bearer {token}"}
|
|
# zwei Messpunkte -> Summe
|
|
m1 = _msg(tid, 7, dauer="80.00")
|
|
m2 = _msg(tid, 7, dauer="79.16")
|
|
m2["messpunkt"] = "durchgang-2"
|
|
client.post("/api/v1/zeitmessungen", json={"messungen": [m1, m2]}, headers=h)
|
|
|
|
from oamc.wertung.service import summe_zeit_fuer_start
|
|
|
|
start_id = client.db.scalars(select(Zeitmessung.start_id)).first()
|
|
assert summe_zeit_fuer_start(client.db, start_id) == __import__("decimal").Decimal("159.16")
|
|
|
|
|
|
def test_startliste_endpunkt_braucht_token(client):
|
|
tid, token = _setup(client)
|
|
assert client.get("/api/v1/turniere/aktiv/startliste").status_code == 401
|
|
r = client.get("/api/v1/turniere/aktiv/startliste", headers={"Authorization": f"Bearer {token}"})
|
|
assert r.status_code == 200
|
|
assert r.json()["starts"][0]["startnummer"] == 7
|