120 lines
5.8 KiB
Python
120 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
def disposable_repository(logged_in):
|
|
client, app, projects, _, headers = logged_in
|
|
root = projects / "orchestration-repo"
|
|
root.mkdir()
|
|
subprocess.run(["git", "init", "-b", "main"], cwd=root, check=True, capture_output=True)
|
|
(root / "app.txt").write_text("base\n", encoding="utf-8")
|
|
subprocess.run(["git", "add", "app.txt"], cwd=root, check=True)
|
|
subprocess.run(
|
|
["git", "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "-m", "Base"],
|
|
cwd=root, check=True, capture_output=True,
|
|
)
|
|
owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
|
|
project_id = str(uuid.uuid4())
|
|
app.state.db.execute(
|
|
"INSERT INTO projects(id,name,path,owner_id,full_access,merge_confirmation_required,created_at) VALUES(?,?,?,?,?,?,?)",
|
|
(project_id, "Disposable", str(root), owner, 1, 1, "2026-01-01"),
|
|
)
|
|
return client, app, headers, root, project_id
|
|
|
|
|
|
def finish_agent(app, orchestration: dict, content: str) -> Path:
|
|
worktree = Path(orchestration["worktree_path"])
|
|
(worktree / "app.txt").write_text(content, encoding="utf-8")
|
|
app.state.db.execute("UPDATE prompt_queue SET status='done',finished_at='2026-01-02' WHERE id=?", (orchestration["queue_id"],))
|
|
app.state.db.execute("UPDATE conversations SET status='completed',worker_pid=NULL WHERE id=?", (orchestration["conversation_id"],))
|
|
return worktree
|
|
|
|
|
|
def test_complete_orchestration_refusal_correction_resume_and_merge(logged_in, monkeypatch):
|
|
client, app, headers, root, project_id = disposable_repository(logged_in)
|
|
monkeypatch.setattr(app.state.codex, "detect", lambda *args: "/synthetic-agent")
|
|
monkeypatch.setattr(app.state.codex, "_spawn", lambda *args: 4242)
|
|
created = client.post(
|
|
"/api/orchestrations", headers=headers,
|
|
json={
|
|
"project_id": project_id, "name": "Refonte jetable",
|
|
"request": "Modifier le fichier de démonstration sans toucher au dépôt principal avant validation.",
|
|
"acceptance_criteria": ["Le contenu final vaut approved", "La vérification dédiée réussit"],
|
|
"tests": [["python3", "-c", "from pathlib import Path; assert Path('app.txt').read_text().strip() in {'draft','approved'}"]],
|
|
"engine": "codex",
|
|
},
|
|
)
|
|
assert created.status_code == 201, created.text
|
|
flow = created.json()
|
|
assert flow["status"] == "DRAFT_REVIEW"
|
|
|
|
rejected = client.post(
|
|
f"/api/orchestrations/{flow['id']}/draft/reject", headers=headers,
|
|
json={"feedback": "Préciser la preuve attendue et conserver la branche dédiée."},
|
|
)
|
|
assert rejected.status_code == 200
|
|
flow = rejected.json()
|
|
assert flow["draft_version"] == 2
|
|
assert "preuve attendue" in flow["draft"]
|
|
|
|
approved = client.post(
|
|
f"/api/orchestrations/{flow['id']}/draft/approve", headers=headers,
|
|
json={"draft_hash": flow["draft_hash"]},
|
|
)
|
|
assert approved.status_code == 200, approved.text
|
|
flow = approved.json()
|
|
worktree = finish_agent(app, flow, "draft\n")
|
|
assert not (root / "app.txt").read_text().startswith("draft")
|
|
|
|
first_detail = client.get(f"/api/orchestrations/{flow['id']}").json()
|
|
second_detail = client.get(f"/api/orchestrations/{flow['id']}").json()
|
|
assert first_detail["status"] == second_detail["status"] == "AWAITING_VERIFICATION"
|
|
assert len([event for event in second_detail["events"] if event["kind"] == "agent.finished"]) == 1
|
|
|
|
verified = client.post(f"/api/orchestrations/{flow['id']}/verify", headers=headers)
|
|
assert verified.status_code == 200, verified.text
|
|
flow = verified.json()
|
|
assert flow["status"] == "USER_REVIEW"
|
|
first_revision = flow["verified_revision"]
|
|
|
|
corrections = client.post(
|
|
f"/api/orchestrations/{flow['id']}/corrections", headers=headers,
|
|
json={"feedback": "La recette refuse draft : remplacer par approved."},
|
|
)
|
|
assert corrections.status_code == 200, corrections.text
|
|
flow = corrections.json()
|
|
assert flow["conversation_id"] == first_detail["conversation_id"]
|
|
assert Path(flow["worktree_path"]) == worktree
|
|
finish_agent(app, flow, "approved\n")
|
|
assert client.get(f"/api/orchestrations/{flow['id']}").json()["status"] == "AWAITING_VERIFICATION"
|
|
|
|
flow = client.post(f"/api/orchestrations/{flow['id']}/verify", headers=headers).json()
|
|
assert flow["status"] == "USER_REVIEW"
|
|
assert flow["verified_revision"] != first_revision
|
|
flow = client.post(
|
|
f"/api/orchestrations/{flow['id']}/approve", headers=headers,
|
|
json={"revision": flow["verified_revision"]},
|
|
).json()
|
|
assert flow["status"] == "AWAITING_MERGE_CONFIRMATION"
|
|
assert client.post(f"/api/orchestrations/{flow['id']}/merge", headers=headers, json={"confirm": False}).status_code == 409
|
|
|
|
merged = client.post(f"/api/orchestrations/{flow['id']}/merge", headers=headers, json={"confirm": True})
|
|
assert merged.status_code == 200, merged.text
|
|
assert merged.json()["status"] == "MERGED"
|
|
assert (root / "app.txt").read_text(encoding="utf-8") == "approved\n"
|
|
|
|
|
|
def test_orchestration_cannot_be_started_by_unauthorized_user(logged_in):
|
|
client, app, headers, _, project_id = disposable_repository(logged_in)
|
|
app.state.auth.create_user("other", "a-different-long-test-password", is_admin=False)
|
|
client.post("/api/auth/logout", headers=headers)
|
|
login = client.post("/api/auth/login", json={"username": "other", "password": "a-different-long-test-password"}).json()
|
|
response = client.post(
|
|
"/api/orchestrations", headers={"X-CSRF-Token": login["csrf_token"]},
|
|
json={"project_id": project_id, "name": "Interdit", "request": "Cette demande ne doit jamais démarrer dans le projet privé.", "acceptance_criteria": ["Refus"], "tests": []},
|
|
)
|
|
assert response.status_code == 404
|