137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
import agents
|
|
import features
|
|
import projekte
|
|
import pruefung
|
|
|
|
ANTWORT = """```json
|
|
{"probleme": [{"feld": "regeln", "funktion": "Etwas tun", "original": "Zwei Aussagen, und noch eine",
|
|
"problem": "Zwei Aussagen je Regel", "vorschlag": "Zwei Aussagen"}]}
|
|
```"""
|
|
|
|
|
|
@pytest.fixture
|
|
def projekt(datenordner):
|
|
projekte.anlegen("Test")
|
|
features.anlegen("Test", "<!-- titel -->\nEins\n")
|
|
features.anlegen("Test", "<!-- titel -->\nZwei\n")
|
|
return "Test"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def kein_lauf():
|
|
pruefung._laeufe.clear()
|
|
yield
|
|
pruefung._laeufe.clear()
|
|
|
|
|
|
async def warte_bis_fertig(projekt):
|
|
for _ in range(50):
|
|
if not pruefung.status(projekt)["laeuft"]:
|
|
return
|
|
await asyncio.sleep(0.01)
|
|
raise AssertionError("Prüfung wurde nicht fertig")
|
|
|
|
|
|
async def test_lauf_sammelt_probleme(projekt, monkeypatch):
|
|
monkeypatch.setattr(agents, "pruefe_konfiguration", lambda: None)
|
|
monkeypatch.setattr(agents, "frage", lambda system, inhalt: _antwort())
|
|
await pruefung.starte(projekt)
|
|
await warte_bis_fertig(projekt)
|
|
|
|
s = pruefung.status(projekt)
|
|
assert s["gesamt"] == 2
|
|
assert s["fertig"] == 2
|
|
assert s["meldung"] == ""
|
|
assert len(s["probleme"]) == 2
|
|
assert s["probleme"][0]["feature"] in ("eins", "zwei")
|
|
assert s["probleme"][0]["vorschlag"] == "Zwei Aussagen"
|
|
|
|
|
|
async def _antwort():
|
|
return ANTWORT
|
|
|
|
|
|
async def test_ohne_konfiguration_kein_lauf(projekt, monkeypatch):
|
|
monkeypatch.setattr(agents.config, "MINIMAX_API_KEY", "")
|
|
with pytest.raises(ValueError):
|
|
await pruefung.starte(projekt)
|
|
assert pruefung.status(projekt)["laeuft"] is False
|
|
|
|
|
|
async def test_providerfehler_landet_in_meldung(projekt, monkeypatch):
|
|
monkeypatch.setattr(agents, "pruefe_konfiguration", lambda: None)
|
|
|
|
async def kaputt(system, inhalt):
|
|
raise RuntimeError("Provider antwortet mit HTTP 500")
|
|
|
|
monkeypatch.setattr(agents, "frage", kaputt)
|
|
await pruefung.starte(projekt)
|
|
await warte_bis_fertig(projekt)
|
|
assert pruefung.status(projekt)["meldung"] == "Provider antwortet mit HTTP 500"
|
|
|
|
|
|
async def test_verworfener_lauf_stoppt(projekt, monkeypatch):
|
|
monkeypatch.setattr(agents, "pruefe_konfiguration", lambda: None)
|
|
monkeypatch.setattr(agents.config, "PRUEFUNG_PARALLEL", 1)
|
|
aufrufe = []
|
|
|
|
async def zaehle(system, inhalt):
|
|
aufrufe.append(inhalt)
|
|
pruefung.beende("Test") # Panel wird nach der ersten Karte geschlossen
|
|
return '{"probleme": []}'
|
|
|
|
monkeypatch.setattr(agents, "frage", zaehle)
|
|
await pruefung.starte(projekt)
|
|
for _ in range(50):
|
|
await asyncio.sleep(0.01)
|
|
assert len(aufrufe) == 1 # zweite Karte wird nicht mehr angefragt
|
|
|
|
|
|
async def test_parallel_begrenzt(projekt, monkeypatch):
|
|
monkeypatch.setattr(agents, "pruefe_konfiguration", lambda: None)
|
|
monkeypatch.setattr(agents.config, "PRUEFUNG_PARALLEL", 1)
|
|
gleichzeitig = 0
|
|
hoechststand = 0
|
|
|
|
async def langsam(system, inhalt):
|
|
nonlocal gleichzeitig, hoechststand
|
|
gleichzeitig += 1
|
|
hoechststand = max(hoechststand, gleichzeitig)
|
|
await asyncio.sleep(0.02)
|
|
gleichzeitig -= 1
|
|
return '{"probleme": []}'
|
|
|
|
monkeypatch.setattr(agents, "frage", langsam)
|
|
await pruefung.starte(projekt)
|
|
await warte_bis_fertig(projekt)
|
|
assert hoechststand == 1
|
|
assert pruefung.status(projekt)["fertig"] == 2
|
|
|
|
|
|
async def test_status_ohne_lauf():
|
|
assert pruefung.status("Fremd") == {
|
|
"laeuft": False, "fertig": 0, "gesamt": 0, "probleme": [], "meldung": ""
|
|
}
|
|
|
|
|
|
def test_antwort_lesen_ohne_zaun():
|
|
probleme = pruefung._antwort_lesen('{"probleme": [{"original": "x", "vorschlag": "y"}]}')
|
|
assert probleme == [{"original": "x", "vorschlag": "y", "funktion": ""}]
|
|
|
|
|
|
def test_listenstrich_wird_entfernt():
|
|
probleme = pruefung._antwort_lesen(
|
|
'{"probleme": [{"original": "- alt", "vorschlag": "- neu", "funktion": " F "}]}')
|
|
assert probleme[0]["original"] == "alt"
|
|
assert probleme[0]["vorschlag"] == "neu"
|
|
assert probleme[0]["funktion"] == "F"
|
|
|
|
|
|
def test_antwort_ohne_json():
|
|
with pytest.raises(ValueError):
|
|
pruefung._antwort_lesen("keine Ahnung")
|