138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
"""Integration tests: GET /api/documents/content — the viewer's data source.
|
|
|
|
Uses the real compose Postgres (``db`` fixture) and FastAPI's TestClient:
|
|
* 200 with the full field set for a seeded document (all formats);
|
|
* 404 for an unknown (source, path) pair;
|
|
* 404 for traversal-style ``path`` values (no filesystem access → no leak).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.models import Chunk, Document
|
|
|
|
|
|
def _seed_doc(
|
|
db,
|
|
source: str = "Homelab",
|
|
path: str = "kubernetes.md",
|
|
title: str = "Kubernetes Homelab Cluster",
|
|
content: str = "# Kubernetes\n\nTalos on 3 nodes.",
|
|
chunks: int = 2,
|
|
) -> None:
|
|
"""Truncate the KB and insert one document with ``chunks`` chunk rows."""
|
|
db.execute(text("TRUNCATE chunks, documents"))
|
|
db.commit()
|
|
doc = Document(
|
|
source=source,
|
|
path=path,
|
|
full_path=f"/tmp/{path}",
|
|
title=title,
|
|
content=content,
|
|
content_hash="a" * 64,
|
|
indexed_at=datetime.now(UTC),
|
|
)
|
|
db.add(doc)
|
|
db.flush()
|
|
if chunks:
|
|
db.add_all(
|
|
Chunk(document_id=doc.id, position=i, content=f"chunk {i}", embedding=[0.01] * 768)
|
|
for i in range(chunks)
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def test_content_200_all_fields(client, db) -> None:
|
|
_seed_doc(db)
|
|
try:
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Homelab", "path": "kubernetes.md"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert set(body) == {"source", "path", "title", "format", "content", "indexed_at", "chunks"}
|
|
assert body["source"] == "Homelab"
|
|
assert body["path"] == "kubernetes.md"
|
|
assert body["title"] == "Kubernetes Homelab Cluster"
|
|
assert body["format"] == "md"
|
|
assert body["content"] == "# Kubernetes\n\nTalos on 3 nodes."
|
|
assert body["chunks"] == 2
|
|
datetime.fromisoformat(body["indexed_at"]) # raises if not ISO-8601
|
|
finally:
|
|
db.execute(text("TRUNCATE chunks, documents"))
|
|
db.commit()
|
|
|
|
|
|
def test_content_404_unknown_path(client, db) -> None:
|
|
_seed_doc(db)
|
|
try:
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Homelab", "path": "nope/missing.md"},
|
|
)
|
|
assert r.status_code == 404
|
|
assert r.json() == {"detail": "document not found"}
|
|
# A pair that exists under a DIFFERENT source is also 404 — both
|
|
# values must match the row.
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Deployments", "path": "kubernetes.md"},
|
|
)
|
|
assert r.status_code == 404
|
|
finally:
|
|
db.execute(text("TRUNCATE chunks, documents"))
|
|
db.commit()
|
|
|
|
|
|
def test_content_404_traversal_style_path_no_leak(client, db) -> None:
|
|
"""DB-only lookup: traversal strings are just non-existent rows — 404,
|
|
and the response must not carry anything from the filesystem."""
|
|
_seed_doc(db)
|
|
try:
|
|
for path in ("../../etc/passwd", "../kubernetes.md", "..%2F..%2Fetc%2Fpasswd"):
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Homelab", "path": path},
|
|
)
|
|
assert r.status_code == 404, path
|
|
assert r.json() == {"detail": "document not found"}, path
|
|
assert "root:" not in r.text, path
|
|
finally:
|
|
db.execute(text("TRUNCATE chunks, documents"))
|
|
db.commit()
|
|
|
|
|
|
def test_content_format_from_suffix(client, db) -> None:
|
|
"""format = lowercased path suffix: yaml documents (phase 09 corpus) and
|
|
the no-suffix fallback both flow through the same endpoint."""
|
|
try:
|
|
_seed_doc(
|
|
db,
|
|
path="container_gitlab/gitlab-compose.yaml",
|
|
title="gitlab-compose",
|
|
content="services:\n gitlab:\n image: gitlab/gitlab-ce",
|
|
chunks=0,
|
|
)
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Homelab", "path": "container_gitlab/gitlab-compose.yaml"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["format"] == "yaml"
|
|
assert body["chunks"] == 0 # outerjoin → zero, not missing
|
|
|
|
_seed_doc(db, path="README", title="README", content="plain text, no suffix", chunks=0)
|
|
r = client.get(
|
|
"/api/documents/content",
|
|
params={"source": "Homelab", "path": "README"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["format"] == "text" # no-suffix fallback
|
|
finally:
|
|
db.execute(text("TRUNCATE chunks, documents"))
|
|
db.commit()
|