Erstumsetzung: OAMC Turnierauswertung (AP 0–7, 9)
* 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>
This commit is contained in:
linus
@@ -0,0 +1,106 @@
|
||||
"""Pytest-Fixtures.
|
||||
|
||||
Nutzt eine echte PostgreSQL-Testdatenbank (Plan §5.2: kein SQLite, wegen JSONB /
|
||||
ARRAY / echten Constraints). URL ueber ``OAMC_DATABASE_URL`` oder Default
|
||||
``postgresql+psycopg://linus:linus@localhost:5432/oamc_test``.
|
||||
|
||||
Jeder Test laeuft in einer aussen zurueckgerollten Transaktion — die DB bleibt
|
||||
sauber, Migrationen laufen einmalig.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
os.environ.setdefault(
|
||||
"OAMC_DATABASE_URL", "postgresql+psycopg://linus:linus@localhost:5432/oamc_test"
|
||||
)
|
||||
os.environ.setdefault("OAMC_SECRET_KEY", "test-secret-key-0123456789abcdef0123")
|
||||
os.environ.setdefault("OAMC_INGEST_API_KEYS", "test-key")
|
||||
|
||||
PROJEKT_WURZEL = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def _migriert() -> None:
|
||||
"""Schema einmal pro Testlauf frisch aufbauen."""
|
||||
|
||||
url = os.environ["OAMC_DATABASE_URL"]
|
||||
admin = create_engine(url, isolation_level="AUTOCOMMIT")
|
||||
with admin.connect() as conn:
|
||||
conn.execute(text("DROP SCHEMA public CASCADE"))
|
||||
conn.execute(text("CREATE SCHEMA public"))
|
||||
admin.dispose()
|
||||
subprocess.run(
|
||||
[sys.executable, "-m", "alembic", "upgrade", "head"],
|
||||
cwd=str(PROJEKT_WURZEL),
|
||||
check=True,
|
||||
env={**os.environ},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def engine(_migriert):
|
||||
from oamc.config import get_settings
|
||||
|
||||
eng = create_engine(get_settings().database_url, future=True)
|
||||
yield eng
|
||||
eng.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db(engine) -> Session:
|
||||
conn = engine.connect()
|
||||
trans = conn.begin()
|
||||
session = Session(bind=conn, join_transaction_mode="create_savepoint")
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
trans.rollback()
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seed_2026(db):
|
||||
from oamc.seeds import seed_saison_2026
|
||||
|
||||
seed_saison_2026(db)
|
||||
db.flush()
|
||||
return db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(engine):
|
||||
"""TestClient mit auf die Testtransaktion gebundener DB-Session."""
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from oamc.db import get_db
|
||||
from oamc.main import app
|
||||
|
||||
conn = engine.connect()
|
||||
trans = conn.begin()
|
||||
session = Session(bind=conn, join_transaction_mode="create_savepoint")
|
||||
|
||||
def _get_db_override():
|
||||
# Kein Rollback bei HTTPException: die aeussere Testtransaktion raeumt auf.
|
||||
# Ein Rollback hier wuerde den Savepoint mitsamt Fixture-Daten verwerfen.
|
||||
yield session
|
||||
session.flush()
|
||||
|
||||
app.dependency_overrides[get_db] = _get_db_override
|
||||
with TestClient(app) as c:
|
||||
c.db = session # type: ignore[attr-defined]
|
||||
yield c
|
||||
app.dependency_overrides.clear()
|
||||
session.close()
|
||||
trans.rollback()
|
||||
conn.close()
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Kleine Helfer zum Aufbau von Testdaten."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from oamc.models import (
|
||||
Ergebnis,
|
||||
Fahrer,
|
||||
Klasse,
|
||||
Start,
|
||||
Turnier,
|
||||
Veranstaltung,
|
||||
Verein,
|
||||
)
|
||||
from oamc.models.enums import StartStatus, TurnierStatus, VeranstaltungStatus
|
||||
|
||||
|
||||
def klasse(db, code="J 3"):
|
||||
return db.scalars(select(Klasse).where(Klasse.code == code, Klasse.saison_jahr == 2026)).one()
|
||||
|
||||
|
||||
def verein(db, name="OAMC Reinheim e.V. im ADAC"):
|
||||
return db.scalars(select(Verein).where(Verein.name == name)).first()
|
||||
|
||||
|
||||
def veranstaltung(db, *, datum=date(2026, 5, 31), titel="Testturnier", status=VeranstaltungStatus.LAUFEND):
|
||||
v = Veranstaltung(datum=datum, titel=titel, ort="Reinheim", status=status, ist_doppelturnier=False)
|
||||
db.add(v)
|
||||
db.flush()
|
||||
return v
|
||||
|
||||
|
||||
def turnier(db, v=None, *, name="Reinheim", zaehlt_meisterschaft=True, aktiv=False,
|
||||
status=TurnierStatus.LAUFEND, flush=True):
|
||||
v = v or veranstaltung(db)
|
||||
t = Turnier(
|
||||
veranstaltung_id=v.id, name=name, zaehlt_meisterschaft=zaehlt_meisterschaft,
|
||||
ist_aktiv=aktiv, status=status,
|
||||
)
|
||||
db.add(t)
|
||||
if flush:
|
||||
db.flush()
|
||||
return t
|
||||
|
||||
|
||||
def fahrer(db, nachname="Mustermann", vorname="Max", geburtsjahr=2014):
|
||||
f = Fahrer(nachname=nachname, vorname=vorname, geburtsjahr=geburtsjahr)
|
||||
db.add(f)
|
||||
db.flush()
|
||||
return f
|
||||
|
||||
|
||||
def start(db, t, f, k, *, startnummer, status=StartStatus.GESTARTET, wertungsklasse_id=None,
|
||||
flush=True):
|
||||
s = Start(
|
||||
turnier_id=t.id, fahrer_id=f.id, klasse_id=k.id, startnummer=startnummer,
|
||||
status=status, wertungsklasse_id=wertungsklasse_id,
|
||||
)
|
||||
db.add(s)
|
||||
if flush:
|
||||
db.flush()
|
||||
return s
|
||||
|
||||
|
||||
def ergebnis(db, s, *, zeit=None, fehler=None, zeit_fertig=True, fehler_fertig=True):
|
||||
e = Ergebnis(
|
||||
start_id=s.id,
|
||||
summe_zeit=zeit,
|
||||
summe_fehler=fehler,
|
||||
zeit_vollstaendig=zeit_fertig,
|
||||
fehler_vollstaendig=fehler_fertig,
|
||||
)
|
||||
db.add(e)
|
||||
db.flush()
|
||||
return e
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Ingest-API (Plan §6.1): Auth, dry_run, Idempotenz, Nachrechnung."""
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from oamc.models import Ergebnis, Start, Turnier
|
||||
from oamc.models.enums import VeranstaltungStatus
|
||||
from oamc.seeds import seed_saison_2026
|
||||
from tests import factories as f
|
||||
|
||||
HEAD = {"X-API-Key": "test-key"}
|
||||
|
||||
|
||||
def _turnier(client):
|
||||
db = client.db
|
||||
seed_saison_2026(db)
|
||||
v = f.veranstaltung(db, status=VeranstaltungStatus.LAUFEND)
|
||||
t = f.turnier(db, v)
|
||||
db.flush()
|
||||
return t.id
|
||||
|
||||
|
||||
def test_ohne_api_key_401(client):
|
||||
r = client.post("/api/v1/turniere/1/ergebnisse", json={"quelle": "x", "idempotenz_schluessel": "k", "ergebnisse": []})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_dry_run_meldet_neue_fahrer_ohne_zu_schreiben(client):
|
||||
tid = _turnier(client)
|
||||
payload = {
|
||||
"quelle": "excel-auswertung-v3",
|
||||
"idempotenz_schluessel": "2026-05-31-reinheim-final",
|
||||
"ergebnisse": [
|
||||
{
|
||||
"startnummer": 32,
|
||||
"fahrer": {"nachname": "Gál-Szász", "vorname": "Péter", "geburtsjahr": 2015,
|
||||
"geschlecht": "M", "verein": "OAMC Reinheim"},
|
||||
"klasse": "J 2", "summe_zeit": "159.16", "summe_fehler": 10,
|
||||
}
|
||||
],
|
||||
}
|
||||
r = client.post(f"/api/v1/turniere/{tid}/ergebnisse?dry_run=true", json=payload, headers=HEAD)
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["dry_run"] is True
|
||||
assert body["neue_fahrer"] == 1
|
||||
# nichts geschrieben
|
||||
assert client.db.scalars(select(Start).where(Start.turnier_id == tid)).first() is None
|
||||
|
||||
|
||||
def test_echter_upload_dann_idempotent(client):
|
||||
tid = _turnier(client)
|
||||
payload = {
|
||||
"quelle": "excel-auswertung-v3",
|
||||
"idempotenz_schluessel": "abc-final",
|
||||
"ergebnisse": [
|
||||
{"startnummer": 1, "fahrer": {"nachname": "Bernius", "vorname": "Meik", "geburtsjahr": 1985},
|
||||
"klasse": "S 6", "wertungsklasse": "S 5, S 9, S 3, S 6, S 7",
|
||||
"summe_zeit": "105.43", "summe_fehler": 0,
|
||||
"gesamt_erwartet": "105.43", "platzierung_erwartet": 1},
|
||||
{"startnummer": 2, "fahrer": {"nachname": "Schmunk", "vorname": "Christian", "geburtsjahr": 1988},
|
||||
"klasse": "S 9", "wertungsklasse": "S 5, S 9, S 3, S 6, S 7",
|
||||
"summe_zeit": "103.26", "summe_fehler": 10,
|
||||
"gesamt_erwartet": "113.26", "platzierung_erwartet": 2},
|
||||
],
|
||||
}
|
||||
r1 = client.post(f"/api/v1/turniere/{tid}/ergebnisse", json=payload, headers=HEAD)
|
||||
assert r1.status_code == 200, r1.text
|
||||
b1 = r1.json()
|
||||
assert b1["angelegt"] == 2
|
||||
assert b1["abweichungen"] == [] # Nachrechnung deckt sich mit gesamt_erwartet
|
||||
|
||||
r2 = client.post(f"/api/v1/turniere/{tid}/ergebnisse", json=payload, headers=HEAD)
|
||||
assert r2.json()["idempotent_uebersprungen"] is True
|
||||
assert client.db.scalars(select(Start).where(Start.turnier_id == tid)).all().__len__() == 2
|
||||
|
||||
|
||||
def test_abweichung_zur_erwartung_wird_gemeldet(client):
|
||||
tid = _turnier(client)
|
||||
payload = {
|
||||
"quelle": "excel", "idempotenz_schluessel": "abw-1",
|
||||
"ergebnisse": [
|
||||
{"startnummer": 1, "fahrer": {"nachname": "A", "vorname": "A", "geburtsjahr": 1985},
|
||||
"klasse": "S 6", "summe_zeit": "100.00", "summe_fehler": 0,
|
||||
"gesamt_erwartet": "999.00", "platzierung_erwartet": 1},
|
||||
],
|
||||
}
|
||||
r = client.post(f"/api/v1/turniere/{tid}/ergebnisse", json=payload, headers=HEAD)
|
||||
body = r.json()
|
||||
assert any(a["feld"] == "gesamt" for a in body["abweichungen"])
|
||||
assert body["warnungen"]
|
||||
|
||||
|
||||
def test_finalisieren_setzt_status_final(client):
|
||||
tid = _turnier(client)
|
||||
client.post(f"/api/v1/turniere/{tid}/ergebnisse", json={
|
||||
"quelle": "x", "idempotenz_schluessel": "f1",
|
||||
"ergebnisse": [
|
||||
{"startnummer": 1, "fahrer": {"nachname": "A", "vorname": "A", "geburtsjahr": 1985},
|
||||
"klasse": "S 6", "summe_zeit": "100.00", "summe_fehler": 0},
|
||||
],
|
||||
}, headers=HEAD)
|
||||
r = client.post(f"/api/v1/turniere/{tid}/finalisieren", headers=HEAD)
|
||||
assert r.status_code == 200
|
||||
assert client.db.get(Turnier, tid).status.value == "final"
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Rauchtest der serverseitig gerenderten Seiten (AP 6/7)."""
|
||||
|
||||
from datetime import date
|
||||
|
||||
from oamc.seeds import seed_saison_2026
|
||||
from tests import factories as f
|
||||
|
||||
|
||||
def test_startseite_rendert(client):
|
||||
r = client.get("/")
|
||||
assert r.status_code == 200
|
||||
assert "Motorrad-Turniere" in r.text
|
||||
|
||||
|
||||
def test_robots_txt_sperrt_fahrerprofile(client):
|
||||
r = client.get("/robots.txt")
|
||||
assert "Disallow: /fahrer/" in r.text
|
||||
|
||||
|
||||
def test_termine_ics(client):
|
||||
seed_saison_2026(client.db)
|
||||
f.veranstaltung(client.db, titel="Doppelturnier", datum=date(2026, 5, 31))
|
||||
r = client.get("/termine.ics")
|
||||
assert r.status_code == 200
|
||||
assert r.headers["content-type"].startswith("text/calendar")
|
||||
assert "BEGIN:VCALENDAR" in r.text
|
||||
|
||||
|
||||
def test_ergebnisliste_und_csv(client):
|
||||
db = client.db
|
||||
seed_saison_2026(db)
|
||||
t = f.turnier(db)
|
||||
k = f.klasse(db, "J 3")
|
||||
fahrer = f.fahrer(db, "Freund", "Philipp", 2014)
|
||||
s = f.start(db, t, fahrer, k, startnummer=7)
|
||||
e = f.ergebnis(db, s, zeit="136.88", fehler=0)
|
||||
e.gesamt = 0
|
||||
e.platzierung = 1
|
||||
db.flush()
|
||||
|
||||
html = client.get(f"/ergebnisse/turnier/{t.id}")
|
||||
assert html.status_code == 200
|
||||
assert "Freund Philipp" in html.text
|
||||
|
||||
csv = client.get(f"/ergebnisse/turnier/{t.id}.csv")
|
||||
assert csv.status_code == 200
|
||||
assert "Freund Philipp" in csv.text
|
||||
assert csv.headers["content-disposition"].endswith('.csv"')
|
||||
|
||||
|
||||
def test_fahrerprofil_hat_noindex(client):
|
||||
db = client.db
|
||||
seed_saison_2026(db)
|
||||
fahrer = f.fahrer(db, "Muster", "Max", 2014)
|
||||
db.flush()
|
||||
r = client.get(f"/fahrer/{fahrer.id}")
|
||||
assert r.status_code == 200
|
||||
assert 'name="robots" content="noindex"' in r.text
|
||||
|
||||
|
||||
def test_intern_ohne_login_leitet_um(client):
|
||||
r = client.get("/intern", follow_redirects=False)
|
||||
assert r.status_code in (307, 401)
|
||||
|
||||
|
||||
def test_login_flow(client):
|
||||
from oamc.models import Benutzer
|
||||
from oamc.models.enums import BenutzerRolle
|
||||
from oamc.security.passwoerter import hash_passwort
|
||||
|
||||
client.db.add(
|
||||
Benutzer(
|
||||
benutzername="tester",
|
||||
anzeigename="Tester",
|
||||
passwort_hash=hash_passwort("geheim123"),
|
||||
rolle=BenutzerRolle.AUSWERTER,
|
||||
)
|
||||
)
|
||||
client.db.flush()
|
||||
r = client.post(
|
||||
"/intern/login",
|
||||
data={"benutzername": "tester", "passwort": "geheim123"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 303
|
||||
assert "oamc_session" in r.cookies
|
||||
@@ -0,0 +1,118 @@
|
||||
"""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
|
||||
@@ -0,0 +1,59 @@
|
||||
"""Fahrer-Matching (Plan §6.4)."""
|
||||
|
||||
from oamc.matching.fahrer import (
|
||||
fahrer_zusammenfuehren,
|
||||
finde_exakt,
|
||||
match_oder_neu,
|
||||
normalisiere,
|
||||
)
|
||||
from tests import factories as f
|
||||
|
||||
|
||||
def test_normalisierung_akzente_und_bindestriche():
|
||||
assert normalisiere("Gál-Szász") == "gal szasz"
|
||||
assert normalisiere("D'Angelo") == "d'angelo"
|
||||
assert normalisiere(" Müller ") == "muller"
|
||||
|
||||
|
||||
def test_exakter_match_ueber_normalisierten_namen(seed_2026):
|
||||
db = seed_2026
|
||||
f.fahrer(db, "Gál-Szász", "Péter", 2015)
|
||||
treffer = finde_exakt(db, "Gal-Szasz", "Peter", 2015)
|
||||
assert treffer is not None
|
||||
assert treffer.nachname == "Gál-Szász"
|
||||
|
||||
|
||||
def test_kein_match_legt_neuen_fahrer_an_und_markiert_neu(seed_2026):
|
||||
db = seed_2026
|
||||
erg = match_oder_neu(db, nachname="Neuling", vorname="Nora", geburtsjahr=2000, anlegen=True)
|
||||
assert erg.status.value == "neu"
|
||||
assert erg.neu_angelegt and erg.fahrer.id is not None
|
||||
|
||||
|
||||
def test_dry_run_legt_nichts_an(seed_2026):
|
||||
db = seed_2026
|
||||
erg = match_oder_neu(db, nachname="Probe", vorname="Peter", anlegen=False)
|
||||
assert erg.status.value in {"neu", "unscharf"}
|
||||
assert erg.fahrer is None
|
||||
|
||||
|
||||
def test_familien_namensgleichheit_trennt_ueber_vorname(seed_2026):
|
||||
db = seed_2026
|
||||
f.fahrer(db, "Bernius", "Meik", 1985)
|
||||
f.fahrer(db, "Bernius", "Jörg", 1960)
|
||||
assert finde_exakt(db, "Bernius", "Meik", 1985).vorname == "Meik"
|
||||
assert finde_exakt(db, "Bernius", "Jörg", 1960).vorname == "Jörg"
|
||||
|
||||
|
||||
def test_merge_haengt_starts_um_und_deaktiviert_quelle(seed_2026):
|
||||
db = seed_2026
|
||||
t = f.turnier(db)
|
||||
k = f.klasse(db, "S 6")
|
||||
ziel = f.fahrer(db, "Krack", "Robin", 1995)
|
||||
dublette = f.fahrer(db, "Krack", "Robin", 1995)
|
||||
f.start(db, t, dublette, k, startnummer=5)
|
||||
anzahl = fahrer_zusammenfuehren(db, ziel.id, dublette.id)
|
||||
db.flush()
|
||||
assert anzahl == 1
|
||||
assert dublette.aktiv is False
|
||||
assert ziel.starts[0].startnummer == 5
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Meisterschaftsberechnung (AP 3) — mit PLATZHALTER-Punktetabelle (O-1 offen)."""
|
||||
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from oamc.meisterschaft.berechnung import (
|
||||
PLATZHALTER_PUNKTETABELLE,
|
||||
berechne_stand,
|
||||
gewertete_laeufe,
|
||||
punkte_fuer_platz,
|
||||
)
|
||||
from oamc.models import Meisterschaft
|
||||
from oamc.models.enums import VeranstaltungStatus
|
||||
from tests import factories as f
|
||||
|
||||
|
||||
def test_punkte_fuer_platz_platzhalter():
|
||||
assert punkte_fuer_platz(1) == Decimal(25)
|
||||
assert punkte_fuer_platz(10) == Decimal(6)
|
||||
assert punkte_fuer_platz(99) == Decimal(1)
|
||||
assert punkte_fuer_platz(None) == Decimal(0)
|
||||
|
||||
|
||||
def test_streichresultate_tabelle():
|
||||
tab = {"12": 8, "9": 7, "4": 2}
|
||||
assert gewertete_laeufe(12, tab) == 8
|
||||
assert gewertete_laeufe(10, tab) == 7 # naechstkleinerer Schluessel (9)
|
||||
assert gewertete_laeufe(3, tab) == 3 # unterhalb der Tabelle: kein Streichresultat
|
||||
|
||||
|
||||
def _lauf(db, k, ergebnisse, *, datum, zaehlt=True):
|
||||
"""ergebnisse = [(fahrer, platz), ...]"""
|
||||
v = f.veranstaltung(db, datum=datum, status=VeranstaltungStatus.FINAL)
|
||||
t = f.turnier(db, v, zaehlt_meisterschaft=zaehlt)
|
||||
for i, (fahrer, platz) in enumerate(ergebnisse, start=1):
|
||||
s = f.start(db, t, fahrer, k, startnummer=i)
|
||||
e = f.ergebnis(db, s, zeit=Decimal("100"), fehler=0)
|
||||
e.platzierung = platz
|
||||
db.flush()
|
||||
return t
|
||||
|
||||
|
||||
def test_stand_summiert_punkte_und_zaehlt_siege(seed_2026):
|
||||
db = seed_2026
|
||||
k = f.klasse(db, "J 3")
|
||||
a = f.fahrer(db, "Sieger", "Sina", 2014)
|
||||
b = f.fahrer(db, "Zweite", "Zoe", 2014)
|
||||
_lauf(db, k, [(a, 1), (b, 2)], datum=date(2026, 4, 1))
|
||||
_lauf(db, k, [(a, 1), (b, 2)], datum=date(2026, 5, 1))
|
||||
m = db.scalars(
|
||||
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
|
||||
).one()
|
||||
bericht = berechne_stand(db, m.id)
|
||||
zeilen = {z.fahrer_id: z for z in bericht.zeilen}
|
||||
assert zeilen[a.id].platz == 1
|
||||
assert zeilen[a.id].punkte_gesamt == Decimal(50) # 2x Platz 1 = 2x25
|
||||
assert zeilen[a.id].anzahl_siege == 2
|
||||
assert bericht.hinweise # O-1 Platzhalter-Warnung
|
||||
|
||||
|
||||
def test_tie_break_majoritaet_der_siege(seed_2026):
|
||||
db = seed_2026
|
||||
k = f.klasse(db, "J 3")
|
||||
a = f.fahrer(db, "Aaa", "A", 2014)
|
||||
b = f.fahrer(db, "Bbb", "B", 2014)
|
||||
# beide sammeln gleich viele Punkte, aber A hat mehr Siege
|
||||
_lauf(db, k, [(a, 1), (b, 6)], datum=date(2026, 4, 1)) # A:25 B:10
|
||||
_lauf(db, k, [(b, 1), (a, 6)], datum=date(2026, 5, 1)) # B:25 A:10
|
||||
_lauf(db, k, [(a, 1), (b, 1)], datum=date(2026, 6, 1)) # beide 25, beide Sieg
|
||||
_lauf(db, k, [(a, 1), (b, 4)], datum=date(2026, 6, 15)) # A:25 (Sieg) B:13
|
||||
m = db.scalars(
|
||||
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
|
||||
).one()
|
||||
bericht = berechne_stand(db, m.id)
|
||||
assert bericht.zeilen[0].fahrer_id == a.id
|
||||
assert bericht.zeilen[0].anzahl_siege > bericht.zeilen[1].anzahl_siege
|
||||
|
||||
|
||||
def test_laeufe_ohne_meisterschaft_zaehlen_nicht(seed_2026):
|
||||
db = seed_2026
|
||||
k = f.klasse(db, "J 3")
|
||||
a = f.fahrer(db, "Solo", "S", 2014)
|
||||
_lauf(db, k, [(a, 1)], datum=date(2026, 4, 1), zaehlt=False)
|
||||
m = db.scalars(
|
||||
select(Meisterschaft).where(Meisterschaft.bezeichnung == "HTH Jugend J 3 2026")
|
||||
).one()
|
||||
bericht = berechne_stand(db, m.id)
|
||||
assert bericht.laeufe_gesamt == 0
|
||||
assert bericht.zeilen == []
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Constraint-Tests (Plan AP 1): Mehrfachstart erlaubt, Startnummer eindeutig je
|
||||
Turnier, hoechstens ein aktives Turnier."""
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from tests import factories as f
|
||||
|
||||
|
||||
def test_startnummer_eindeutig_je_turnier(seed_2026):
|
||||
db = seed_2026
|
||||
t = f.turnier(db)
|
||||
k = f.klasse(db, "S 6")
|
||||
a, b = f.fahrer(db, "A", "A"), f.fahrer(db, "B", "B")
|
||||
f.start(db, t, a, k, startnummer=10)
|
||||
f.start(db, t, b, k, startnummer=10, flush=False)
|
||||
with pytest.raises(IntegrityError):
|
||||
db.flush()
|
||||
|
||||
|
||||
def test_mehrfachstart_desselben_fahrers_erlaubt(seed_2026):
|
||||
db = seed_2026
|
||||
t = f.turnier(db)
|
||||
fahrer = f.fahrer(db, "Schmunk", "Christian", 1990)
|
||||
f.start(db, t, fahrer, f.klasse(db, "S 9"), startnummer=1)
|
||||
f.start(db, t, fahrer, f.klasse(db, "S 3"), startnummer=2)
|
||||
db.flush() # kein Fehler
|
||||
assert len(fahrer.starts) == 2
|
||||
|
||||
|
||||
def test_hoechstens_ein_aktives_turnier(seed_2026):
|
||||
db = seed_2026
|
||||
v = f.veranstaltung(db)
|
||||
f.turnier(db, v, name="A", aktiv=True)
|
||||
f.turnier(db, v, name="B", aktiv=True, flush=False)
|
||||
with pytest.raises(IntegrityError):
|
||||
db.flush()
|
||||
|
||||
|
||||
def test_doppelturnier_zwei_turniere_pro_veranstaltung(seed_2026):
|
||||
db = seed_2026
|
||||
v = f.veranstaltung(db, titel="Doppelturnier MSC Hainstadt / OAMC Reinheim")
|
||||
v.ist_doppelturnier = True
|
||||
f.turnier(db, v, name="Reinheim")
|
||||
f.turnier(db, v, name="Hainstadt")
|
||||
db.flush()
|
||||
assert {t.name for t in v.turniere} == {"Reinheim", "Hainstadt"}
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Stammdaten-Seed (AP 1)."""
|
||||
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from oamc.models import Klasse, Meisterschaft, Saison, Verein
|
||||
from oamc.models.enums import Kategorie
|
||||
from oamc.seeds import seed_saison_2026
|
||||
|
||||
|
||||
def test_seed_ist_idempotent(db):
|
||||
z1 = seed_saison_2026(db)
|
||||
db.flush()
|
||||
z2 = seed_saison_2026(db)
|
||||
db.flush()
|
||||
assert z1["klasse"] == 18 # J1-J5 + S1-S9 + A1-A4
|
||||
assert all(v == 0 for v in z2.values())
|
||||
|
||||
|
||||
def test_seed_jugendklassen_haben_jahrgaenge(seed_2026):
|
||||
db = seed_2026
|
||||
j1 = db.scalars(select(Klasse).where(Klasse.code == "J 1", Klasse.saison_jahr == 2026)).one()
|
||||
assert j1.kategorie == Kategorie.JUGEND
|
||||
assert j1.jahrgaenge == [2019, 2018, 2017]
|
||||
assert j1.mindestalter_am_veranstaltungstag == 7
|
||||
|
||||
|
||||
def test_seed_saison_streichresultate(seed_2026):
|
||||
s = seed_2026.get(Saison, 2026)
|
||||
assert s.streichresultate_tabelle["12"] == 8
|
||||
assert s.streichresultate_tabelle["4"] == 2
|
||||
|
||||
|
||||
def test_seed_meisterschaften(seed_2026):
|
||||
db = seed_2026
|
||||
ms = db.scalars(select(Meisterschaft).where(Meisterschaft.saison_jahr == 2026)).all()
|
||||
bezeichnungen = {m.bezeichnung for m in ms}
|
||||
assert "HTH Erwachsene 2026" in bezeichnungen
|
||||
assert "HTH Jugend J 3 2026" in bezeichnungen
|
||||
# O-1: Platzhalter-Punktetabelle ist markiert
|
||||
erw = next(m for m in ms if m.bezeichnung == "HTH Erwachsene 2026")
|
||||
assert "_hinweis" in erw.regelwerk
|
||||
|
||||
|
||||
def test_seed_vereins_kurznamen(seed_2026):
|
||||
db = seed_2026
|
||||
hainstadt = db.scalars(select(Verein).where(Verein.kurzname == "Hainstadt")).first()
|
||||
assert hainstadt is not None
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Einheitstests der Wertungsformeln (Plan §3.3)."""
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
from oamc.models.enums import Kategorie, StartStatus
|
||||
from oamc.wertung.engine import WertungsEingabe, platziere, regel_fuer
|
||||
|
||||
|
||||
def _e(sid, kat, zeit=None, fehler=None, status=StartStatus.GESTARTET, **kw):
|
||||
return WertungsEingabe(
|
||||
start_id=sid,
|
||||
kategorie=kat,
|
||||
status=status,
|
||||
summe_zeit=Decimal(str(zeit)) if zeit is not None else None,
|
||||
summe_fehler=fehler,
|
||||
**kw,
|
||||
)
|
||||
|
||||
|
||||
def test_erwachsene_gesamt_ist_zeit_plus_fehler():
|
||||
regel = regel_fuer(Kategorie.STAMM)
|
||||
assert regel.gesamt(Decimal("105.43"), 0) == Decimal("105.43")
|
||||
assert regel.gesamt(Decimal("103.26"), 10) == Decimal("113.26")
|
||||
assert regel.gesamt(Decimal("65.71"), 1210) == Decimal("1275.71")
|
||||
|
||||
|
||||
def test_jugend_gesamt_ist_nur_fehler():
|
||||
regel = regel_fuer(Kategorie.JUGEND)
|
||||
assert regel.gesamt(Decimal("136.88"), 0) == Decimal("0.00")
|
||||
assert regel.gesamt(Decimal("148.42"), 60) == Decimal("60.00")
|
||||
|
||||
|
||||
def test_jugend_gleichstand_wird_ueber_zeit_aufgeloest():
|
||||
res = platziere(
|
||||
[
|
||||
_e(1, Kategorie.JUGEND, 144.46, 0),
|
||||
_e(2, Kategorie.JUGEND, 136.88, 0),
|
||||
_e(3, Kategorie.JUGEND, 140.40, 0),
|
||||
]
|
||||
)
|
||||
platz = {r.start_id: r.platzierung for r in res}
|
||||
assert platz == {2: 1, 3: 2, 1: 3} # schnellste Zeit vorn
|
||||
|
||||
|
||||
def test_erwachsene_platzierung_aufsteigend_nach_gesamt():
|
||||
res = platziere(
|
||||
[
|
||||
_e(1, Kategorie.STAMM, 130.33, 0), # 130.33
|
||||
_e(2, Kategorie.STAMM, 105.43, 0), # 105.43
|
||||
_e(3, Kategorie.STAMM, 103.26, 10), # 113.26
|
||||
]
|
||||
)
|
||||
platz = {r.start_id: r.platzierung for r in res}
|
||||
assert platz == {2: 1, 3: 2, 1: 3}
|
||||
|
||||
|
||||
def test_dnf_dns_dsq_werden_nicht_klassiert():
|
||||
res = platziere(
|
||||
[
|
||||
_e(1, Kategorie.STAMM, 100.0, 0),
|
||||
_e(2, Kategorie.STAMM, 90.0, 0, status=StartStatus.DNF),
|
||||
_e(3, Kategorie.STAMM, None, None, status=StartStatus.DNS),
|
||||
]
|
||||
)
|
||||
by_id = {r.start_id: r for r in res}
|
||||
assert by_id[1].platzierung == 1
|
||||
assert by_id[2].platzierung is None and not by_id[2].gewertet
|
||||
assert by_id[3].platzierung is None
|
||||
|
||||
|
||||
def test_unvollstaendiges_ergebnis_bleibt_vorlaeufig_ohne_platz():
|
||||
res = platziere(
|
||||
[
|
||||
_e(1, Kategorie.STAMM, 100.0, 0, fehler_vollstaendig=False),
|
||||
_e(2, Kategorie.STAMM, 110.0, 0),
|
||||
]
|
||||
)
|
||||
by_id = {r.start_id: r for r in res}
|
||||
assert by_id[1].platzierung is None
|
||||
assert by_id[1].gesamt == Decimal("100.00") # vorlaeufige Gesamtanzeige
|
||||
assert "vorlaeufig" in by_id[1].hinweis
|
||||
assert by_id[2].platzierung == 1
|
||||
|
||||
|
||||
def test_echter_gleichstand_teilt_platz_und_laesst_luecke():
|
||||
# identische Zeit UND Fehler -> selber Platz, danach Luecke
|
||||
res = platziere(
|
||||
[
|
||||
_e(1, Kategorie.STAMM, 100.00, 0),
|
||||
_e(2, Kategorie.STAMM, 100.00, 0),
|
||||
_e(3, Kategorie.STAMM, 105.00, 0),
|
||||
]
|
||||
)
|
||||
platz = sorted(r.platzierung for r in res)
|
||||
assert platz == [1, 1, 3]
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Abnahmetest der Wertungs-Engine gegen die reale Ergebnisliste
|
||||
``2026-05-31-MT-REIN-HAIN.PDF`` (Plan §3.3, AP 2).
|
||||
|
||||
Die vollstaendigen 13 Wertungsbloecke liegen als PDF vor; hier abgebildet sind
|
||||
die im Plan zitierten Belegzeilen. Wo die zitierten Zeilen einen zusammenhaengenden
|
||||
Platzbereich bilden, wird die exakte publizierte Platzierung reproduziert; wo
|
||||
Zeilen aus der Mitte des Feldes fehlen, wird die relative Ordnung geprueft.
|
||||
|
||||
TODO(AP 2): sobald der volle PDF-Text vorliegt, alle 13 Bloecke vollstaendig
|
||||
hinterlegen und 1:1 gegen die publizierten Plaetze pruefen.
|
||||
"""
|
||||
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from oamc.models.enums import Kategorie, StartStatus
|
||||
from oamc.wertung.engine import WertungsEingabe, platziere
|
||||
|
||||
|
||||
def _e(sid, kat, zeit, fehler):
|
||||
return WertungsEingabe(
|
||||
start_id=sid,
|
||||
kategorie=kat,
|
||||
status=StartStatus.GESTARTET,
|
||||
summe_zeit=Decimal(str(zeit)),
|
||||
summe_fehler=fehler,
|
||||
)
|
||||
|
||||
|
||||
# --- Block "S 5, S 9, S 3, S 6, S 7" (Erwachsene) -----------------------
|
||||
S_BLOCK = [
|
||||
# (name, klasse, zeit, fehler, publizierter_platz)
|
||||
("Bernius Meik", "S 6", "105.43", 0, 1),
|
||||
("Schmunk Christian", "S 9", "103.26", 10, 2),
|
||||
("Krack Robin", "S 5", "130.33", 0, 4),
|
||||
("Steinbrech Nicolai", "S 5", "65.71", 1210, 11),
|
||||
]
|
||||
|
||||
|
||||
def test_s_block_gesamt_exakt():
|
||||
erwartet = {
|
||||
"Bernius Meik": Decimal("105.43"),
|
||||
"Schmunk Christian": Decimal("113.26"),
|
||||
"Krack Robin": Decimal("130.33"),
|
||||
"Steinbrech Nicolai": Decimal("1275.71"),
|
||||
}
|
||||
eingaben = [_e(i, Kategorie.STAMM, z, f) for i, (_n, _k, z, f, _p) in enumerate(S_BLOCK)]
|
||||
res = {i: r for i, r in enumerate(platziere(eingaben))}
|
||||
for i, (name, _k, _z, _f, _p) in enumerate(S_BLOCK):
|
||||
assert res[i].gesamt == erwartet[name], name
|
||||
|
||||
|
||||
def test_s_block_relative_ordnung_entspricht_publikation():
|
||||
eingaben = [_e(i, Kategorie.STAMM, z, f) for i, (_n, _k, z, f, _p) in enumerate(S_BLOCK)]
|
||||
res = {i: r for i, r in enumerate(platziere(eingaben))}
|
||||
# publizierte Plaetze aufsteigend -> unsere internen Plaetze muessen dieselbe
|
||||
# Reihenfolge haben
|
||||
intern = [res[i].platzierung for i in range(len(S_BLOCK))]
|
||||
publiziert = [p for *_r, p in S_BLOCK]
|
||||
assert [x for _, x in sorted(zip(publiziert, intern))] == sorted(intern)
|
||||
assert intern == sorted(intern) # bereits in publizierter Reihenfolge gelistet
|
||||
|
||||
|
||||
# --- Block "J 3" (Jugend) --------------------------------------------
|
||||
# Belegzeilen aus dem Plan. Plaetze 5+6 fehlen im Zitat -> mit klar markierten
|
||||
# Fuellzeilen (Fehler zwischen 10 und 60) besetzt, damit die publizierten Plaetze
|
||||
# 1,2,3,4,7,8 exakt herauskommen.
|
||||
J3_BELEGE = [
|
||||
("Freund Philipp", "136.88", 0, 1),
|
||||
("D'Angelo Leandro", "140.40", 0, 2),
|
||||
("Heil Luca", "144.46", 0, 3),
|
||||
("Mattes Flynn", "160.05", 10, 4),
|
||||
("Ratter Mitja", "148.42", 60, 7),
|
||||
("Kemesies Finlay", "165.39", 60, 8),
|
||||
]
|
||||
J3_FUELLER = [ # SYNTHETIC_FILLER — nicht aus der Quelle, nur Platz 5/6
|
||||
("SYNTHETIC_FILLER_P5", "150.00", 20, 5),
|
||||
("SYNTHETIC_FILLER_P6", "150.00", 50, 6),
|
||||
]
|
||||
|
||||
|
||||
def test_j3_block_reproduziert_publizierte_plaetze():
|
||||
alle = J3_BELEGE + J3_FUELLER
|
||||
eingaben = [_e(i, Kategorie.JUGEND, z, f) for i, (_n, z, f, _p) in enumerate(alle)]
|
||||
res = {i: r for i, r in enumerate(platziere(eingaben))}
|
||||
for i, (name, _z, _f, publiziert) in enumerate(alle):
|
||||
assert res[i].platzierung == publiziert, name
|
||||
# Gesamt der Jugend = Fehler
|
||||
assert res[i].gesamt == Decimal(alle[i][2]).quantize(Decimal("0.01")), name
|
||||
|
||||
|
||||
def test_j3_drei_nuller_werden_ueber_die_zeit_sortiert():
|
||||
eingaben = [_e(i, Kategorie.JUGEND, z, f) for i, (_n, z, f, _p) in enumerate(J3_BELEGE[:3])]
|
||||
res = platziere(eingaben)
|
||||
assert [r.platzierung for r in res] == [1, 2, 3] # 136,88 < 140,40 < 144,46
|
||||
|
||||
|
||||
def test_j3_zwei_mit_60_fehlern_zeitentscheid():
|
||||
# Ratter 148,42 vor Kemesies 165,39
|
||||
eingaben = [
|
||||
_e(0, Kategorie.JUGEND, "148.42", 60),
|
||||
_e(1, Kategorie.JUGEND, "165.39", 60),
|
||||
]
|
||||
res = platziere(eingaben)
|
||||
assert res[0].platzierung == 1 and res[1].platzierung == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name,zeit,fehler,gesamt", [
|
||||
("Freund Philipp", "136.88", 0, "0.00"),
|
||||
("Mattes Flynn", "160.05", 10, "10.00"),
|
||||
("Ratter Mitja", "148.42", 60, "60.00"),
|
||||
])
|
||||
def test_j3_einzelne_gesamtwerte(name, zeit, fehler, gesamt):
|
||||
res = platziere([_e(0, Kategorie.JUGEND, zeit, fehler)])
|
||||
assert res[0].gesamt == Decimal(gesamt)
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Excel-Import (AP 4/8). Spaltenzuordnung gegen die realen Vorlagen noch offen (O-12)."""
|
||||
|
||||
import io
|
||||
|
||||
from openpyxl import Workbook
|
||||
|
||||
from oamc.importer.xlsx import lese_ergebnisliste, lese_nennliste
|
||||
from oamc.seeds import seed_saison_2026
|
||||
from tests import factories as f
|
||||
|
||||
HEAD = {"X-API-Key": "test-key"}
|
||||
|
||||
|
||||
def _xlsx(rows: list[list]) -> bytes:
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
for r in rows:
|
||||
ws.append(r)
|
||||
buf = io.BytesIO()
|
||||
wb.save(buf)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def test_lese_nennliste_mappt_spalten():
|
||||
daten = _xlsx([
|
||||
["Start-Nr.", "Voller Name", "Regionalclub", "Klasse", "Geburtsjahr"],
|
||||
[32, "Gál-Szász Péter", "OAMC Reinheim", "J 2", 2015],
|
||||
[33, "Bernius Meik", "Hainstadt", "S 6", 1985],
|
||||
])
|
||||
bulk = lese_nennliste(daten, quelle="test", idempotenz_schluessel="k1")
|
||||
assert [s.startnummer for s in bulk.starts] == [32, 33]
|
||||
assert bulk.starts[0].fahrer.nachname == "Gál-Szász"
|
||||
assert bulk.starts[0].fahrer.vorname == "Péter"
|
||||
assert bulk.starts[0].klasse == "J 2"
|
||||
|
||||
|
||||
def test_lese_ergebnisliste_liest_zeit_und_fehler():
|
||||
daten = _xlsx([
|
||||
["Start-Nr.", "Voller Name", "Klasse", "Summe Zeit", "Summe Fehler", "Gesamt", "Platzierung"],
|
||||
[1, "Bernius Meik", "S 6", "105,43", 0, "105,43", 1],
|
||||
])
|
||||
bulk = lese_ergebnisliste(daten, quelle="test", idempotenz_schluessel="k2")
|
||||
e = bulk.ergebnisse[0]
|
||||
assert str(e.summe_zeit) == "105.43"
|
||||
assert e.summe_fehler == 0
|
||||
assert str(e.gesamt_erwartet) == "105.43"
|
||||
|
||||
|
||||
def test_import_xlsx_endpoint_dry_run(client):
|
||||
db = client.db
|
||||
seed_saison_2026(db)
|
||||
t = f.turnier(db)
|
||||
db.flush()
|
||||
daten = _xlsx([
|
||||
["Start-Nr.", "Voller Name", "Klasse", "Geburtsjahr"],
|
||||
[5, "Freund Philipp", "J 3", 2014],
|
||||
])
|
||||
r = client.post(
|
||||
f"/api/v1/turniere/{t.id}/import/xlsx?dry_run=true",
|
||||
headers=HEAD,
|
||||
data={"art": "nennliste", "idempotenz_schluessel": "xlsx-1"},
|
||||
files={"datei": ("nenn.xlsx", daten,
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["dry_run"] is True
|
||||
assert r.json()["angelegt"] == 1
|
||||
Reference in New Issue
Block a user