phase: 97_kb_tree_catalog
All completion criteria verified — everything is green, no defects found. Final report: ## Phase 97 final verification pass — ALL GREEN **Verified (no code changes needed):** - `GET /api/docs/tree` (admin), `build_kb_tree` pure builder, `PATCH /api/folders/summary`, migration 0018 (`manually_edited`, head confirmed), generator skip/keep + `kept_manual` stat, RAG tree UI + edit affordance in `sources.js`/`index.html`/`styles.css` - `tests/e2e/test_kb_tree.py`: 8 passed — top level, drill source/folder, edit round-trip, clear, manual-desc-survives-sync, reload fallback, anonymous gate - Integration: tree shape/order/403/empty/indexed-only + PATCH update/create/root/clear/404/403/no-LLM + stat-walk equivalence (in `test_docs_api.py`); 3-field `folder_summaries=` import token preserved **Gates (exact commands):** - `uv run pytest --cov=app --cov-report=term-missing` → **2053 passed**, TOTAL coverage **99%** (>90% ✓) - `uv run ruff check . && uv run pyright` → **All checks passed / 0 errors** - `uv run pytest tests/e2e/test_kb_tree.py -v --no-cov` → **8 passed** in isolation - 30 story/RAG-view E2E suites run **one per process**: all passed, incl. `test_ls_tree_drilldown` (agent `ls` byte-identical ✓), `test_import_documents`, `test_edit_summaries`, `test_admin_auth`, `test_kb_overview` **Completion criteria:** tree view ✓ · edit round-trip + clear ✓ · manual persists/clear resets ✓ · `ls` unchanged ✓ · pytest/coverage/lint ✓ · E2E isolation ✓ · commit — left to harness per protocol (working tree untouched, `git add/commit` not run) **Deviations:** none. **Next pending phase:** none — `todo/` contains only 97 (96 already committed).
This commit is contained in:
@@ -1,15 +1,68 @@
|
||||
"""Integration tests: GET /api/docs — empty shape + populated shape.
|
||||
"""Integration tests: GET /api/docs — empty shape + populated shape —
|
||||
and GET /api/docs/tree (phase 97, task 02 — the full recursive KB tree
|
||||
the RAG view renders: shape, ordering, counts, summaries, the 403 gate,
|
||||
the superset rule, and the stat-walk equivalence with ``GET /api/docs``).
|
||||
|
||||
Uses the real compose Postgres (``db`` fixture) and FastAPI's TestClient.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import itertools
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from sqlalchemy import text
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import func, select, text
|
||||
|
||||
from app.models import Chunk, Document
|
||||
import app.api.docs as docs_api
|
||||
from app.config import Settings
|
||||
from app.core import tokens as token_service
|
||||
from app.main import app as fastapi_app
|
||||
from app.models import Chunk, Document, FolderSummary, GitSource
|
||||
from app.rag import git_sources as rag_git_sources
|
||||
|
||||
_TREE_TABLES = "chunks, documents, folder_summaries, git_sources"
|
||||
|
||||
|
||||
def _truncate_tree_tables(db) -> None:
|
||||
db.execute(text(f"TRUNCATE {_TREE_TABLES}"))
|
||||
db.commit()
|
||||
|
||||
|
||||
_counter = itertools.count()
|
||||
|
||||
|
||||
def _seed_doc(db, source: str, path: str, title: str, n_chunks: int, indexed_at: datetime) -> None:
|
||||
"""One indexed document with *n_chunks* content chunks (unique hash)."""
|
||||
doc = Document(
|
||||
source=source,
|
||||
path=path,
|
||||
full_path=f"/tmp/{source}/{path}",
|
||||
title=title,
|
||||
content=f"# {title}\n\nBody.",
|
||||
content_hash=f"{next(_counter):064d}",
|
||||
indexed_at=indexed_at,
|
||||
)
|
||||
db.add(doc)
|
||||
db.flush()
|
||||
db.add_all(
|
||||
Chunk(document_id=doc.id, position=i, content=f"chunk {i}", embedding=[0.01] * 768)
|
||||
for i in range(n_chunks)
|
||||
)
|
||||
|
||||
|
||||
def _tree_file_nodes(sources) -> list[dict]:
|
||||
"""Every file node of a tree response, walked recursively."""
|
||||
files: list[dict] = []
|
||||
for source in sources:
|
||||
for child in source["children"]:
|
||||
if child["kind"] == "file":
|
||||
files.append(child)
|
||||
else:
|
||||
files.extend(_tree_file_nodes([child]))
|
||||
return files
|
||||
|
||||
|
||||
def test_docs_empty_shape(admin_client, db) -> None:
|
||||
@@ -77,3 +130,521 @@ def test_docs_response_matches_schema_shape(admin_client, db) -> None:
|
||||
for d in body["documents"]:
|
||||
assert set(d) == {"id", "source", "path", "title", "chunks", "indexed_at"}
|
||||
assert isinstance(d["chunks"], int) and d["chunks"] >= 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# GET /api/docs/tree (phase 97, task 02).
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_docs_tree_populated_shape_order_counts_summaries(admin_client, db) -> None:
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
# Registry order (added_at) is Homelab → Deployments — deliberately
|
||||
# NOT alphabetical (the registry order leads, the phase-97 rule).
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
db.add(
|
||||
GitSource(
|
||||
url="https://github.com/reese/Deployments.git",
|
||||
kind="git",
|
||||
added_at=base + timedelta(hours=1),
|
||||
)
|
||||
)
|
||||
_seed_doc(db, "Homelab", "k8s/talos.md", "Talos", 3, base)
|
||||
_seed_doc(db, "Homelab", "k8s/cluster.md", "Cluster", 2, base)
|
||||
_seed_doc(db, "Homelab", "k8s/helm/charts.md", "Charts", 1, base)
|
||||
_seed_doc(db, "Homelab", "root-note.md", "Root note", 4, base)
|
||||
_seed_doc(db, "Deployments", "deploy-a.md", "Deploy A", 1, base)
|
||||
_seed_doc(db, "Deployments", "deploy-b.md", "Deploy B", 0, base)
|
||||
db.add_all(
|
||||
[
|
||||
FolderSummary(source="Homelab", folder_path="", summary="Homelab docs."),
|
||||
FolderSummary(source="Homelab", folder_path="k8s", summary="K8s stuff."),
|
||||
# A stored row for a folder with NO indexed descendants —
|
||||
# e.g. a manual row surviving the prune below the 2-doc
|
||||
# minimum (phase 97, task 01): the tree's folders come from
|
||||
# INDEXED paths (the existence rule), not from summary rows.
|
||||
FolderSummary(source="Deployments", folder_path="orphan", summary="Ghost."),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
|
||||
r = admin_client.get("/api/docs/tree")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert set(body) == {"sources"}
|
||||
sources = body["sources"]
|
||||
assert [s["name"] for s in sources] == ["Homelab", "Deployments"]
|
||||
|
||||
homelab, deployments = sources
|
||||
assert set(homelab) == {"name", "documents", "summary", "children"}
|
||||
assert homelab["documents"] == 4 # the whole recursive count
|
||||
assert homelab["summary"] == "Homelab docs." # the (source, "") row
|
||||
assert deployments["summary"] is None # no stored root row
|
||||
assert deployments["documents"] == 2
|
||||
|
||||
# Homelab: subfolder first (path order among folders), then the
|
||||
# root files; the nested folder k8s/helm recurses one level deeper.
|
||||
k8s, root_note = homelab["children"]
|
||||
assert k8s["kind"] == "folder"
|
||||
assert k8s["path"] == "k8s"
|
||||
assert k8s["documents"] == 3 # talos + cluster + charts (subtree)
|
||||
assert k8s["summary"] == "K8s stuff."
|
||||
helm, cluster, talos = k8s["children"]
|
||||
assert (helm["kind"], helm["path"], helm["documents"], helm["summary"]) == (
|
||||
"folder",
|
||||
"k8s/helm",
|
||||
1,
|
||||
None,
|
||||
)
|
||||
assert [
|
||||
(c["kind"], c["path"], c["title"], c["chunks"])
|
||||
for c in (cluster, talos)
|
||||
] == [("file", "k8s/cluster.md", "Cluster", 2), ("file", "k8s/talos.md", "Talos", 3)]
|
||||
chart = helm["children"][0]
|
||||
assert (chart["kind"], chart["path"], chart["title"], chart["chunks"]) == (
|
||||
"file",
|
||||
"k8s/helm/charts.md",
|
||||
"Charts",
|
||||
1,
|
||||
)
|
||||
datetime.fromisoformat(chart["indexed_at"]) # valid ISO-8601
|
||||
assert (root_note["kind"], root_note["path"], root_note["title"], root_note["chunks"]) == (
|
||||
"file",
|
||||
"root-note.md",
|
||||
"Root note",
|
||||
4,
|
||||
)
|
||||
|
||||
# Deployments: flat — direct files in catalog order, and the
|
||||
# ``orphan`` summary row does NOT create a folder node.
|
||||
assert [(c["kind"], c["path"]) for c in deployments["children"]] == [
|
||||
("file", "deploy-a.md"),
|
||||
("file", "deploy-b.md"),
|
||||
]
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_docs_tree_403_anonymous(client, db) -> None:
|
||||
"""Admin-only, like ``GET /api/docs``: anonymous → 403
|
||||
``admin only`` (the RAG view's anonymous gate never fetches the
|
||||
tree)."""
|
||||
_truncate_tree_tables(db)
|
||||
r = client.get("/api/docs/tree")
|
||||
assert r.status_code == 403
|
||||
assert r.json() == {"detail": "admin only"}
|
||||
|
||||
|
||||
def test_docs_tree_empty_registry_and_catalog(
|
||||
admin_client, db, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Nothing registered, nothing indexed → ``{"sources": []}``.
|
||||
|
||||
"Empty registry" means the ``git_sources`` table AND the
|
||||
``BOR_GIT_SOURCES`` env fallback are both empty — the dev ``.env``
|
||||
names a source, so the resolver's ``get_settings`` is patched with
|
||||
a fresh ``Settings(_env_file=None, git_sources="")`` (the
|
||||
``test_sync_api`` / ``test_git_sources_api`` pattern: the dev
|
||||
``.env`` never leaks in)."""
|
||||
monkeypatch.setattr(
|
||||
rag_git_sources,
|
||||
"get_settings",
|
||||
lambda: Settings(_env_file=None, git_sources=""), # pyright: ignore[reportCallIssue]
|
||||
)
|
||||
_truncate_tree_tables(db)
|
||||
r = admin_client.get("/api/docs/tree")
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"sources": []}
|
||||
|
||||
|
||||
def test_docs_tree_indexed_only_source_after_registered(admin_client, db) -> None:
|
||||
"""The superset rule: a registered 0-document source still lists
|
||||
(first — the registry leads), and indexed-only sources (documents
|
||||
whose source is not in ``git_sources`` — ad-hoc imports, removed
|
||||
but not-yet-pruned sources) trail in alphabetical order."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Alpha.git", kind="git", added_at=base))
|
||||
_seed_doc(db, "Zeta", "z1.md", "Z1", 1, base)
|
||||
_seed_doc(db, "Midx", "m1.md", "M1", 2, base)
|
||||
db.commit()
|
||||
|
||||
r = admin_client.get("/api/docs/tree")
|
||||
assert r.status_code == 200
|
||||
sources = r.json()["sources"]
|
||||
assert [s["name"] for s in sources] == ["Alpha", "Midx", "Zeta"]
|
||||
alpha, midx, zeta = sources
|
||||
assert (alpha["documents"], alpha["children"], alpha["summary"]) == (0, [], None)
|
||||
assert midx["documents"] == 1
|
||||
assert zeta["documents"] == 1
|
||||
assert [c["path"] for c in midx["children"]] == ["m1.md"]
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# PATCH /api/folders/summary (phase 97, task 03) — the admin
|
||||
# folder-description editor: update / create / source-root / clear /
|
||||
# double-clear, the 404s, the 403s, and the round-trip through
|
||||
# ``GET /api/docs/tree`` (task 02).
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
#: A deliberately old stamp: a save must ADVANCE ``updated_at`` past it.
|
||||
OLDER_STAMP = datetime(2020, 1, 1, tzinfo=UTC)
|
||||
|
||||
|
||||
def _seed_folder_pair(db, source: str, folder: str, base: datetime) -> None:
|
||||
"""Two documents under ``source/folder`` — the ≥ 2-document
|
||||
minimum a folder must hold to be summarizable."""
|
||||
_seed_doc(db, source, f"{folder}/one.md", f"{folder} one", 1, base)
|
||||
_seed_doc(db, source, f"{folder}/two.md", f"{folder} two", 2, base)
|
||||
|
||||
|
||||
def _get_folder_row(db, source: str, folder: str) -> FolderSummary | None:
|
||||
return db.scalar(
|
||||
select(FolderSummary).where(
|
||||
FolderSummary.source == source, FolderSummary.folder_path == folder
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def test_folder_summary_update_ai_row_flips_manual_and_advances_stamp(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""Update an AI-written row: the text is replaced, ``manually_edited``
|
||||
flips to true, ``updated_at`` advances past the seeded stamp — and a
|
||||
second save with different text updates IN PLACE (same PK, one
|
||||
row). The round-trip: ``GET /api/docs/tree`` shows the new text on
|
||||
the folder node."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_folder_pair(db, "Homelab", "k8s", base)
|
||||
db.add(
|
||||
FolderSummary(
|
||||
source="Homelab",
|
||||
folder_path="k8s",
|
||||
summary="AI text.",
|
||||
manually_edited=False,
|
||||
updated_at=OLDER_STAMP,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
db.expire_all()
|
||||
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "k8s", "summary": " Owner words. "},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert set(r.json()) == {"source", "folder_path", "summary"}
|
||||
assert r.json() == {"source": "Homelab", "folder_path": "k8s", "summary": "Owner words."}
|
||||
|
||||
db.expire_all()
|
||||
row = _get_folder_row(db, "Homelab", "k8s")
|
||||
assert row is not None
|
||||
assert row.summary == "Owner words." # stripped before storing
|
||||
assert row.manually_edited is True # a manual row from this save on
|
||||
assert row.updated_at > OLDER_STAMP # fresh stamp, not the seeded one
|
||||
|
||||
# The task-02 tree shows the owner's text on the folder node.
|
||||
tree = admin_client.get("/api/docs/tree").json()
|
||||
k8s = next(c for c in tree["sources"][0]["children"] if c["kind"] == "folder")
|
||||
assert (k8s["path"], k8s["summary"]) == ("k8s", "Owner words.")
|
||||
|
||||
# A second save updates in place — same row, no second PK.
|
||||
r2 = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "k8s", "summary": "Owner words v2."},
|
||||
)
|
||||
assert r2.status_code == 200, r2.text
|
||||
assert r2.json()["summary"] == "Owner words v2."
|
||||
db.expire_all()
|
||||
assert db.scalar(select(func.count()).select_from(FolderSummary)) == 1 # one row
|
||||
row = _get_folder_row(db, "Homelab", "k8s")
|
||||
assert row is not None
|
||||
assert row.summary == "Owner words v2."
|
||||
assert row.manually_edited is True
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_create_where_no_row_exists(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""A manual description can be CREATED where no row exists — a
|
||||
< 2-document folder the generator never wrote (or its fail-soft
|
||||
miss): insert, not update, with ``manually_edited = true``."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_doc(db, "Homelab", "solo/only.md", "Only", 1, base) # 1-doc folder
|
||||
db.commit()
|
||||
db.expire_all()
|
||||
assert _get_folder_row(db, "Homelab", "solo") is None # no AI row for < 2 docs
|
||||
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "solo", "summary": "One-off scripts."},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {
|
||||
"source": "Homelab",
|
||||
"folder_path": "solo",
|
||||
"summary": "One-off scripts.",
|
||||
}
|
||||
db.expire_all()
|
||||
row = _get_folder_row(db, "Homelab", "solo")
|
||||
assert row is not None
|
||||
assert row.summary == "One-off scripts."
|
||||
assert row.manually_edited is True
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_root_save_on_registered_zero_doc_source(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""``folder_path = ""`` is valid for ANY allowed source — a registered
|
||||
0-document source (nothing indexed yet) still accepts a root
|
||||
description: the source check is registered-OR-indexed."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Empty.git", kind="git", added_at=base))
|
||||
db.commit()
|
||||
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Empty", "folder_path": "", "summary": "Docs incoming."},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {"source": "Empty", "folder_path": "", "summary": "Docs incoming."}
|
||||
db.expire_all()
|
||||
row = _get_folder_row(db, "Empty", "")
|
||||
assert row is not None
|
||||
assert row.manually_edited is True
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_source_root_round_trips_through_tree(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""The source root (``folder_path: ""``) round-trips through the
|
||||
task-02 tree endpoint: save → the source node carries the text;
|
||||
clear → it is null again."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_doc(db, "Homelab", "a/b.md", "B", 1, base)
|
||||
_seed_doc(db, "Homelab", "c.md", "C", 1, base)
|
||||
db.commit()
|
||||
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "", "summary": "All the homelab docs."},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {
|
||||
"source": "Homelab",
|
||||
"folder_path": "",
|
||||
"summary": "All the homelab docs.",
|
||||
}
|
||||
tree = admin_client.get("/api/docs/tree").json()
|
||||
assert tree["sources"][0]["summary"] == "All the homelab docs."
|
||||
db.expire_all()
|
||||
assert _get_folder_row(db, "Homelab", "") is not None
|
||||
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "", "summary": " "},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {"source": "Homelab", "folder_path": "", "summary": None}
|
||||
tree = admin_client.get("/api/docs/tree").json()
|
||||
assert tree["sources"][0]["summary"] is None
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_clear_deletes_row_and_double_clear_is_noop(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""Empty/whitespace clears: the row is deleted — AI-written OR
|
||||
manual, either way it is gone (the next KB-changing sync regenerates
|
||||
an AI row: the reset path) — and the response ``summary`` is null.
|
||||
A second clear with no row is a 200 no-op. The task-02 tree shows
|
||||
null on both folder nodes after the clears."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_folder_pair(db, "Homelab", "k8s", base)
|
||||
_seed_folder_pair(db, "Homelab", "manual", base)
|
||||
db.add(
|
||||
FolderSummary(
|
||||
source="Homelab",
|
||||
folder_path="k8s",
|
||||
summary="AI text.",
|
||||
manually_edited=False,
|
||||
updated_at=OLDER_STAMP,
|
||||
)
|
||||
)
|
||||
db.add(
|
||||
FolderSummary(
|
||||
source="Homelab",
|
||||
folder_path="manual",
|
||||
summary="Owner text.",
|
||||
manually_edited=True,
|
||||
updated_at=OLDER_STAMP,
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
db.expire_all()
|
||||
|
||||
for folder, blank in (("k8s", " "), ("manual", "")): # whitespace, then empty
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": folder, "summary": blank},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {"source": "Homelab", "folder_path": folder, "summary": None}
|
||||
db.expire_all()
|
||||
assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0 # both rows gone
|
||||
|
||||
# A second clear (no row) is a 200 no-op.
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": "Homelab", "folder_path": "k8s", "summary": ""},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {"source": "Homelab", "folder_path": "k8s", "summary": None}
|
||||
db.expire_all()
|
||||
assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0 # still nothing
|
||||
|
||||
# The task-02 tree shows null on both folder nodes after the clears.
|
||||
tree = admin_client.get("/api/docs/tree").json()
|
||||
nodes = {
|
||||
c["path"]: c for c in tree["sources"][0]["children"] if c["kind"] == "folder"
|
||||
}
|
||||
assert nodes["k8s"]["summary"] is None
|
||||
assert nodes["manual"]["summary"] is None
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_404_unknown_source_folder_and_traversal(
|
||||
admin_client: TestClient, db
|
||||
) -> None:
|
||||
"""404s, in check order: an unknown source (neither registered nor
|
||||
indexed) wins over the folder check — including for a folder that
|
||||
EXISTS under another source; an unknown folder, a traversal folder
|
||||
path, and a file merely sharing a folder's name (no indexed
|
||||
descendant) are all ``folder not found``. Nothing is written."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_folder_pair(db, "Homelab", "real", base)
|
||||
_seed_doc(db, "Homelab", "note", "A file named like a folder", 1, base)
|
||||
db.commit()
|
||||
|
||||
cases = (
|
||||
# (source, folder_path, detail)
|
||||
("Ghost", "", "source not found"), # unknown source, root
|
||||
("Ghost", "real", "source not found"), # folder exists ELSEWHERE — source wins
|
||||
("Homelab", "nope", "folder not found"), # unknown folder
|
||||
("Homelab", "../../etc", "folder not found"), # traversal: no prefix match
|
||||
("Homelab", "note", "folder not found"), # file named like a folder, no descendants
|
||||
("Homelab", "real/", "folder not found"), # trailing slash: strict prefix rule
|
||||
)
|
||||
for source, folder, detail in cases:
|
||||
r = admin_client.patch(
|
||||
"/api/folders/summary",
|
||||
json={"source": source, "folder_path": folder, "summary": "whatever"},
|
||||
)
|
||||
assert r.status_code == 404, (source, folder, r.status_code)
|
||||
assert r.json() == {"detail": detail}, (source, folder)
|
||||
|
||||
db.expire_all()
|
||||
assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0 # nothing written
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_403_anonymous_and_token_user(
|
||||
client: TestClient, db
|
||||
) -> None:
|
||||
"""The ``require_admin`` gate: an anonymous caller AND a live
|
||||
access-token user who is not the admin (phase 79 token users exist)
|
||||
both get 403 ``admin only`` — the endpoint gate is the API-level
|
||||
defense in depth, not the RAG view's render gate."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_folder_pair(db, "Homelab", "k8s", base)
|
||||
db.execute(text("TRUNCATE api_tokens"))
|
||||
db.commit()
|
||||
body = {"source": "Homelab", "folder_path": "k8s", "summary": "x"}
|
||||
try:
|
||||
# Anonymous (the shared ``client`` is unsigned in this module).
|
||||
r = client.patch("/api/folders/summary", json=body)
|
||||
assert r.status_code == 403
|
||||
assert r.json() == {"detail": "admin only"}
|
||||
|
||||
# A live access-token user who is not the admin (phase 79):
|
||||
# signed in via the public token login, the row is active.
|
||||
_row, plaintext = token_service.create_token(db, "pin-holder")
|
||||
db.commit()
|
||||
holder = TestClient(fastapi_app)
|
||||
s = holder.post("/api/token-auth", json={"token": plaintext})
|
||||
assert s.status_code == 204, s.text
|
||||
r = holder.patch("/api/folders/summary", json=body)
|
||||
assert r.status_code == 403
|
||||
assert r.json() == {"detail": "admin only"}
|
||||
finally:
|
||||
db.execute(text("TRUNCATE api_tokens"))
|
||||
db.commit()
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
|
||||
def test_folder_summary_patch_never_constructs_an_llm_client() -> None:
|
||||
"""Source pin (the house pattern): a folder description is NEVER
|
||||
embedded — no chunk, no retrieval role beyond the ``ls`` line — so
|
||||
the PATCH handler must never touch the LLM client (the deliberate
|
||||
contrast with the phase-57 document-summary re-embed)."""
|
||||
src = inspect.getsource(docs_api.update_folder_summary)
|
||||
assert "LLMClient" not in src
|
||||
|
||||
|
||||
def test_docs_tree_stat_walk_equivalence_with_flat_list(admin_client, db) -> None:
|
||||
"""The RAG view computes its KB-wide stat cards by walking the
|
||||
in-memory tree — the walk must yield EXACTLY what
|
||||
``GET /api/docs`` reports for the same data (the stat-card values
|
||||
are unchanged by the redesign)."""
|
||||
_truncate_tree_tables(db)
|
||||
base = datetime.now(UTC)
|
||||
db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
|
||||
_seed_doc(db, "Homelab", "a/b/c.md", "C", 3, base)
|
||||
_seed_doc(db, "Homelab", "a/d.md", "D", 2, base)
|
||||
_seed_doc(db, "Homelab", "top.md", "Top", 5, base)
|
||||
_seed_doc(db, "Homelab", "solo.md", "Solo", 0, base)
|
||||
db.commit()
|
||||
|
||||
tree = admin_client.get("/api/docs/tree").json()
|
||||
flat = admin_client.get("/api/docs").json()
|
||||
|
||||
files = _tree_file_nodes(tree["sources"])
|
||||
assert len(files) == len(flat["documents"]) # document count
|
||||
assert sum(f["chunks"] for f in files) == sum(d["chunks"] for d in flat["documents"])
|
||||
# The tree's per-source counts match the flat list's per-source counts.
|
||||
for source in tree["sources"]:
|
||||
flat_count = sum(
|
||||
1 for d in flat["documents"] if d["source"] == source["name"]
|
||||
)
|
||||
assert source["documents"] == flat_count
|
||||
source_files = _tree_file_nodes([source])
|
||||
assert sum(f["chunks"] for f in source_files) == sum(
|
||||
d["chunks"] for d in flat["documents"] if d["source"] == source["name"]
|
||||
)
|
||||
|
||||
_truncate_tree_tables(db)
|
||||
|
||||
Reference in New Issue
Block a user