File size: 3,453 Bytes
ed65693
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
def test_health(client):
    response = client.get("/health")

    assert response.status_code == 200
    assert response.json()["status"] == "ok"


def test_upload_and_query_text_document(client):
    from app.main import store

    def fake_add_file(file_path):
        store.documents.add(file_path.name)
        return {"filename": file_path.name, "chunks": 1, "characters": 84}

    def fake_answer(question, top_k, mode):
        return {
            "answer": "[notes.txt] Semantic search finds similar text chunks.",
            "sources": [
                {
                    "source": "notes.txt",
                    "chunk": 1,
                    "score": 0.91,
                    "text": "Semantic search finds similar text chunks.",
                }
            ],
            "retrieval_latency_ms": 1.23,
            "telemetry": {"mode": mode},
        }

    store.add_file = fake_add_file
    store.answer = fake_answer

    upload = client.post(
        "/upload",
        files={
            "file": (
                "notes.txt",
                "Python is used with FastAPI to build APIs. Semantic search finds similar text chunks.",
                "text/plain",
            )
        },
    )

    assert upload.status_code == 200
    assert upload.json()["chunks"] >= 1

    query = client.post(
        "/query",
        json={"question": "What is semantic search?", "top_k": 2, "mode": "hybrid_fixed"},
    )

    assert query.status_code == 200
    data = query.json()
    assert "Semantic search" in data["answer"]
    assert data["sources"][0]["source"] == "notes.txt"
    assert "retrieval_latency_ms" in data
    assert data["telemetry"]["mode"] == "hybrid_fixed"


def test_modes_endpoint_lists_all_modes(client):
    response = client.get("/modes")

    assert response.status_code == 200
    data = response.json()
    assert data["default"] == "hybrid_calibrated_rerank"
    assert set(data["modes"]) == {
        "dense",
        "sparse",
        "rrf",
        "hybrid_fixed",
        "hybrid_calibrated",
        "hybrid_fixed_rerank",
        "hybrid_calibrated_rerank",
    }


def test_query_rejects_unknown_mode(client):
    response = client.post(
        "/query",
        json={"question": "What is AI?", "top_k": 3, "mode": "unknown"},
    )

    assert response.status_code == 400
    assert "unknown mode" in response.json()["detail"].lower()


def test_upload_rejects_unsupported_file(client):
    response = client.post(
        "/upload",
        files={"file": ("data.csv", "col1,col2\na,b", "text/csv")},
    )
    assert response.status_code == 400
    assert "supported" in response.json()["detail"].lower()


def test_query_empty_index(client):
    response = client.post("/query", json={"question": "What is AI?", "top_k": 3})
    assert response.status_code == 200
    data = response.json()
    assert data["sources"] == []


def test_query_validation(client):
    # Question too short (min_length=3)
    response = client.post("/query", json={"question": "hi", "top_k": 1})
    assert response.status_code == 422

    # top_k out of range
    response = client.post("/query", json={"question": "What is AI?", "top_k": 99})
    assert response.status_code == 422


def test_list_documents_empty(client):
    response = client.get("/documents")
    assert response.status_code == 200
    data = response.json()
    assert data["documents"] == []
    assert data["index_ready"] is False