File size: 3,302 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 93 94 95 96 97 | # tests/test_parse.py
import json
from pathlib import Path
from fractus_vorax.compiler.parse import detect_format, flatten_record, parse_file
def _write(tmp_path, name, content):
p = tmp_path / name
p.write_text(content, encoding="utf-8")
return p
def test_detect_format(tmp_path):
assert detect_format(_write(tmp_path, "a.csv", "x\n1")) == "csv"
assert detect_format(_write(tmp_path, "a.tsv", "x\n1")) == "tsv"
assert detect_format(_write(tmp_path, "a.jsonl", "{}")) == "jsonl"
assert detect_format(_write(tmp_path, "a.json", "[]")) == "json"
assert detect_format(_write(tmp_path, "a.md", "# t")) == "md"
assert detect_format(_write(tmp_path, "a.txt", "hi")) == "txt"
assert detect_format(_write(tmp_path, "a.xyz", "hi")) == "raw"
def test_parse_csv_two_columns_is_qa(tmp_path):
p = _write(tmp_path, "cap.csv", "question,answer\nwhat is the capital of france,paris\nwho wrote hamlet,william shakespeare\n")
atoms = parse_file(p)
assert len(atoms) == 2
assert atoms[0].statement == "what is the capital of france"
assert atoms[0].context == "paris"
assert atoms[0].provenance.startswith("cap.csv")
assert atoms[0].confidence == 1.0
def test_parse_csv_wide_table_serializes_pairs(tmp_path):
p = _write(tmp_path, "rows.csv", "name,city,height\neiffel,paris,330\n")
atoms = parse_file(p)
assert len(atoms) == 1
assert "name: eiffel" in atoms[0].statement
assert "city: paris" in atoms[0].statement
def test_parse_jsonl(tmp_path):
line = json.dumps({"name": "eiffel tower", "city": "paris", "height_m": 330})
p = _write(tmp_path, "d.jsonl", line + "\n")
atoms = parse_file(p)
assert len(atoms) == 1
assert "name: eiffel tower" in atoms[0].statement
assert "height_m: 330" in atoms[0].statement
def test_parse_json_list(tmp_path):
p = _write(tmp_path, "d.json", json.dumps([{"a": 1}, {"b": 2}]))
atoms = parse_file(p)
assert len(atoms) == 2
def test_flatten_record_nested():
pairs = flatten_record({"x": {"y": "z"}, "l": [1, 2]})
assert "x.y: z" in pairs
assert "l: 1; 2" in pairs
def test_parse_txt_paragraphs(tmp_path):
p = _write(tmp_path, "doc.txt", "Premier paragraphe sur paris.\n\nDeuxieme paragraphe.\n")
atoms = parse_file(p)
assert len(atoms) == 2
assert atoms[0].statement == "Premier paragraphe sur paris."
def test_parse_md_headings_become_context(tmp_path):
p = _write(tmp_path, "doc.md", "# Capitales\n\nparis est la capitale de la france\n")
atoms = parse_file(p)
assert atoms[0].context == "Capitales"
assert "paris" in atoms[0].statement
def test_parse_raw_fallback_reads_lines(tmp_path):
p = _write(tmp_path, "weird.xyz", "ligne une\nligne deux\n")
atoms = parse_file(p)
assert len(atoms) == 2
def test_parse_jsonl_skips_malformed_lines(tmp_path):
p = _write(tmp_path, "d.jsonl", '{"a": 1}\nnot json {\n{"b": 2}\n')
atoms = parse_file(p)
assert len(atoms) == 2
assert "a: 1" in atoms[0].statement
assert "b: 2" in atoms[1].statement
def test_parse_csv_latin1_does_not_crash(tmp_path):
p = tmp_path / "latin.csv"
p.write_bytes(b"question,answer\ncaf\xe9,paris\n")
atoms = parse_file(p)
assert len(atoms) == 1
assert atoms[0].context == "paris"
|