mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-21 14:08:11 +02:00
Retire la pile Python/React : Loki repart de la base Go d'AJEAN
L'ancien Loki (FastAPI + React + Ollama) est remplacé par le fork d'AJEAN. Tout reste accessible dans l'historique (git show bb4fb1b:backend/app/agent.py).
This commit is contained in:
87 files changed
-18730
No files matched your search
Whitespace-only changes.
@@ -1,124 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
from app import agent # noqa: E402
|
||||
|
||||
|
||||
def _convo():
|
||||
return [
|
||||
{"role": "system", "content": "consigne"},
|
||||
{"role": "user", "content": "fais la tâche"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_relance_apres_reflexion_seule(monkeypatch):
|
||||
"""1er appel : réflexion seule -> relance ; 2e appel : réponse finale."""
|
||||
calls: list[list[dict]] = []
|
||||
|
||||
async def fake_chat(model, convo, **kwargs):
|
||||
calls.append([dict(m) for m in convo])
|
||||
if len(calls) == 1:
|
||||
yield {"message": {"thinking": "hmm, je réfléchis longuement…"},
|
||||
"done": False}
|
||||
yield {"message": {}, "done": True}
|
||||
else:
|
||||
yield {"message": {"content": "réponse finale"}, "done": True}
|
||||
|
||||
monkeypatch.setattr(agent.ollama, "chat", fake_chat)
|
||||
events = [
|
||||
e async for e in agent.run_agent("test", _convo(), enabled_tools=[])
|
||||
]
|
||||
|
||||
final = [e for e in events if e["type"] == "final"]
|
||||
assert final and "réponse finale" in final[0]["content"]
|
||||
# La relance a bien injecté la consigne de reprise.
|
||||
assert any(
|
||||
"Continue la tâche" in m["content"]
|
||||
for m in calls[1] if m["role"] == "user"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reflexion_pas_renvoyee_au_modele(monkeypatch):
|
||||
"""La pensée est affichée mais jamais réinjectée dans l'historique."""
|
||||
calls: list[list[dict]] = []
|
||||
|
||||
async def fake_chat(model, convo, **kwargs):
|
||||
calls.append([dict(m) for m in convo])
|
||||
if len(calls) == 1:
|
||||
yield {"message": {"thinking": "je planifie",
|
||||
"content": "étape 1",
|
||||
"tool_calls": [{"function": {
|
||||
"name": "list_dir",
|
||||
"arguments": {"path": "."}}}]},
|
||||
"done": True}
|
||||
else:
|
||||
yield {"message": {"content": "terminé"}, "done": True}
|
||||
|
||||
monkeypatch.setattr(agent.ollama, "chat", fake_chat)
|
||||
events = [
|
||||
e async for e in agent.run_agent("test", _convo(), enabled_tools=None)
|
||||
]
|
||||
|
||||
final = [e for e in events if e["type"] == "final"]
|
||||
assert final and "terminé" in final[0]["content"]
|
||||
assert "je planifie" in final[0]["thinking"] # gardée pour l'UI
|
||||
# Aucun message assistant réinjecté ne contient la clé thinking.
|
||||
assistant_turns = [m for m in calls[1] if m["role"] == "assistant"]
|
||||
assert assistant_turns
|
||||
assert all("thinking" not in m for m in assistant_turns)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_coupe_circuit_pensee_interminable(monkeypatch):
|
||||
"""Pensée sans fin -> génération coupée en vol, puis relance qui aboutit."""
|
||||
calls: list[int] = []
|
||||
|
||||
async def fake_chat(model, convo, **kwargs):
|
||||
calls.append(1)
|
||||
if len(calls) == 1:
|
||||
# Flux de pensée « infini » : jamais de done, jamais de contenu.
|
||||
for _ in range(10_000):
|
||||
yield {"message": {"thinking": "x" * 200}, "done": False}
|
||||
else:
|
||||
yield {"message": {"content": "réponse après coupe"}, "done": True}
|
||||
|
||||
monkeypatch.setattr(agent.ollama, "chat", fake_chat)
|
||||
events = [
|
||||
e async for e in agent.run_agent("test", _convo(), enabled_tools=[])
|
||||
]
|
||||
|
||||
final = [e for e in events if e["type"] == "final"]
|
||||
assert final and "réponse après coupe" in final[0]["content"]
|
||||
assert len(calls) == 2 # coupé puis relancé, pas d'épuisement du flux
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reflexion_coupee_apres_deux_impasses(monkeypatch):
|
||||
"""Deux itérations de pensée pure -> think désactivé, tâche finie."""
|
||||
seen_think: list = []
|
||||
|
||||
async def fake_chat(model, convo, think=None, **kwargs):
|
||||
seen_think.append(think)
|
||||
if len(seen_think) <= 2:
|
||||
yield {"message": {"thinking": "boucle de pensée"}, "done": True}
|
||||
else:
|
||||
yield {"message": {"content": "fini sans réfléchir"}, "done": True}
|
||||
|
||||
monkeypatch.setattr(agent.ollama, "chat", fake_chat)
|
||||
events = [
|
||||
e async for e in agent.run_agent(
|
||||
"test", _convo(), enabled_tools=[], think=True
|
||||
)
|
||||
]
|
||||
|
||||
final = [e for e in events if e["type"] == "final"]
|
||||
assert final and "fini sans réfléchir" in final[0]["content"]
|
||||
assert seen_think[-1] is False # think coupé pour l'appel final
|
||||
assert any(e["type"] == "notice" for e in events)
|
||||
@@ -1,62 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
from app import agent, db, mcp_client # noqa: E402
|
||||
|
||||
db.init_db()
|
||||
|
||||
|
||||
def test_prune_compacte_les_anciens_resultats():
|
||||
big = "x" * 5000
|
||||
convo = [
|
||||
{"role": "system", "content": "consigne"},
|
||||
{"role": "user", "content": "tâche"},
|
||||
{"role": "assistant", "content": "", "tool_calls": [{}]},
|
||||
{"role": "tool", "tool_name": "read_file", "content": big},
|
||||
{"role": "assistant", "content": "", "tool_calls": [{}]},
|
||||
{"role": "tool", "tool_name": "grep_search", "content": big},
|
||||
]
|
||||
# Le dernier lot (index 5) commence à 5 : seul l'index 3 est compacté.
|
||||
agent._prune_old_tool_results(convo, before_index=5)
|
||||
assert len(convo[3]["content"]) < 500
|
||||
assert "archivé" in convo[3]["content"]
|
||||
assert convo[5]["content"] == big # lot courant intact
|
||||
assert convo[0]["content"] == "consigne" # système intact
|
||||
|
||||
|
||||
def test_prune_ignore_les_petits_resultats():
|
||||
convo = [{"role": "tool", "tool_name": "list_dir", "content": "court"}]
|
||||
agent._prune_old_tool_results(convo, before_index=1)
|
||||
assert convo[0]["content"] == "court"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_searxng_sans_url_erreur_claire():
|
||||
mgr = mcp_client.McpManager()
|
||||
try:
|
||||
result = await mgr.test_server("searxng")
|
||||
assert result["ok"] is False
|
||||
assert "SEARXNG_URL" in (result["error"] or "")
|
||||
assert "requis" in (result["error"] or "")
|
||||
finally:
|
||||
await mgr.aclose()
|
||||
|
||||
|
||||
def test_migration_v6_monte_les_anciens_defauts():
|
||||
from app import agent_config
|
||||
profiles = db.get_config_value(agent_config.MODEL_PROFILES_KEY) or {}
|
||||
profiles["testmodel:7b"] = {"num_ctx": 8192}
|
||||
profiles["custom:7b"] = {"num_ctx": 5000}
|
||||
db.set_config_value(agent_config.MODEL_PROFILES_KEY, profiles)
|
||||
db.set_config_value(agent_config.PROFILE_STATE_KEY, {"version": 5})
|
||||
|
||||
agent_config._migrate_profiles()
|
||||
|
||||
migrated = db.get_config_value(agent_config.MODEL_PROFILES_KEY)
|
||||
assert migrated["testmodel:7b"]["num_ctx"] == 16384 # ancien défaut monté
|
||||
assert migrated["custom:7b"]["num_ctx"] == 5000 # valeur perso respectée
|
||||
@@ -1,39 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from app import db # noqa: E402
|
||||
from app import mcp_client # noqa: E402
|
||||
|
||||
db.init_db()
|
||||
|
||||
|
||||
def test_catalog_has_preconfigured_servers():
|
||||
for sid in ("playwright", "context7", "fetch", "searxng", "custom"):
|
||||
assert sid in mcp_client.CATALOG
|
||||
assert mcp_client.CATALOG[sid]["label"]
|
||||
|
||||
|
||||
def test_state_defaults_disabled():
|
||||
state = mcp_client.get_mcp_state()
|
||||
assert set(state) == set(mcp_client.CATALOG)
|
||||
assert all(not s["enabled"] for s in state.values())
|
||||
|
||||
|
||||
def test_toggle_persists():
|
||||
mcp_client.set_mcp_state("fetch", enabled=True, params={})
|
||||
assert mcp_client.get_mcp_state()["fetch"]["enabled"] is True
|
||||
mcp_client.set_mcp_state("fetch", enabled=False, params={})
|
||||
assert mcp_client.get_mcp_state()["fetch"]["enabled"] is False
|
||||
|
||||
|
||||
def test_custom_params_persist():
|
||||
mcp_client.set_mcp_state(
|
||||
"custom", enabled=False, params={"command": "npx -y some-mcp"}
|
||||
)
|
||||
assert (
|
||||
mcp_client.get_mcp_state()["custom"]["params"]["command"]
|
||||
== "npx -y some-mcp"
|
||||
)
|
||||
@@ -1,96 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import textwrap
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
from app import db, mcp_client # noqa: E402
|
||||
|
||||
db.init_db()
|
||||
|
||||
# Serveur MCP minimal : un outil "echo" + un outil au nom À TIRETS
|
||||
# (comme Context7 « resolve-library-id »).
|
||||
_FAKE_SERVER = textwrap.dedent("""
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
mcp = FastMCP("fake")
|
||||
|
||||
@mcp.tool()
|
||||
def echo(text: str) -> str:
|
||||
\"\"\"Répète le texte fourni.\"\"\"
|
||||
return "echo:" + text
|
||||
|
||||
@mcp.tool(name="dash-tool-name")
|
||||
def dash_tool(text: str) -> str:
|
||||
\"\"\"Outil au nom à tirets.\"\"\"
|
||||
return "dash:" + text
|
||||
|
||||
mcp.run()
|
||||
""")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def fake_server_cmd(tmp_path):
|
||||
path = tmp_path / "fake_mcp.py"
|
||||
path.write_text(_FAKE_SERVER, encoding="utf-8")
|
||||
return [sys.executable, str(path)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tools_exposed_and_called(fake_server_cmd, monkeypatch):
|
||||
monkeypatch.setitem(
|
||||
mcp_client.CATALOG, "fake",
|
||||
{"label": "Fake", "description": "", "command": fake_server_cmd,
|
||||
"url_param": False, "env_params": [], "expose": None},
|
||||
)
|
||||
mcp_client.set_mcp_state("fake", enabled=True, params={})
|
||||
mgr = mcp_client.McpManager()
|
||||
try:
|
||||
defs = await mgr.tool_definitions()
|
||||
names = [d["function"]["name"] for d in defs]
|
||||
assert "mcp_fake_echo" in names
|
||||
# Nom à tirets exposé assaini (compatibilité function-calling).
|
||||
assert "mcp_fake_dash_tool_name" in names
|
||||
result = await mgr.call_tool("mcp_fake_echo", {"text": "bonjour"})
|
||||
assert result["ok"] is True
|
||||
assert "echo:bonjour" in result["content"]
|
||||
# Appel via le nom assaini -> résolu vers le vrai nom à tirets.
|
||||
dash = await mgr.call_tool("mcp_fake_dash_tool_name", {"text": "x"})
|
||||
assert dash["ok"] is True and "dash:x" in dash["content"]
|
||||
# Tolérance : le modèle répond avec des tirets au lieu d'underscores.
|
||||
mixed = await mgr.call_tool("mcp_fake_dash-tool-name", {"text": "y"})
|
||||
assert mixed["ok"] is True and "dash:y" in mixed["content"]
|
||||
finally:
|
||||
await mgr.aclose()
|
||||
mcp_client.set_mcp_state("fake", enabled=False, params={})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disabled_server_exposes_nothing():
|
||||
mgr = mcp_client.McpManager()
|
||||
try:
|
||||
assert await mgr.tool_definitions() == []
|
||||
finally:
|
||||
await mgr.aclose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_broken_server_never_raises(monkeypatch):
|
||||
monkeypatch.setitem(
|
||||
mcp_client.CATALOG, "broken",
|
||||
{"label": "Broken", "description": "",
|
||||
"command": [sys.executable, "-c", "import sys; sys.exit(3)"],
|
||||
"url_param": False, "env_params": [], "expose": None},
|
||||
)
|
||||
mcp_client.set_mcp_state("broken", enabled=True, params={})
|
||||
mgr = mcp_client.McpManager()
|
||||
try:
|
||||
assert await mgr.tool_definitions() == []
|
||||
assert mgr.statuses()["broken"]["state"] == "error"
|
||||
assert any("broken" in n.lower() or "Broken" in n for n in mgr.notices())
|
||||
finally:
|
||||
await mgr.aclose()
|
||||
mcp_client.set_mcp_state("broken", enabled=False, params={})
|
||||
@@ -1,62 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
import pytest # noqa: E402
|
||||
|
||||
from app import db, tools # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
db.init_db()
|
||||
_ROOT = os.path.abspath(settings.workspace_dir)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_project():
|
||||
tools.set_project(None)
|
||||
yield
|
||||
tools.set_project(None)
|
||||
|
||||
|
||||
def test_racine_par_defaut():
|
||||
assert tools.active_root() == _ROOT
|
||||
|
||||
|
||||
def test_set_project_reracine():
|
||||
tools.set_project("demo")
|
||||
root = tools.active_root()
|
||||
assert root == os.path.join(_ROOT, "demo")
|
||||
assert os.path.isdir(root) # créée à la volée
|
||||
|
||||
|
||||
def test_confinement_conserve():
|
||||
tools.set_project("demo")
|
||||
with pytest.raises(tools.ToolError):
|
||||
tools._safe_path("../hors-projet")
|
||||
|
||||
|
||||
def test_nom_projet_invalide():
|
||||
for bad in ("../x", "UPPER", "a b", "", "x" * 50):
|
||||
with pytest.raises(tools.ToolError):
|
||||
tools.set_project(bad)
|
||||
|
||||
|
||||
def test_session_porte_son_projet():
|
||||
s = db.create_session("t", None, project="demo")
|
||||
assert db.get_session(s["id"])["project"] == "demo"
|
||||
db.set_session_project(s["id"], None)
|
||||
assert db.get_session(s["id"])["project"] is None
|
||||
|
||||
|
||||
def test_aides_contexte_suivent_le_projet():
|
||||
from app.routes.chat import _mentioned_files, _workspace_listing
|
||||
tools.set_project("ctxdemo")
|
||||
root = tools.active_root()
|
||||
with open(os.path.join(root, "app.py"), "w", encoding="utf-8") as f:
|
||||
f.write("x = 1")
|
||||
assert "app.py" in _workspace_listing()
|
||||
assert _mentioned_files("corrige app.py") == ["app.py"]
|
||||
tools.set_project(None)
|
||||
assert _mentioned_files("corrige app.py") == [] # absent de la racine
|
||||
@@ -1,46 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
|
||||
from app.main import app # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
client = TestClient(app)
|
||||
_ROOT = os.path.abspath(settings.workspace_dir)
|
||||
|
||||
|
||||
def test_creation_et_liste():
|
||||
r = client.post("/api/projects", json={"name": "routedemo"})
|
||||
assert r.status_code == 201
|
||||
assert os.path.isdir(os.path.join(_ROOT, "routedemo", ".git"))
|
||||
names = [p["name"] for p in client.get("/api/projects").json()["projects"]]
|
||||
assert "routedemo" in names
|
||||
|
||||
|
||||
def test_nom_invalide_400():
|
||||
assert client.post("/api/projects", json={"name": "../x"}).status_code == 400
|
||||
assert client.post("/api/projects", json={"name": "Demo"}).status_code == 400
|
||||
|
||||
|
||||
def test_existant_400():
|
||||
client.post("/api/projects", json={"name": "dup"})
|
||||
assert client.post("/api/projects", json={"name": "dup"}).status_code == 400
|
||||
|
||||
|
||||
def test_files_re_racine():
|
||||
client.post("/api/projects", json={"name": "scoped"})
|
||||
with open(os.path.join(_ROOT, "scoped", "a.txt"), "w") as f:
|
||||
f.write("x")
|
||||
with open(os.path.join(_ROOT, "racine.txt"), "w") as f:
|
||||
f.write("y")
|
||||
tree = client.get("/api/files", params={"project": "scoped"}).json()["tree"]
|
||||
names = [n["name"] for n in tree]
|
||||
assert "a.txt" in names and "racine.txt" not in names
|
||||
|
||||
|
||||
def test_files_projet_invalide_400():
|
||||
assert client.get("/api/files", params={"project": "../x"}).status_code == 400
|
||||
@@ -1,96 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from app import router # noqa: E402
|
||||
from app.routes.chat import ( # noqa: E402
|
||||
_mentioned_files,
|
||||
_prev_was_code,
|
||||
_session_code_context,
|
||||
_workspace_listing,
|
||||
)
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
|
||||
def test_message_code_explicite():
|
||||
assert router.is_code_task("crée une page html avec un formulaire") is True
|
||||
|
||||
|
||||
def test_message_banal_pas_code():
|
||||
assert router.is_code_task("quelle heure est-il ?") is False
|
||||
|
||||
|
||||
def test_followup_courts():
|
||||
# Suites typiques d'un travail de code, sans vocabulaire code explicite.
|
||||
assert router.is_code_followup("ajoute un bouton rouge en haut") is True
|
||||
assert router.is_code_followup("continue") is True
|
||||
assert router.is_code_followup("enlève le menu et agrandis le titre") is True
|
||||
assert router.is_code_followup("merci beaucoup !") is False
|
||||
|
||||
|
||||
def test_prev_was_code_via_engine():
|
||||
history = [
|
||||
{"role": "user", "content": "crée un site", "meta": None},
|
||||
{"role": "assistant", "content": "fait", "meta": {"engine": "code"}},
|
||||
]
|
||||
assert _prev_was_code(history) is True
|
||||
|
||||
|
||||
def test_prev_was_code_via_write_file():
|
||||
history = [
|
||||
{"role": "assistant", "content": "ok",
|
||||
"meta": {"tools": [{"name": "write_file", "args": {}}]}},
|
||||
]
|
||||
assert _prev_was_code(history) is True
|
||||
|
||||
|
||||
def test_prev_was_code_discussion():
|
||||
history = [
|
||||
{"role": "assistant", "content": "voici la réponse", "meta": None},
|
||||
]
|
||||
assert _prev_was_code(history) is False
|
||||
|
||||
|
||||
def test_mentioned_files_existants_seulement():
|
||||
root = os.path.abspath(settings.workspace_dir)
|
||||
os.makedirs(root, exist_ok=True)
|
||||
with open(os.path.join(root, "index.html"), "w", encoding="utf-8") as f:
|
||||
f.write("<html></html>")
|
||||
found = _mentioned_files("modifie index.html et style.css")
|
||||
assert found == ["index.html"] # style.css n'existe pas
|
||||
|
||||
|
||||
def test_session_code_context_recap_et_fichiers():
|
||||
root = os.path.abspath(settings.workspace_dir)
|
||||
os.makedirs(root, exist_ok=True)
|
||||
with open(os.path.join(root, "jeu.html"), "w", encoding="utf-8") as f:
|
||||
f.write("<html>jeu</html>")
|
||||
history = [
|
||||
{"role": "user", "content": "crée un jeu snake dans jeu.html",
|
||||
"meta": None},
|
||||
{"role": "assistant", "content": "fait",
|
||||
"meta": {"engine": "code",
|
||||
"tools": [{"name": "write_file",
|
||||
"args": {"path": "jeu.html"}}]}},
|
||||
{"role": "user", "content": "il y a des bugs", "meta": None},
|
||||
]
|
||||
recap, files = _session_code_context(history)
|
||||
assert "crée un jeu snake" in recap
|
||||
assert files == ["jeu.html"]
|
||||
|
||||
|
||||
def test_session_code_context_vide_sans_historique():
|
||||
recap, files = _session_code_context([])
|
||||
assert recap == ""
|
||||
assert files == []
|
||||
|
||||
|
||||
def test_workspace_listing_contient_fichiers():
|
||||
root = os.path.abspath(settings.workspace_dir)
|
||||
os.makedirs(root, exist_ok=True)
|
||||
with open(os.path.join(root, "jeu.html"), "w", encoding="utf-8") as f:
|
||||
f.write("x")
|
||||
listing = _workspace_listing()
|
||||
assert "jeu.html" in listing
|
||||
@@ -1,43 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from app import tools # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
# Le workspace effectif peut avoir été fixé par un autre fichier de test
|
||||
# importé avant celui-ci : on écrit là où les outils lisent réellement.
|
||||
_WS = os.path.abspath(settings.workspace_dir)
|
||||
os.makedirs(_WS, exist_ok=True)
|
||||
|
||||
|
||||
def _write(name: str, content: str) -> str:
|
||||
path = os.path.join(_WS, name)
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return name
|
||||
|
||||
|
||||
def test_python_valide():
|
||||
rel = _write("ok.py", "x = 1\nprint(x)\n")
|
||||
assert tools.run_check(rel)["ok"] is True
|
||||
|
||||
|
||||
def test_python_casse():
|
||||
rel = _write("ko.py", "def broken(:\n")
|
||||
result = tools.run_check(rel)
|
||||
assert result["ok"] is False
|
||||
assert result["issues"]
|
||||
|
||||
|
||||
def test_json_casse():
|
||||
rel = _write("ko.json", "{invalid")
|
||||
assert tools.run_check(rel)["ok"] is False
|
||||
|
||||
|
||||
def test_fichier_inconnu_type():
|
||||
rel = _write("notes.txt", "bonjour")
|
||||
result = tools.run_check(rel)
|
||||
assert result["ok"] is True # type non vérifiable = pas d'erreur
|
||||
@@ -1,26 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from app import skills # noqa: E402
|
||||
|
||||
|
||||
def test_five_skills_loaded():
|
||||
assert len(skills.ALL_SKILLS) == 5
|
||||
assert all(s["title"] and s["body"] for s in skills.ALL_SKILLS.values())
|
||||
|
||||
|
||||
def test_bug_message_picks_debug():
|
||||
s = skills.pick_skill("mon script plante avec une erreur TypeError au démarrage")
|
||||
assert s and s["name"] == "debogage-systematique"
|
||||
|
||||
|
||||
def test_web_message_picks_web():
|
||||
s = skills.pick_skill("crée une page html responsive pour ma boutique")
|
||||
assert s and s["name"] == "creation-web"
|
||||
|
||||
|
||||
def test_banal_message_picks_nothing():
|
||||
assert skills.pick_skill("bonjour, quelle heure est-il ?") is None
|
||||
@@ -1,69 +0,0 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
|
||||
os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
|
||||
|
||||
from app import tools # noqa: E402
|
||||
from app.config import settings # noqa: E402
|
||||
|
||||
_WS = os.path.abspath(settings.workspace_dir)
|
||||
os.makedirs(_WS, exist_ok=True)
|
||||
|
||||
|
||||
# ── read_file fenêtré ────────────────────────────────────────────────────
|
||||
|
||||
def _write(name: str, content: str) -> str:
|
||||
with open(os.path.join(_WS, name), "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
return name
|
||||
|
||||
|
||||
def test_read_file_petit_entier():
|
||||
rel = _write("petit.txt", "a\nb\nc\n")
|
||||
result = tools.read_file(rel)
|
||||
assert result["content"] == "a\nb\nc"
|
||||
assert "3 lignes" in result["summary"]
|
||||
|
||||
|
||||
def test_read_file_gros_fenetre():
|
||||
rel = _write("gros.txt", "\n".join(f"ligne {i}" for i in range(1, 501)))
|
||||
result = tools.read_file(rel)
|
||||
assert "ligne 200" in result["content"]
|
||||
assert "ligne 201" not in result["content"].replace("start_line=201", "")
|
||||
assert "start_line=201" in result["content"] # marche à suivre
|
||||
assert "1-200 sur 500" in result["summary"]
|
||||
|
||||
|
||||
def test_read_file_fenetre_suivante():
|
||||
rel = _write("gros2.txt", "\n".join(f"ligne {i}" for i in range(1, 501)))
|
||||
result = tools.read_file(rel, start_line=201)
|
||||
assert "ligne 201" in result["content"]
|
||||
assert "201-400 sur 500" in result["summary"]
|
||||
|
||||
|
||||
# ── sortie shell filtrée ─────────────────────────────────────────────────
|
||||
|
||||
def test_compact_succes_garde_la_fin():
|
||||
out = "\n".join(f"étape {i}" for i in range(1, 101))
|
||||
compact = tools._compact_output(out, 0)
|
||||
assert "étape 100" in compact
|
||||
assert "étape 1\n" not in compact
|
||||
assert "lignes omises" in compact
|
||||
|
||||
|
||||
def test_compact_echec_garde_les_erreurs():
|
||||
out = "\n".join(
|
||||
["compilation démarrée"]
|
||||
+ [f"module {i} ok" for i in range(50)]
|
||||
+ ["ERROR: variable x undefined", "build failed"]
|
||||
)
|
||||
compact = tools._compact_output(out, 1)
|
||||
assert "ERROR: variable x undefined" in compact
|
||||
assert "build failed" in compact
|
||||
assert "module 3 ok" not in compact
|
||||
|
||||
|
||||
def test_dedupe_repetitions():
|
||||
lines = ["warn: deprecated"] * 5 + ["fin"]
|
||||
assert tools._dedupe_lines(lines) == ["warn: deprecated ×5", "fin"]
|
||||
Reference in new issue
Block a user