File size: 3,487 Bytes
1da7ac7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | # tests/test_cli.py
from pathlib import Path
from fractus_vorax.agent import cli
CSV_CONTENT = (
"question,answer\n"
"what is the capital of france,paris\n"
"what is the capital of spain,madrid\n"
"what is the capital of japan,tokyo\n"
"who wrote hamlet,william shakespeare\n"
)
def _make_csv(tmp_path):
p = tmp_path / "cap.csv"
p.write_text(CSV_CONTENT, encoding="utf-8")
return p
def test_ingest_then_ask_end_to_end(tmp_path, capsys):
csv = _make_csv(tmp_path)
brain_dir = tmp_path / "brain"
n = cli.ingest(csv, brain_dir, D=2048, seed=0, kn_cache=tmp_path / "kn")
assert n == 4
assert (brain_dir / "meta.json").exists()
assert (tmp_path / "kn" / "cap.kn" / "manifest.json").exists() # l'artefact .kn existe
lines = cli.ask("what is the capital of japan", brain_dir, k=1)
out = "\n".join(lines)
assert "tokyo" in out
assert "[CARTE]" in out
assert "=>" in out # format v2 : FAIT {statement} => {context} (remplace RÉP:)
print(out)
captured = capsys.readouterr()
assert "tokyo" in captured.out
def test_ask_empty_brain(tmp_path):
lines = cli.ask("anything", tmp_path / "nope", k=1)
assert lines == ["[CARTE] LACUNES: aucune trace (brain vide ou absent)"]
def test_status_reports_counts(tmp_path, capsys):
csv = _make_csv(tmp_path)
cli.ingest(csv, tmp_path / "brain", D=2048, seed=0, kn_cache=tmp_path / "kn")
s = cli.status(tmp_path / "brain")
assert "4" in s
def test_main_subcommands(tmp_path, capsys):
csv = _make_csv(tmp_path)
brain = tmp_path / "brain"
assert cli.main(["ingest", str(csv), "--brain", str(brain), "--D", "2048", "--kn-cache", str(tmp_path / "kn")]) == 0
assert cli.main(["ask", "who wrote hamlet", "--brain", str(brain), "--k", "1"]) == 0
assert "william shakespeare" in capsys.readouterr().out
assert cli.main(["status", "--brain", str(brain)]) == 0
def test_ingest_recompiles_when_source_changes(tmp_path):
csv = tmp_path / "cap.csv"
csv.write_text(
"question,answer\nwhat is the capital of france,paris\nwho wrote hamlet,william shakespeare\n",
encoding="utf-8",
)
brain = tmp_path / "brain"
cli.ingest(csv, brain, D=2048, seed=0, kn_cache=tmp_path / "kn")
# Source éditée : 3 QA nouvelles — le cache .kn doit être invalidé puis recompilé.
csv.write_text(
"question,answer\nwhat color is the sky,blue\nwhat sound does a cow make,moo\nwhat is two plus two,four\n",
encoding="utf-8",
)
cli.ingest(csv, brain, D=2048, seed=0, kn_cache=tmp_path / "kn")
out = "\n".join(cli.ask("what sound does a cow make", brain, k=1))
assert "moo" in out # réponse uniquement présente dans les nouvelles lignes
def test_ingest_reuses_cache_when_source_unchanged(tmp_path):
csv = _make_csv(tmp_path)
kn_dir = tmp_path / "kn" / "cap.kn"
cli.ingest(csv, tmp_path / "brain", D=2048, seed=0, kn_cache=tmp_path / "kn")
mtime = (kn_dir / "atoms.tsv.gz").stat().st_mtime_ns
cli.ingest(csv, tmp_path / "brain2", D=2048, seed=0, kn_cache=tmp_path / "kn")
assert (kn_dir / "atoms.tsv.gz").stat().st_mtime_ns == mtime # pas de recompilation
def test_main_missing_source_clean_error(tmp_path, capsys):
rc = cli.main(["ingest", str(tmp_path / "nope.csv"), "--brain", str(tmp_path / "b")])
assert rc == 1
out = capsys.readouterr().out
assert any(line.startswith("[VORAX] ERREUR:") for line in out.splitlines())
|