419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""Agenten-Übergabe: der Auftrag beschreibt den Diff, der Start braucht ein Verzeichnis."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
import auftrag
|
|
import features
|
|
import planung
|
|
import projekte
|
|
import verzeichnis
|
|
|
|
T = "-" * 40
|
|
|
|
|
|
@pytest.fixture
|
|
def projekt(datenordner):
|
|
projekte.anlegen("A")
|
|
features.anlegen("A", f"<!-- titel -->\nGalerie\n{T}\n<!-- funktion -->\nBild wählen\n")
|
|
features.anlegen("A", f"<!-- titel -->\nShop\n{T}\n<!-- funktion -->\nKarte kaufen\n")
|
|
yield "A"
|
|
auftrag._laeufe.clear()
|
|
|
|
|
|
def test_ohne_planung_kein_auftrag(projekt):
|
|
assert auftrag.bau_auftrag(projekt) == ("", [])
|
|
|
|
|
|
def test_leere_planung_kein_auftrag(projekt):
|
|
planung.starte(projekt)
|
|
assert auftrag.bau_auftrag(projekt) == ("", [])
|
|
|
|
|
|
def test_auftrag_nennt_diff_und_sollzustand(projekt):
|
|
planung.starte(projekt)
|
|
features.anlegen(projekt, f"<!-- titel -->\nMerkliste\n{T}\n<!-- funktion -->\nProdukt merken\n")
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Shop")
|
|
features.loeschen(projekt, f["id"])
|
|
|
|
text, schritte = auftrag.bau_auftrag(projekt)
|
|
texte = [s["text"] for s in schritte]
|
|
assert "Merkliste / Produkt merken: PRÜFEN" in texte
|
|
assert "Karte „Shop“ streichen" in texte
|
|
assert all(s in text for s in texte)
|
|
# Die Zuordnung steht daneben, nicht nur im Satz
|
|
assert {"text": "Merkliste / Produkt merken: PRÜFEN", "karte": "Merkliste",
|
|
"funktion": "Produkt merken"} in schritte
|
|
assert {"text": "Karte „Shop“ streichen", "karte": "Shop",
|
|
"funktion": ""} in schritte
|
|
# Sollzustand: die neue Karte steht voll drin, die gestrichene nicht
|
|
assert "<!-- funktion -->\nProdukt merken" in text
|
|
assert "Karte kaufen" not in text.split("Die betroffenen Karten")[1]
|
|
|
|
|
|
def test_start_braucht_verzeichnis(projekt):
|
|
assert auftrag.starte(projekt, "tu was", ["x"]) is False
|
|
|
|
|
|
def test_start_braucht_auftrag(projekt, tmp_path):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
assert auftrag.starte(projekt, "", []) is False
|
|
|
|
|
|
class Strom:
|
|
"""Attrappe des Claude-Prozesses: gibt vorbereitete Stream-Zeilen aus."""
|
|
|
|
def __init__(self, zeilen, rc=0, gesehen=None):
|
|
import json as _json
|
|
self._zeilen = [(_json.dumps(z) + "\n").encode() for z in zeilen]
|
|
self.returncode = rc
|
|
self._gesehen = gesehen if gesehen is not None else {}
|
|
|
|
class Stdin:
|
|
def write(s, daten): self._gesehen["eingabe"] = daten.decode()
|
|
async def drain(s): pass
|
|
def close(s): pass
|
|
self.stdin = Stdin()
|
|
|
|
class Stdout:
|
|
async def readline(s):
|
|
return self._zeilen.pop(0) if self._zeilen else b""
|
|
self.stdout = Stdout()
|
|
|
|
class Stderr:
|
|
async def read(s): return b""
|
|
self.stderr = Stderr()
|
|
|
|
async def wait(self):
|
|
return self.returncode
|
|
|
|
|
|
def text_ereignis(text):
|
|
return {"type": "assistant", "message": {"content": [{"type": "text", "text": text}]}}
|
|
|
|
|
|
def test_start_fuehrt_claude_im_verzeichnis_aus(projekt, tmp_path, monkeypatch):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
planung.starte(projekt)
|
|
gesehen = {}
|
|
|
|
async def unecht(*kommando, cwd, **k):
|
|
gesehen["kommando"] = kommando
|
|
gesehen["cwd"] = cwd
|
|
return Strom([{"type": "result", "result": "fertig"}], gesehen=gesehen)
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", unecht)
|
|
|
|
async def lauf():
|
|
assert auftrag.starte(projekt, "Setze um: X",
|
|
[{"text": "S1", "karte": "", "funktion": ""}]) is True
|
|
assert auftrag.status(projekt)["laeuft"] is True
|
|
await asyncio.gather(*auftrag._aufgaben)
|
|
|
|
asyncio.run(lauf())
|
|
assert gesehen["kommando"][0] == "claude"
|
|
assert "--output-format" in gesehen["kommando"]
|
|
assert gesehen["cwd"] == str(tmp_path)
|
|
assert gesehen["eingabe"] == "Setze um: X"
|
|
stand = auftrag.status(projekt)
|
|
assert stand["text"] == "fertig"
|
|
assert stand["laeuft"] is False
|
|
|
|
|
|
def test_schritte_werden_abgehakt_und_planung_uebernommen(projekt, tmp_path, monkeypatch):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
planung.starte(projekt)
|
|
features.anlegen(projekt, f"<!-- titel -->\nMerkliste\n{T}\n<!-- funktion -->\nProdukt merken\n")
|
|
anweisung, schritte = auftrag.bau_auftrag(projekt)
|
|
|
|
async def unecht(*a, **k):
|
|
return Strom([
|
|
{"type": "assistant", "message": {"content": [
|
|
{"type": "tool_use", "name": "Edit", "input": {"file_path": "x.py"}}]}},
|
|
text_ereignis(f'ERLEDIGT: {schritte[0]["text"]}'),
|
|
{"type": "result", "result": "alles gebaut"},
|
|
])
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", unecht)
|
|
|
|
async def lauf():
|
|
auftrag.starte(projekt, anweisung, schritte)
|
|
await asyncio.gather(*auftrag._aufgaben)
|
|
|
|
asyncio.run(lauf())
|
|
stand = auftrag.status(projekt)
|
|
assert stand["schritte"] == [{**schritte[0], "fertig": True}]
|
|
assert stand["uebernommen"] is True
|
|
assert planung.laeuft(projekt) is False
|
|
assert "Merkliste" in [f["sektionen"]["titel"] for f in features.liste(projekt)]
|
|
|
|
|
|
def test_fehlschlag_laesst_die_planung_stehen(projekt, tmp_path, monkeypatch):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
planung.starte(projekt)
|
|
features.anlegen(projekt, f"<!-- titel -->\nMerkliste\n{T}\n<!-- funktion -->\nProdukt merken\n")
|
|
anweisung, schritte = auftrag.bau_auftrag(projekt)
|
|
|
|
async def unecht(*a, **k):
|
|
return Strom([], rc=1)
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", unecht)
|
|
|
|
async def lauf():
|
|
auftrag.starte(projekt, anweisung, schritte)
|
|
await asyncio.gather(*auftrag._aufgaben)
|
|
|
|
asyncio.run(lauf())
|
|
stand = auftrag.status(projekt)
|
|
assert stand["uebernommen"] is False
|
|
assert stand["fehler"]
|
|
assert planung.laeuft(projekt) is True
|
|
assert "Merkliste" not in [f["sektionen"]["titel"] for f in features.liste(projekt, nur_stand=True)]
|
|
|
|
|
|
def test_ungefaehre_meldung_hakt_den_naechsten_schritt_ab():
|
|
lauf = {"schritte": [{"text": "Galerie / Bild wählen: NEU", "fertig": False},
|
|
{"text": "Karte „Shop“ streichen", "fertig": False}]}
|
|
auftrag._hake_ab("Test", lauf, "ERLEDIGT: Galerie / Bild wählen (NEU)")
|
|
assert lauf["schritte"][0]["fertig"] is True
|
|
auftrag._hake_ab("Test", lauf, "ERLEDIGT: völlig anderes Zeug")
|
|
assert lauf["schritte"][1]["fertig"] is False
|
|
|
|
|
|
def test_fehlender_claude_wird_gemeldet(projekt, tmp_path, monkeypatch):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
|
|
async def kaputt(*a, **k):
|
|
raise FileNotFoundError("claude")
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", kaputt)
|
|
|
|
async def lauf():
|
|
auftrag.starte(projekt, "X", [{"text": "S", "karte": "", "funktion": ""}])
|
|
await asyncio.gather(*auftrag._aufgaben)
|
|
|
|
asyncio.run(lauf())
|
|
assert "claude-CLI nicht gefunden" in auftrag.status(projekt)["fehler"]
|
|
assert auftrag.status(projekt)["laeuft"] is False
|
|
|
|
|
|
def test_zweiter_start_wird_abgelehnt(projekt, tmp_path):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
auftrag._laeufe[projekt] = {"laeuft": True, "fehler": "", "text": "",
|
|
"schritte": [], "aktivitaet": "", "uebernommen": False}
|
|
with pytest.raises(ValueError):
|
|
auftrag.starte(projekt, "X", [{"text": "S", "karte": "", "funktion": ""}])
|
|
|
|
|
|
# ------------------------------------------------------------- Verzeichnis
|
|
|
|
def test_verzeichnis_wird_gemerkt(projekt, tmp_path):
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
assert verzeichnis.lies(projekt) == str(tmp_path)
|
|
verzeichnis.loesche(projekt)
|
|
assert verzeichnis.lies(projekt) == ""
|
|
|
|
|
|
def test_verzeichnis_lehnt_wurzel_und_relatives_ab(projekt):
|
|
with pytest.raises(ValueError):
|
|
verzeichnis.pruefe("/")
|
|
with pytest.raises(ValueError):
|
|
verzeichnis.pruefe("relativ/pfad")
|
|
with pytest.raises(FileNotFoundError):
|
|
verzeichnis.pruefe("/gibt/es/nicht")
|
|
|
|
|
|
def test_verzeichnis_liegt_im_stand_nicht_in_der_kopie(projekt, tmp_path):
|
|
planung.starte(projekt)
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
planung.brich_ab(projekt)
|
|
assert verzeichnis.lies(projekt) == str(tmp_path)
|
|
|
|
|
|
# ------------------------------------------------------------- Durchblättern
|
|
|
|
def test_ordner_listet_nur_sichtbare_verzeichnisse(tmp_path):
|
|
(tmp_path / "b").mkdir()
|
|
(tmp_path / "A").mkdir()
|
|
(tmp_path / ".versteckt").mkdir()
|
|
(tmp_path / "datei.txt").write_text("x")
|
|
antwort = verzeichnis.ordner(str(tmp_path))
|
|
assert antwort["ordner"] == ["A", "b"]
|
|
assert antwort["pfad"] == str(tmp_path)
|
|
assert antwort["eltern"] == str(tmp_path.parent)
|
|
|
|
|
|
def test_ordner_ohne_pfad_beginnt_daheim():
|
|
from pathlib import Path
|
|
assert verzeichnis.ordner("")["pfad"] == str(Path.home())
|
|
|
|
|
|
def test_ordner_lehnt_datei_und_relatives_ab(tmp_path):
|
|
(tmp_path / "d.txt").write_text("x")
|
|
with pytest.raises(FileNotFoundError):
|
|
verzeichnis.ordner(str(tmp_path / "d.txt"))
|
|
with pytest.raises(ValueError):
|
|
verzeichnis.ordner("relativ")
|
|
|
|
|
|
def test_vision_gehoert_nie_zum_auftrag(projekt):
|
|
features.anlegen(projekt, "<!-- titel -->\nVision\n<!-- art -->\nvision\n<!-- ziel -->\nAlt\n")
|
|
planung.starte(projekt)
|
|
v = next(f for f in features.liste(projekt) if f["sektionen"]["art"] == "vision")
|
|
features.speichern(projekt, v["id"], v["roh"].replace("Alt", "Neu"))
|
|
assert auftrag.bau_auftrag(projekt) == ("", [])
|
|
|
|
features.anlegen(projekt, f"<!-- titel -->\nMerkliste\n{T}\n<!-- funktion -->\nProdukt merken\n")
|
|
text, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Merkliste / Produkt merken: PRÜFEN"]
|
|
assert "Vision" not in text
|
|
|
|
|
|
# ------------------------------------------------- Stand: bestellt wird gebaut
|
|
|
|
def _stand(projekt, karte, funktion):
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == karte)
|
|
return next(x["stand"] for x in f["sektionen"]["funktionen"] if x["text"] == funktion)
|
|
|
|
|
|
def test_abgehakte_funktion_gilt_als_gebaut(projekt, tmp_path, monkeypatch):
|
|
"""Der Agent hat es gebaut — dann darf die Karte nicht „bestellt“ behaupten."""
|
|
verzeichnis.setze(projekt, str(tmp_path))
|
|
planung.starte(projekt)
|
|
features.anlegen(projekt, f"<!-- titel -->\nMerkliste\n{T}\n<!-- funktion -->\n"
|
|
f"Produkt merken\n<!-- stand -->\noffen\n{T}\n"
|
|
f"<!-- funktion -->\nProdukt entfernen\n<!-- stand -->\noffen\n")
|
|
anweisung, schritte = auftrag.bau_auftrag(projekt)
|
|
merken = next(s for s in schritte if s["funktion"] == "Produkt merken")
|
|
|
|
async def unecht(*a, **k):
|
|
return Strom([text_ereignis(f'ERLEDIGT: {merken["text"]}'),
|
|
{"type": "result", "result": "eins gebaut"}])
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", unecht)
|
|
|
|
async def lauf():
|
|
auftrag.starte(projekt, anweisung, schritte)
|
|
await asyncio.gather(*auftrag._aufgaben)
|
|
|
|
asyncio.run(lauf())
|
|
# Nur der abgehakte Schritt wandert um; der andere bleibt bestellt
|
|
assert _stand(projekt, "Merkliste", "Produkt merken") == "gebaut"
|
|
assert _stand(projekt, "Merkliste", "Produkt entfernen") == "offen"
|
|
|
|
|
|
def test_gestrichene_funktion_wird_nicht_umgestellt(projekt, tmp_path):
|
|
"""Was weg ist, ist nicht gebaut — der Schritt trägt darum keinen Namen."""
|
|
planung.starte(projekt)
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Galerie")
|
|
features.speichern(projekt, f["id"], "<!-- titel -->\nGalerie\n")
|
|
_, schritte = auftrag.bau_auftrag(projekt)
|
|
weg = next(s for s in schritte if "Bild wählen" in s["text"])
|
|
assert weg["funktion"] == ""
|
|
|
|
|
|
def test_offen_vermerk_bleibt_bei_anderer_funktion(projekt):
|
|
roh = ("<!-- titel -->\nGalerie\n"
|
|
"<!-- funktion -->\nBild wählen\n<!-- stand -->\noffen\n"
|
|
"<!-- funktion -->\nBild schließen\n<!-- stand -->\noffen\n")
|
|
neu = auftrag._als_gebaut(roh, "Bild schließen")
|
|
assert neu.count("offen") == 1
|
|
assert "Bild wählen\n<!-- stand -->\noffen" in neu
|
|
|
|
|
|
def test_unbekannte_bloecke_bleiben_erhalten(projekt):
|
|
"""Textoperation statt Neuschreiben: der Rest der Datei bleibt Zeichen für Zeichen."""
|
|
roh = ("<!-- titel -->\nGalerie\n<!-- eigenes -->\nnicht anfassen\n"
|
|
"<!-- funktion -->\nBild wählen\n<!-- stand -->\noffen\n"
|
|
"<!-- regeln -->\n- Eine Regel\n")
|
|
neu = auftrag._als_gebaut(roh, "Bild wählen")
|
|
assert "<!-- eigenes -->\nnicht anfassen" in neu
|
|
assert "<!-- regeln -->\n- Eine Regel" in neu
|
|
assert "offen" not in neu
|
|
|
|
|
|
# ---------------------------------------- offen heißt: noch nicht bewiesen
|
|
|
|
def _auf_offen(projekt, karte, funktion):
|
|
"""Wie ein Klick auf den Stand: gebaut zurück auf offen."""
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == karte)
|
|
roh = f["roh"].replace(f"<!-- funktion -->\n{funktion}\n",
|
|
f"<!-- funktion -->\n{funktion}\n<!-- stand -->\noffen\n")
|
|
features.speichern(projekt, f["id"], roh)
|
|
|
|
|
|
def test_zurueckgesetzte_funktion_kommt_in_die_arbeitsliste(projekt):
|
|
"""Eine kaputte Umsetzung meldet man, indem man die Funktion auf offen setzt.
|
|
|
|
Die Karte ändert sich dadurch, aber sonst nichts — vorher blieb der Klick folgenlos.
|
|
"""
|
|
planung.starte(projekt)
|
|
_auf_offen(projekt, "Galerie", "Bild wählen")
|
|
text, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Galerie / Bild wählen: PRÜFEN"]
|
|
# Der Sollzustand der Karte gehört dazu, sonst weiß der Agent nicht, wogegen er prüft
|
|
assert "<!-- funktion -->\nBild wählen" in text
|
|
assert auftrag.PRUEFEN in text
|
|
|
|
|
|
def test_offene_funktion_ohne_kartenaenderung(projekt, tmp_path):
|
|
"""Offen in der Kopie und im Stand: kein Diff, trotzdem Arbeit."""
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Shop")
|
|
features.speichern(projekt, f["id"], f["roh"].replace(
|
|
"<!-- funktion -->\nKarte kaufen\n",
|
|
"<!-- funktion -->\nKarte kaufen\n<!-- stand -->\noffen\n"))
|
|
planung.starte(projekt)
|
|
assert planung.aenderungen(projekt) == {}
|
|
_, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Shop / Karte kaufen: PRÜFEN"]
|
|
|
|
|
|
def test_offen_schlaegt_geaendert(projekt):
|
|
"""Eine offene Funktion wird geprüft, nicht bloß angepasst."""
|
|
planung.starte(projekt)
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Galerie")
|
|
features.speichern(projekt, f["id"], f["roh"].replace(
|
|
"<!-- funktion -->\nBild wählen\n",
|
|
"<!-- funktion -->\nBild wählen\n<!-- stand -->\noffen\n"
|
|
"<!-- regeln -->\n- Eine neue Regel\n"))
|
|
_, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Galerie / Bild wählen: PRÜFEN"]
|
|
|
|
|
|
def test_gebaut_und_geaendert_bleibt_geaendert(projekt):
|
|
planung.starte(projekt)
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Galerie")
|
|
features.speichern(projekt, f["id"], f["roh"].replace(
|
|
"<!-- funktion -->\nBild wählen\n",
|
|
"<!-- funktion -->\nBild wählen\n<!-- regeln -->\n- Eine neue Regel\n"))
|
|
_, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Galerie / Bild wählen: GEÄNDERT"]
|
|
|
|
|
|
def test_gestrichene_karte_zieht_ihre_offenen_nicht_nach(projekt):
|
|
"""Eine gestrichene Karte darf nicht zusätzlich als „prüfen“ auftauchen."""
|
|
f = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Shop")
|
|
features.speichern(projekt, f["id"], f["roh"].replace(
|
|
"<!-- funktion -->\nKarte kaufen\n",
|
|
"<!-- funktion -->\nKarte kaufen\n<!-- stand -->\noffen\n"))
|
|
planung.starte(projekt)
|
|
g = next(x for x in features.liste(projekt) if x["sektionen"]["titel"] == "Shop")
|
|
features.loeschen(projekt, g["id"])
|
|
_, schritte = auftrag.bau_auftrag(projekt)
|
|
assert [s["text"] for s in schritte] == ["Karte „Shop“ streichen"]
|
|
|
|
|
|
def test_ohne_planung_kein_lauf(projekt):
|
|
"""Der Agent arbeitet auf der Planung; ohne sie gibt es keinen Auftrag."""
|
|
_auf_offen(projekt, "Galerie", "Bild wählen")
|
|
assert auftrag.bau_auftrag(projekt) == ("", [])
|
|
|
|
|
|
def test_auftrag_verlangt_offenlegung_des_ungeprueften(projekt):
|
|
planung.starte(projekt)
|
|
_auf_offen(projekt, "Galerie", "Bild wählen")
|
|
text, _ = auftrag.bau_auftrag(projekt)
|
|
assert "nicht prüfen konntest" in text
|
|
assert "Hake nur ab, was du geprüft hast" in text
|