110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
import json
|
|
import pytest
|
|
from pathlib import Path
|
|
from chunks.chunker import run_pipeline, find_source
|
|
|
|
|
|
SAMPLE_MD = """# Sezione Principale
|
|
|
|
## Introduzione
|
|
|
|
Questo è il primo paragrafo della sezione introduttiva. Contiene testo sufficiente.
|
|
|
|
## Contenuto
|
|
|
|
Il secondo paragrafo parla di argomenti tecnici. Continua per alcune righe.
|
|
|
|
```python
|
|
def esempio():
|
|
return 42
|
|
```
|
|
|
|
## Indice
|
|
|
|
Questo contenuto deve essere saltato.
|
|
|
|
## Conclusioni
|
|
|
|
Paragrafo conclusivo del documento di test.
|
|
"""
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_stem(tmp_path):
|
|
stem = "test_doc"
|
|
source_dir = tmp_path / "sources" / f"{stem}_output" / "auto"
|
|
source_dir.mkdir(parents=True)
|
|
(source_dir / f"{stem}.md").write_text(SAMPLE_MD, encoding="utf-8")
|
|
(tmp_path / "chunks").mkdir()
|
|
return tmp_path, stem
|
|
|
|
|
|
def test_run_pipeline_produces_output_files(tmp_stem):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
out_dir = root / "chunks" / stem
|
|
assert (out_dir / "chunks.json").exists()
|
|
assert (out_dir / "meta.json").exists()
|
|
assert (out_dir / "report.json").exists()
|
|
|
|
|
|
def test_chunks_json_schema(tmp_stem):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
data = json.loads((root / "chunks" / stem / "chunks.json").read_text())
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
c = data[0]
|
|
for field in ("chunk_id", "content_original", "content_for_embedding",
|
|
"header_path", "block_ids", "flags", "neighbors", "assets"):
|
|
assert field in c, f"campo mancante: {field}"
|
|
|
|
|
|
def test_indice_section_skipped(tmp_stem):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
data = json.loads((root / "chunks" / stem / "chunks.json").read_text())
|
|
all_content = " ".join(c["content_original"] for c in data)
|
|
assert "saltato" not in all_content
|
|
|
|
|
|
def test_code_block_preserved(tmp_stem):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
data = json.loads((root / "chunks" / stem / "chunks.json").read_text())
|
|
all_content = " ".join(c["content_original"] for c in data)
|
|
assert "def esempio" in all_content
|
|
|
|
|
|
def test_force_flag_regenerates(tmp_stem):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
first = json.loads((root / "chunks" / stem / "chunks.json").read_text())
|
|
run_pipeline(stem=stem, root=root, force=True)
|
|
second = json.loads((root / "chunks" / stem / "chunks.json").read_text())
|
|
assert len(first) == len(second)
|
|
|
|
|
|
def test_no_force_skips_existing(tmp_stem, capsys):
|
|
root, stem = tmp_stem
|
|
run_pipeline(stem=stem, root=root)
|
|
run_pipeline(stem=stem, root=root, force=False)
|
|
captured = capsys.readouterr()
|
|
assert "skip" in captured.out.lower() or "esiste" in captured.out.lower()
|
|
|
|
|
|
def test_find_source_output_auto(tmp_path):
|
|
stem = "doc"
|
|
path = tmp_path / "sources" / f"{stem}_output" / "auto" / f"{stem}.md"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text("# T\n")
|
|
assert find_source(stem, tmp_path) == path
|
|
|
|
|
|
def test_find_source_flat(tmp_path):
|
|
stem = "doc"
|
|
path = tmp_path / "sources" / f"{stem}.md"
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text("# T\n")
|
|
assert find_source(stem, tmp_path) == path
|