Spaces:
Sleeping
Sleeping
File size: 6,554 Bytes
22da335 | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """Tests for ragqa.cli.
The CLI wires modules together. Tests use a real chunker (with the
`make_pdf` fixture) but mock the embedder (no torch load) and LLM
(no HTTP). That keeps each test < 100ms but exercises the actual
file paths and command parsing.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from unittest.mock import patch, MagicMock
import numpy as np
import pytest
from ragqa.cli import main
# ───────────────────────── fakes ───────────────────────────────────────────
def _patched_embedder(dim: int = 4):
"""Patch ragqa.embedding.SentenceTransformer so Embedder doesn't load
PyTorch + the 80MB model in tests."""
mock_st = MagicMock()
mock_st.get_sentence_embedding_dimension.return_value = dim
def encode_side_effect(texts, normalize_embeddings=True, **_):
rng = np.random.RandomState(0)
arr = rng.randn(len(texts), dim).astype(np.float32)
if normalize_embeddings:
arr /= np.linalg.norm(arr, axis=1, keepdims=True)
return arr
mock_st.encode.side_effect = encode_side_effect
return mock_st
# ───────────────────────── ingest ──────────────────────────────────────────
def test_ingest_creates_index_from_single_pdf(make_pdf, tmp_path, capsys):
pdf = make_pdf(["The cat sat on the mat. " * 20])
out_dir = tmp_path / "myindex"
with patch("ragqa.embedding.SentenceTransformer",
return_value=_patched_embedder()):
rc = main(["ingest", "-i", str(pdf), "-o", str(out_dir)])
assert rc == 0
assert (out_dir / "index.faiss").is_file()
assert (out_dir / "chunks.json").is_file()
assert (out_dir / "meta.json").is_file()
meta = json.loads((out_dir / "meta.json").read_text())
assert meta["n_chunks"] >= 1
def test_ingest_creates_index_from_directory(make_pdf, tmp_path):
pdf_dir = tmp_path / "pdfs"
pdf_dir.mkdir()
# Put two PDFs in a directory; the ingester should pick up both.
p1 = make_pdf(["Doc one content. " * 20])
p2 = make_pdf(["Doc two content. " * 20])
(pdf_dir / "a.pdf").write_bytes(p1.read_bytes())
(pdf_dir / "b.pdf").write_bytes(p2.read_bytes())
out_dir = tmp_path / "idx"
with patch("ragqa.embedding.SentenceTransformer",
return_value=_patched_embedder()):
rc = main(["ingest", "-i", str(pdf_dir), "-o", str(out_dir)])
assert rc == 0
chunks = json.loads((out_dir / "chunks.json").read_text())
sources = {c["source_file"] for c in chunks}
assert sources == {"a.pdf", "b.pdf"}
def test_ingest_with_no_pdfs_exits_error(tmp_path, capsys):
empty_dir = tmp_path / "empty"
empty_dir.mkdir()
out_dir = tmp_path / "idx"
rc = main(["ingest", "-i", str(empty_dir), "-o", str(out_dir)])
assert rc != 0
err = capsys.readouterr().err.lower()
assert "no pdf" in err or "not found" in err
# ───────────────────────── ask ─────────────────────────────────────────────
def _build_index_for_ask(make_pdf, tmp_path):
"""Helper: ingest a tiny PDF so the ask tests have something to load."""
pdf = make_pdf(["The capital of France is Paris. " * 5])
out = tmp_path / "idx"
with patch("ragqa.embedding.SentenceTransformer",
return_value=_patched_embedder()):
rc = main(["ingest", "-i", str(pdf), "-o", str(out)])
assert rc == 0
return out
def test_ask_without_api_key_exits_error(make_pdf, tmp_path, capsys, monkeypatch):
monkeypatch.delenv("GROQ_API_KEY", raising=False)
idx = _build_index_for_ask(make_pdf, tmp_path)
rc = main(["ask", "--index", str(idx), "What is the capital?"])
assert rc != 0
err = capsys.readouterr().err.lower()
assert "groq_api_key" in err
def test_ask_loads_index_and_prints_answer(make_pdf, tmp_path, capsys, monkeypatch):
idx = _build_index_for_ask(make_pdf, tmp_path)
monkeypatch.setenv("GROQ_API_KEY", "fake")
# Mock both the embedder (for query encode) and the LLM HTTP layer.
with patch("ragqa.embedding.SentenceTransformer",
return_value=_patched_embedder()):
mock_resp = MagicMock(status_code=200)
mock_resp.json.return_value = {
"choices": [{"message": {"content": "Paris [1]."}}]
}
with patch("ragqa.llm.requests.post", return_value=mock_resp):
# Force min_score = 0 so the random-vector fake retrieval
# actually returns chunks.
rc = main([
"ask", "--index", str(idx), "--min-score", "0",
"What is the capital?",
])
assert rc == 0
out = capsys.readouterr().out
assert "Paris" in out
def test_ask_prints_sources_section_when_citations_present(
make_pdf, tmp_path, capsys, monkeypatch,
):
idx = _build_index_for_ask(make_pdf, tmp_path)
monkeypatch.setenv("GROQ_API_KEY", "fake")
with patch("ragqa.embedding.SentenceTransformer",
return_value=_patched_embedder()):
mock_resp = MagicMock(status_code=200)
mock_resp.json.return_value = {
"choices": [{"message": {"content": "Paris [1]."}}]
}
with patch("ragqa.llm.requests.post", return_value=mock_resp):
main([
"ask", "--index", str(idx), "--min-score", "0",
"What is the capital?",
])
out = capsys.readouterr().out
# Citation block should reference the source PDF and a page number.
assert "Sources" in out or "sources" in out
assert "page" in out.lower()
# ───────────────────────── help + parser ───────────────────────────────────
def test_help_exits_zero(capsys):
"""`ragqa --help` should exit 0 with usage info."""
with pytest.raises(SystemExit) as exc:
main(["--help"])
assert exc.value.code == 0
def test_no_subcommand_exits_nonzero(capsys):
"""Calling `ragqa` with no subcommand should error, not silently do nothing."""
with pytest.raises(SystemExit) as exc:
main([])
assert exc.value.code != 0
|