90 lines
5.2 KiB
Python
90 lines
5.2 KiB
Python
"""Plan der KI und Planungsmodus (→ große Aufgaben)."""
|
|
|
|
from sqlalchemy import select
|
|
from test_branches import README, FakePRs, ask, branch, fake_prs, hx, setup, work # noqa: F401
|
|
|
|
from app.changes.models import ChatBranch
|
|
from app.chat.models import ChatSession, Message
|
|
from app.chat.plan import EXECUTE_MESSAGE
|
|
from app.tasks import states
|
|
from app.tasks.models import TaskRun
|
|
|
|
PLAN = [{"text": "README lesen", "status": "done"}, {"text": "Abschnitt Installation ergänzen", "status": "open"}]
|
|
|
|
|
|
def offered_names(call) -> set[str]:
|
|
return {t["name"] for t in call["tools"] or []}
|
|
|
|
|
|
def test_planungsmodus_nur_lesen_und_planen(setup, db, settings, fake_provider):
|
|
chat = f"/chats/{setup['chat_id']}"
|
|
page = setup["ben"].get(chat).text
|
|
assert "Erst planen" in page and "Planungsmodus:" not in page
|
|
response = hx(setup["ben"], f"{chat}/plan-mode", {"on": "true"})
|
|
assert response.status_code == 200 and "Planungsmodus: die KI ändert nichts" in response.text
|
|
# Der Schalter am Eingabefeld wird mit ersetzt
|
|
assert 'id="plan-toggle" class="cluster" hx-swap-oob="true"' in response.text and "Planungsmodus an" in response.text
|
|
assert 'aria-pressed="true"' in setup["ben"].get(chat).text
|
|
assert "Plan umsetzen</button>" not in response.text # ohne Plan nichts umzusetzen
|
|
fake_provider.script = [[("git_read_file", {"path": "README.md"})], [("update_plan", {"steps": PLAN})],
|
|
"Plan steht."]
|
|
ask(setup["ben"], setup["chat_id"])
|
|
work(db, settings)
|
|
first = fake_provider.calls[0]
|
|
assert "PLANUNGSMODUS" in first["instructions"]
|
|
names = offered_names(first)
|
|
assert "update_plan" in names and "git_read_file" in names and "commit_change" not in names
|
|
with db.session() as session:
|
|
assert session.get(ChatSession, setup["chat_id"]).plan == PLAN
|
|
panel = setup["ben"].get(f"{chat}/plan").text
|
|
assert "1 von 2 erledigt" in panel and "Abschnitt Installation ergänzen" in panel and "Plan umsetzen</button>" in panel
|
|
# Selbst wenn das Modell commit_change aufruft: im Planungsmodus nicht angeboten, also abgelehnt
|
|
fake_provider.script = [[("commit_change", {"message": "Heimlich", "changes": [README]})], "Ok."]
|
|
ask(setup["ben"], setup["chat_id"])
|
|
work(db, settings)
|
|
with db.session() as session:
|
|
assert session.scalar(select(ChatBranch.id)) is None
|
|
denied = [i["output"] for i in fake_provider.calls[-1]["input"] if i.get("type") == "function_call_output"][-1]
|
|
assert "ist in dieser Antwort nicht verfügbar (Planungsmodus" in denied
|
|
# Plan umsetzen: Modus aus, Umsetzung startet als Nachricht, schreibende Tools wieder da
|
|
assert setup["dora"].post(f"{chat}/plan/execute").status_code == 404
|
|
response = setup["ben"].post(f"{chat}/plan/execute")
|
|
assert response.status_code == 303
|
|
assert "Plan überarbeiten" in setup["ben"].get(chat).text # Modus aus, Plan vorhanden
|
|
fake_provider.script = [[("commit_change", {"message": "Installation", "changes": [README]})],
|
|
[("update_plan", {"steps": [{**s, "status": "done"} for s in PLAN]})], "Fertig."]
|
|
work(db, settings)
|
|
call = fake_provider.calls[-3]
|
|
assert call["input"][-1]["content"] == EXECUTE_MESSAGE and "commit_change" in offered_names(call)
|
|
assert "PLANUNGSMODUS" not in call["instructions"]
|
|
assert "[x] README lesen" in call["instructions"] and "[ ] Abschnitt Installation" in call["instructions"]
|
|
with db.session() as session:
|
|
chat_row = session.get(ChatSession, setup["chat_id"])
|
|
assert not chat_row.plan_mode and all(s["status"] == "done" for s in chat_row.plan)
|
|
assert session.scalar(select(ChatBranch.id)) is not None
|
|
|
|
|
|
def test_plan_in_pr_beschreibung_und_ohne_modus(setup, db, settings, fake_provider, fake_prs):
|
|
fake_provider.script = [[("update_plan", {"steps": PLAN})],
|
|
[("commit_change", {"message": "README ergänzen", "changes": [README]})], "Erledigt."]
|
|
ask(setup["ben"], setup["chat_id"])
|
|
work(db, settings)
|
|
assert "lege zuerst mit update_plan einen Plan an" in fake_provider.calls[0]["instructions"]
|
|
hx(setup["ben"], f"/branches/{branch(db).id}/pull-request", {"title": "README ergänzen"})
|
|
description = fake_prs.created[0]["description"]
|
|
assert "Plan:\n- [x] README lesen\n- [ ] Abschnitt Installation ergänzen" in description
|
|
|
|
|
|
def test_plan_ohne_plan_nicht_umsetzbar_und_validierung(setup, db, settings, fake_provider):
|
|
chat = f"/chats/{setup['chat_id']}"
|
|
assert setup["ben"].post(f"{chat}/plan/execute").status_code == 400
|
|
assert setup["dora"].post(f"{chat}/plan-mode", {"on": "true"}).status_code == 404
|
|
fake_provider.script = [[("update_plan", {"steps": [{"text": "x", "status": "fertig"}]})], "Ok."]
|
|
ask(setup["ben"], setup["chat_id"])
|
|
work(db, settings)
|
|
output = [i["output"] for i in fake_provider.calls[-1]["input"] if i.get("type") == "function_call_output"][-1]
|
|
assert "Ungültige" in output or "Fehler" in output
|
|
with db.session() as session:
|
|
assert session.get(ChatSession, setup["chat_id"]).plan == []
|
|
assert session.scalar(select(TaskRun.status).order_by(TaskRun.created.desc())) == states.COMPLETED
|
|
assert session.scalar(select(Message.id).where(Message.role == "assistant")) is not None
|