brain-of-reese/tests/integration/test_docs_api_dates.py

"""Integration: the phase-106 date API surface (task 05, D7/D8/D9).

``GET /api/docs`` and ``GET /api/documents/content`` serve the
document's ``created_at``; ``GET /api/docs/tree`` serves the file
``created_at`` verbatim plus the DERIVED subtree-max ``updated_at`` on
folder/source nodes (``null`` for a 0-document registered source); and
the admin-only ``PATCH /api/documents/date`` matrix — set (ISO date or
full ISO datetime, future folds to today, ``created_at_manual``
flagged), clear (null/absent → flag drops, stored date stands),
malformed 422, unknown-pair 404, anonymous/token-user 403 — the
phase-57 split intact (the viewer stays user-gated, the edit is
admin-gated).

Uses the real compose Postgres (``db`` fixture) and FastAPI's
TestClient, mirroring ``test_docs_api.py``.
"""
from __future__ import annotations

import inspect
import uuid
from datetime import UTC, datetime, timedelta

from fastapi.testclient import TestClient
from sqlalchemy import select, text

import app.api.docs as docs_api
from app.core import tokens as token_service
from app.main import app as fastapi_app
from app.models import Chunk, Document, GitSource

_TRUNCATE = "chunks, documents, git_sources"

#: Deliberately distinct creation stamps (the D9 max fixture) — kept
#: separate from ``indexed_at`` so a confused-column pin fails loudly.
C0 = datetime(2020, 1, 1, 0, 0, 0, tzinfo=UTC)
C1 = datetime(2021, 6, 15, 12, 0, 0, tzinfo=UTC)
C2 = datetime(2022, 3, 1, 6, 0, 0, tzinfo=UTC)
C3 = datetime(2023, 11, 30, 23, 59, 59, tzinfo=UTC)


def _truncate(db) -> None:
    db.execute(text(f"TRUNCATE {_TRUNCATE}"))
    db.commit()


def _seed_doc(
    db,
    source: str,
    path: str,
    title: str,
    n_chunks: int,
    indexed_at: datetime,
    created_at: datetime,
) -> Document:
    """One indexed document with an explicit creation date (D8)."""
    doc = Document(
        source=source,
        path=path,
        full_path=f"/tmp/{source}/{path}",
        title=title,
        content=f"# {title}\n\nBody.",
        content_hash=uuid.uuid4().hex,  # unique per row (no real sha needed)
        indexed_at=indexed_at,
        created_at=created_at,
    )
    db.add(doc)
    db.flush()
    if n_chunks:
        db.add_all(
            Chunk(document_id=doc.id, position=i, content=f"chunk {i}", embedding=[0.01] * 768)
            for i in range(n_chunks)
        )
    db.commit()
    return doc


def _tree_node(sources: list[dict], name: str) -> dict:
    return next(s for s in sources if s["name"] == name)


# ---------------------------------------------------------------------------
# Reads — ``created_at`` on the two flat surfaces + the tree (D8/D9).
# ---------------------------------------------------------------------------


def test_docs_list_reports_created_at_per_row(admin_client, db) -> None:
    """``GET /api/docs`` gains ``created_at`` per row (ISO-8601,
    verbatim from the row) — ``indexed_at`` untouched (a different
    concept: the index time)."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "a/top.md", "Top", 1, base, C2)
    _seed_doc(db, "Homelab", "a/b/deep.md", "Deep", 2, base + timedelta(hours=1), C3)
    try:
        r = admin_client.get("/api/docs")
        assert r.status_code == 200
        body = r.json()
        # (source, path) order is unchanged by the new column
        # ("a/b/deep.md" < "a/top.md" — the slash sorts before the 't').
        assert [d["path"] for d in body["documents"]] == ["a/b/deep.md", "a/top.md"]
        by_path = {d["path"]: d for d in body["documents"]}
        assert by_path["a/top.md"]["created_at"] == C2.isoformat()
        assert by_path["a/b/deep.md"]["created_at"] == C3.isoformat()
        # ``indexed_at`` still reports the (different) index stamp —
        # the two concepts never get confused on the wire.
        assert abs(
            datetime.fromisoformat(by_path["a/top.md"]["indexed_at"]) - base
        ).total_seconds() < 5
        assert by_path["a/top.md"]["indexed_at"] != by_path["a/top.md"]["created_at"]
    finally:
        _truncate(db)


def test_document_content_carries_created_at(admin_client, db) -> None:
    """``GET /api/documents/content`` gains ``created_at`` (the viewer's
    top meta row renders the ``Created`` badge from it, task 08)."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 1, base, C1)
    try:
        r = admin_client.get(
            "/api/documents/content", params={"source": "Homelab", "path": "k8s.md"}
        )
        assert r.status_code == 200
        body = r.json()
        assert "created_at" in body
        assert body["created_at"] == C1.isoformat()
        # The two stamps are distinct concepts and both ride the shape.
        assert body["indexed_at"] != body["created_at"]
    finally:
        _truncate(db)


def test_tree_file_dates_and_derived_subtree_max(admin_client, db) -> None:
    """``GET /api/docs/tree``: file ``created_at`` verbatim; folder and
    source ``updated_at`` = the subtree's MAX document ``created_at``
    (D9 — derived in the pure builder, never stored); a registered
    0-document source reports ``updated_at: null`` with no children."""
    _truncate(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    db.add(
        GitSource(
            url="https://github.com/reese/Empty.git",
            kind="git",
            added_at=base + timedelta(hours=1),
        )
    )
    # The D9 max fixture: the deepest file holds the newest date, so a
    # folder that only LOOKS recent at its own level still reports the
    # deeper date (deeper beats shallower at every level here).
    _seed_doc(db, "Homelab", "a/b/c/deep.md", "Deep", 1, base, C3)
    _seed_doc(db, "Homelab", "a/b/shallow.md", "Shallow", 0, base, C1)
    _seed_doc(db, "Homelab", "a/top.md", "Top", 1, base, C2)
    _seed_doc(db, "Homelab", "root.md", "Root", 0, base, C0)
    try:
        r = admin_client.get("/api/docs/tree")
        assert r.status_code == 200
        sources = r.json()["sources"]
        # Registry order leads; the 0-document registered source lists.
        assert [s["name"] for s in sources] == ["Homelab", "Empty"]
        homelab, empty = sources
        assert (empty["documents"], empty["children"], empty["updated_at"]) == (0, [], None)

        assert homelab["documents"] == 4
        assert homelab["updated_at"] == C3.isoformat()
        a = next(c for c in homelab["children"] if c["kind"] == "folder")
        root_md = next(c for c in homelab["children"] if c["kind"] == "file")
        assert a["path"] == "a"
        assert a["documents"] == 3
        assert a["updated_at"] == C3.isoformat()  # deep C3 beats the direct C2
        assert root_md["created_at"] == C0.isoformat()
        a_b = next(c for c in a["children"] if c["kind"] == "folder")
        top = next(c for c in a["children"] if c["kind"] == "file")
        assert a_b["path"] == "a/b"
        assert a_b["documents"] == 2
        assert a_b["updated_at"] == C3.isoformat()  # deep C3 beats the direct C1
        assert top["created_at"] == C2.isoformat()
        # a/b's children: the subfolder first, then its ONE direct file.
        a_b_c, shallow = a_b["children"]
        assert (a_b_c["kind"], a_b_c["path"], a_b_c["documents"]) == ("folder", "a/b/c", 1)
        assert a_b_c["updated_at"] == C3.isoformat()
        assert (shallow["kind"], shallow["path"], shallow["created_at"]) == (
            "file",
            "a/b/shallow.md",
            C1.isoformat(),
        )
        (deep,) = a_b_c["children"]
        assert (deep["kind"], deep["path"], deep["created_at"]) == (
            "file",
            "a/b/c/deep.md",
            C3.isoformat(),
        )
        # File nodes carry their own date only — no ``updated_at`` key.
        for node in (root_md, top, deep, shallow):
            assert "updated_at" not in node
    finally:
        _truncate(db)


# ---------------------------------------------------------------------------
# PATCH /api/documents/date (D7) — the admin document-date editor.
# ---------------------------------------------------------------------------


def test_patch_date_set_round_trips_and_flags_manual(admin_client, db) -> None:
    """Set a bare ``YYYY-MM-DD``: the parse (midnight UTC) is stored
    verbatim, ``created_at_manual`` flips to true, the response echoes
    the stored state, and BOTH read surfaces confirm on re-GET."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    try:
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "2020-01-02"},
        )
        assert r.status_code == 200, r.text
        assert set(r.json()) == {"source", "path", "created_at", "created_at_manual"}
        assert r.json() == {
            "source": "Homelab",
            "path": "k8s.md",
            "created_at": "2020-01-02T00:00:00+00:00",
            "created_at_manual": True,
        }
        # The viewer's re-render source (the echo) and both GETs agree.
        content = admin_client.get(
            "/api/documents/content", params={"source": "Homelab", "path": "k8s.md"}
        ).json()
        assert content["created_at"] == "2020-01-02T00:00:00+00:00"
        list_row = next(
            d
            for d in admin_client.get("/api/docs").json()["documents"]
            if d["path"] == "k8s.md"
        )
        assert list_row["created_at"] == "2020-01-02T00:00:00+00:00"
        # DB state: the flag is set (the sync-time importer then skips
        # this row — D1/D4).
        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "k8s.md")
        )
        assert row is not None
        assert row.created_at == datetime(2020, 1, 2, tzinfo=UTC)
        assert row.created_at_manual is True
    finally:
        _truncate(db)


def test_patch_date_full_iso_datetime_is_converted_to_utc(admin_client, db) -> None:
    """A full ISO datetime with a NON-UTC offset is converted to UTC
    before storing (D3 — the single normalization choke point)."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    try:
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "2021-06-15T12:30:00+02:00"},
        )
        assert r.status_code == 200, r.text
        assert r.json()["created_at"] == "2021-06-15T10:30:00+00:00"
        assert r.json()["created_at_manual"] is True
    finally:
        _truncate(db)


def test_patch_date_malformed_422_and_leaves_row_untouched(admin_client, db) -> None:
    """A malformed non-null value 422s in the HANDLER (the model field
    is an unconstrained ``str | None`` on purpose, so the detail can
    name the field) — and the row is left untouched."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    try:
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "not-a-date"},
        )
        assert r.status_code == 422
        assert r.json() == {
            "detail": "date must be an ISO date or datetime (e.g. 2024-06-15)"
        }
        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "k8s.md")
        )
        assert row is not None
        assert row.created_at == C0  # untouched
        assert row.created_at_manual is False  # untouched
    finally:
        _truncate(db)


def test_patch_date_null_clear_drops_flag_and_keeps_date(admin_client, db) -> None:
    """The CLEAR (D7): ``date: null`` drops ``created_at_manual`` ONLY —
    the stored date stands until the next sync refreshes it (the API is
    DB-only). An ABSENT ``date`` key is the same operation."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    try:
        # Set first, so there is a correction to clear.
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "2020-01-02"},
        )
        assert r.status_code == 200 and r.json()["created_at_manual"] is True

        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": None},
        )
        assert r.status_code == 200, r.text
        assert r.json() == {
            "source": "Homelab",
            "path": "k8s.md",
            "created_at": "2020-01-02T00:00:00+00:00",  # the date STOOD
            "created_at_manual": False,
        }

        # Re-set, then clear with the key ABSENT — same operation.
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "2020-01-02"},
        )
        assert r.status_code == 200 and r.json()["created_at_manual"] is True
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md"},
        )
        assert r.status_code == 200, r.text
        assert r.json()["created_at_manual"] is False
        assert r.json()["created_at"] == "2020-01-02T00:00:00+00:00"

        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "k8s.md")
        )
        assert row is not None
        assert row.created_at == datetime(2020, 1, 2, tzinfo=UTC)  # still standing
        assert row.created_at_manual is False
    finally:
        _truncate(db)


def test_patch_date_future_folds_to_today(admin_client, db) -> None:
    """A manually set FUTURE date also folds to today (D3 — the same
    normalization choke point as the sourced path), and it is STILL
    flagged manual (the owner's correction survives syncs until
    cleared)."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    try:
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "k8s.md", "date": "2999-01-01"},
        )
        assert r.status_code == 200, r.text
        stored = datetime.fromisoformat(r.json()["created_at"])
        assert abs((stored - datetime.now(UTC)).total_seconds()) < 300  # ≈ today
        assert r.json()["created_at_manual"] is True
        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "k8s.md")
        )
        assert row is not None
        assert row.created_at_manual is True
        assert row.created_at != datetime(2999, 1, 1, tzinfo=UTC)
    finally:
        _truncate(db)


def test_patch_date_404_unknown_pair_and_traversal(admin_client, db) -> None:
    """Row-lookup semantics (the ``/documents/content`` rule): unknown
    pairs — including traversal strings — are simply not rows → 404
    ``document not found``; nothing is written."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "real.md", "Real", 0, base, C0)
    try:
        for source, path in (
            ("Ghost", "x.md"),  # unknown source
            ("Homelab", "nope.md"),  # known source, unknown path
            ("Homelab", "../../etc/passwd"),  # traversal: not a row
        ):
            r = admin_client.patch(
                "/api/documents/date",
                json={"source": source, "path": path, "date": "2020-01-02"},
            )
            assert r.status_code == 404, (source, path, r.status_code)
            assert r.json() == {"detail": "document not found"}, (source, path)
        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "real.md")
        )
        assert row is not None
        assert row.created_at_manual is False  # nothing written
    finally:
        _truncate(db)


def test_patch_date_403_anonymous_and_token_user(client, db) -> None:
    """The ``require_admin`` gate (the phase-57 split): an anonymous
    caller AND a live access-token user who is not the admin both get
    403 ``admin only`` — while the viewer content itself stays
    user-gated (a token holder could read the document, just not edit
    its date)."""
    _truncate(db)
    base = datetime.now(UTC)
    _seed_doc(db, "Homelab", "k8s.md", "K8s", 0, base, C0)
    db.execute(text("TRUNCATE api_tokens"))
    db.commit()
    body = {"source": "Homelab", "path": "k8s.md", "date": "2020-01-02"}
    try:
        # Anonymous (the shared ``client`` is unsigned in this module).
        r = client.patch("/api/documents/date", 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).
        _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/documents/date", json=body)
        assert r.status_code == 403
        assert r.json() == {"detail": "admin only"}
        # …and the SPLIT: the same holder CAN read the document (the
        # viewer stays user-gated, phase 79).
        got = holder.get(
            "/api/documents/content", params={"source": "Homelab", "path": "k8s.md"}
        )
        assert got.status_code == 200
        db.expire_all()
        row = db.scalar(
            select(Document).where(Document.source == "Homelab", Document.path == "k8s.md")
        )
        assert row is not None
        assert row.created_at_manual is False  # the rejected edits wrote nothing
    finally:
        db.execute(text("TRUNCATE api_tokens"))
        db.commit()
        _truncate(db)


def test_patch_date_handler_is_db_only_and_never_constructs_an_llm_client() -> None:
    """Source pin (the house pattern of the phase-97 folder-summary
    editor): a date is NEVER embedded — no chunk, no retrieval role —
    so the handler must never touch the LLM client (the deliberate
    contrast with the phase-57 ``is_summary`` re-embed)."""
    src = inspect.getsource(docs_api.update_document_date)
    assert "LLMClient" not in src