58 lines
3.1 KiB
Python
58 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from codexpi.rules_service import RulesService
|
|
from codexpi.session_worker import engine_command
|
|
|
|
|
|
def add_project(app, projects: Path, owner: str, engine: str = "codex") -> tuple[dict, dict]:
|
|
root = projects / f"rules-{engine}"
|
|
root.mkdir()
|
|
project_id, conversation_id = str(uuid.uuid4()), str(uuid.uuid4())
|
|
app.state.db.execute(
|
|
"INSERT INTO projects(id,name,path,owner_id,engine,project_rules,created_at) VALUES(?,?,?,?,?,?,?)",
|
|
(project_id, f"Rules {engine}", str(root), owner, engine, "Toujours exécuter le test projet ciblé.", "2026-01-01"),
|
|
)
|
|
conversation = app.state.codex.create_conversation(project_id, "Rules", user_id=owner, engine=engine, conversation_id=conversation_id)
|
|
return app.state.db.one("SELECT * FROM projects WHERE id=?", (project_id,)), conversation
|
|
|
|
|
|
@pytest.mark.parametrize("engine", ["codex", "claude", "opencode"])
|
|
def test_rules_are_snapshotted_for_every_supported_engine(logged_in, monkeypatch, engine):
|
|
_, app, projects, _, _ = logged_in
|
|
owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
|
|
project, conversation = add_project(app, projects, owner, engine)
|
|
monkeypatch.setattr(app.state.codex, "detect", lambda *args: "/synthetic-agent")
|
|
monkeypatch.setattr(app.state.codex, "_spawn", lambda *args: 123)
|
|
|
|
queued = app.state.codex.enqueue(conversation, project, "Fais le travail", requested_by=owner)
|
|
row = app.state.db.one("SELECT * FROM prompt_queue WHERE id=?", (queued["id"],))
|
|
|
|
assert row["rules_version"] == 1
|
|
assert row["project_rules_snapshot"] == "Toujours exécuter le test projet ciblé."
|
|
envelope = RulesService(app.state.db).envelope(row["rules_version"], row["project_rules_snapshot"])
|
|
command = engine_command(engine, "/synthetic-agent", envelope, None, False)
|
|
assert any("RÈGLES COMMUNES" in value for value in command)
|
|
assert any("CONSIGNES DU PROJET" in value for value in command)
|
|
|
|
|
|
def test_rule_update_is_versioned_and_existing_queue_keeps_snapshot(logged_in, monkeypatch):
|
|
client, app, projects, _, headers = logged_in
|
|
owner = app.state.db.one("SELECT id FROM users LIMIT 1")["id"]
|
|
project, conversation = add_project(app, projects, owner)
|
|
monkeypatch.setattr(app.state.codex, "detect", lambda *args: "/synthetic-agent")
|
|
monkeypatch.setattr(app.state.codex, "_spawn", lambda *args: 123)
|
|
first = app.state.codex.enqueue(conversation, project, "Avant", requested_by=owner)
|
|
|
|
content = "# Règles communes\n\n" + "\n".join(f"- Règle vérifiable {index}." for index in range(1, 8))
|
|
response = client.put("/api/settings/rules", headers=headers, json={"content": content})
|
|
assert response.status_code == 200, response.text
|
|
second = app.state.codex.enqueue(conversation, project, "Après", requested_by=owner)
|
|
|
|
assert app.state.db.one("SELECT rules_version FROM prompt_queue WHERE id=?", (first["id"],))["rules_version"] == 1
|
|
assert app.state.db.one("SELECT rules_version FROM prompt_queue WHERE id=?", (second["id"],))["rules_version"] == 2
|