Spaces:
Running
Running
| import json | |
| import textwrap | |
| from pathlib import Path | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from repograg.chunker import chunk_file | |
| from repograg.embedder import HashEmbedder | |
| from repograg.store import ChunkStore | |
| from repograg import server | |
| SAMPLE = textwrap.dedent(''' | |
| def expected_score(ra, rb): | |
| """Elo expected score for player A given both ratings.""" | |
| return 1 / (1 + 10 ** ((rb - ra) / 400)) | |
| def update_elo(ratings, a, b, result, k=32): | |
| """Update ratings dict in place after a match result.""" | |
| ea = expected_score(ratings.get(a, 1500), ratings.get(b, 1500)) | |
| ratings[a] = ratings.get(a, 1500) + k * (result - ea) | |
| ''') | |
| def client(tmp_path: Path): | |
| # build a tiny repo + index | |
| repo_root = tmp_path / "repo" | |
| repo_root.mkdir() | |
| src = repo_root / "elo.py" | |
| src.write_text(SAMPLE) | |
| emb = HashEmbedder(dim=64) | |
| chunks = chunk_file(src, repo_root=repo_root) | |
| store = ChunkStore(tmp_path / "demo.db", embedding_dim=64) | |
| store.add_chunks(chunks, emb.embed([c.embed_text for c in chunks])) | |
| store.close() | |
| (tmp_path / "showcase.json").write_text(json.dumps([{ | |
| "id": "demo", | |
| "name": "Demo", | |
| "description": "tiny fixture repo", | |
| "db": "demo.db", | |
| "root": str(repo_root), | |
| }])) | |
| fresh = server.AppState() | |
| fresh.load_manifest(tmp_path) | |
| fresh.embedder = emb # skip real model load | |
| fresh.answerer_checked = True # no LLM in tests | |
| old_state = server.state | |
| server.state = fresh | |
| yield TestClient(server.app) | |
| server.state = old_state | |
| def test_health_and_repos(client): | |
| assert client.get("/api/health").json()["status"] == "ok" | |
| repos = client.get("/api/repos").json() | |
| assert repos[0]["id"] == "demo" | |
| assert repos[0]["chunks"] > 0 | |
| assert repos[0]["browsable"] is True | |
| def test_query_returns_results_and_caches(client): | |
| body = {"repo": "demo", "query": "update_elo", "k": 3} | |
| first = client.post("/api/query", json=body).json() | |
| assert first["route"] == "symbol" | |
| assert first["cached"] is False | |
| assert any(r["name"] == "update_elo" for r in first["results"]) | |
| r = first["results"][0] | |
| assert {"file_path", "start_line", "end_line", "snippet", "sources"} <= set(r) | |
| second = client.post("/api/query", json=body).json() | |
| assert second["cached"] is True | |
| assert second["results"] == first["results"] | |
| def test_query_unknown_repo_404(client): | |
| resp = client.post("/api/query", json={"repo": "nope", "query": "x"}) | |
| assert resp.status_code == 404 | |
| def test_files_and_file_content(client): | |
| files = client.get("/api/files", params={"repo": "demo"}).json()["files"] | |
| assert "elo.py" in files | |
| body = client.get("/api/file", params={"repo": "demo", "path": "elo.py"}).json() | |
| assert "def update_elo" in body["content"] | |
| def test_file_path_traversal_blocked(client): | |
| resp = client.get("/api/file", params={"repo": "demo", "path": "../secrets.txt"}) | |
| assert resp.status_code == 400 | |
| def test_file_missing_404(client): | |
| resp = client.get("/api/file", params={"repo": "demo", "path": "nope.py"}) | |
| assert resp.status_code == 404 | |