brain-of-reese/tests/unit/test_models.py

"""Unit tests: SQLAlchemy models register the pgvector schema on the metadata.

Importing :mod:`app.models` is what Alembic's ``env.py`` and the runtime rely
on; these tests lock the table/column contract (PLAN §5) without a live DB.
"""
from __future__ import annotations

from pgvector.sqlalchemy import Vector
from sqlalchemy import TextClause, UniqueConstraint
from sqlalchemy.sql.schema import DefaultClause

import app.models  # noqa: F401  (import registers all tables on Base.metadata)
from app.db import Base


def test_all_tables_registered() -> None:
    tables = Base.metadata.tables
    assert "documents" in tables
    assert "chunks" in tables
    assert "query_log" in tables
    assert "ui_settings" in tables  # phase 91: the single-row UI settings


def test_ui_settings_single_row_nullable_contract() -> None:
    """Phase 91 (9 identity colors after phase 92, task 01; the 8
    semantic state colors after phase 93, task 01 — B3 revised): the
    single-row UI settings table — Integer PK ``id`` with the
    Python-side ``default=1`` (the row is always id 1), the 3 strings
    VARCHAR(300) and the 17 palette colors VARCHAR(7), ALL nullable
    (NULL = default — B1: env value for the strings, the built-in
    palette for the colors)."""
    settings_table = Base.metadata.tables["ui_settings"]
    assert set(settings_table.c.keys()) == {
        "id", "app_name", "input_placeholder", "footer_text",
        "bg", "surface", "ink", "ink_soft", "line", "grid_line",
        "brand", "brand_soft", "brand_ink",
        "ok_bg", "ok_ink", "err_bg", "err_ink", "err_line",
        "accent_bg", "accent_ink", "accent_line",
    }
    pk = settings_table.c["id"]
    assert pk.primary_key is True, "ui_settings.id must be the PK"
    assert pk.default is not None, "id needs the Python-side default=1"
    for name in ("app_name", "input_placeholder", "footer_text"):
        col = settings_table.c[name]
        assert col.nullable is True, f"{name} must be NULL (env default)"
        assert getattr(col.type, "length", None) == 300, f"{name} must be String(300)"
    for name in ("bg", "surface", "ink", "ink_soft", "line", "grid_line",
                 "brand", "brand_soft", "brand_ink",
                 "ok_bg", "ok_ink", "err_bg", "err_ink", "err_line",
                 "accent_bg", "accent_ink", "accent_line"):
        col = settings_table.c[name]
        assert col.nullable is True, f"{name} must be NULL (the built-in)"
        assert getattr(col.type, "length", None) == 7, f"{name} must be String(7) — #rrggbb"


def test_chunks_embedding_is_vector_768() -> None:
    chunks = Base.metadata.tables["chunks"]
    col = chunks.c["embedding"]
    assert isinstance(col.type, Vector)
    assert col.type.dim == 768
    # Embeddings are two-phase: inserted first, embedded later.
    assert col.nullable is True


def test_chunks_reference_documents_cascade() -> None:
    chunks = Base.metadata.tables["chunks"]
    fkc = list(chunks.foreign_key_constraints)[0]
    assert fkc.elements[0].column.table.name == "documents"
    assert fkc.ondelete == "CASCADE"


def test_documents_image_context_column_contract() -> None:
    """Phase 125 (task 01, LOCKED A1): ``documents.image_context`` —
    TEXT, NULLABLE, NO server default. NULL is the standalone-image /
    pre-phase-125 value (no backfill — A4 keeps those rows
    byte-identical)."""
    from sqlalchemy import Text as SAText

    documents = Base.metadata.tables["documents"]
    col = documents.c["image_context"]
    assert col.nullable is True, "image_context must be NULLABLE (pre-phase rows)"
    assert isinstance(col.type, SAText), "image_context must be TEXT"
    assert col.server_default is None, (
        "image_context must carry NO server default — NULL is the"
        " standalone/pre-phase value"
    )
    assert col.default is None, (
        "image_context must carry NO ORM default clause either (an"
        " omitted column inserts NULL, the text-doc value)"
    )


def test_document_image_context_orm_parses() -> None:
    """A freshly constructed ``Document`` (no DB) accepts the field in
    both states: a set context string round-trips as-is, and
    ``None`` (and the omitted kwarg) read NULL — the standalone-image
    / pre-phase state."""
    from app.models import Document

    base = dict(
        source="s",
        path="p.md",
        full_path="/tmp/s/p.md",
        title="p",
        content="c",
        content_hash="1" * 64,
    )
    enriched = Document(**base, image_context="some context")
    assert enriched.image_context == "some context"

    explicit_null = Document(**base, image_context=None)
    assert explicit_null.image_context is None

    omitted = Document(**base)
    assert omitted.image_context is None, (
        "an omitted image_context must read NULL (the text-doc value)"
    )


def test_documents_unique_source_path() -> None:
    documents = Base.metadata.tables["documents"]
    uq = [
        c
        for c in documents.constraints
        if isinstance(c, UniqueConstraint)
        and {col.name for col in c.columns} == {"source", "path"}
    ]
    assert uq, "documents must be unique on (source, path) — the upsert key"


def test_documents_file_hash_column_contract() -> None:
    """Phase 147 (task 01): ``documents.file_hash`` — the sha256 of the
    RAW SOURCE FILE's bytes (the sync's change signal): VARCHAR(64),
    NULLABLE (NULL = "unknown" — pre-phase rows until the first
    post-upgrade sync stamps it), NO server default, and NO index
    (the fast path compares it in Python after the unique
    ``(source, path)`` lookup — internal bookkeeping, never a query
    surface). ``content_hash`` (0001) stays untouched: NOT NULL and
    still indexed — the retriever's dedupe/caching identity."""
    from sqlalchemy import String as SAString

    documents = Base.metadata.tables["documents"]
    col = documents.c["file_hash"]
    assert col.nullable is True, ("file_hash must be NULLABLE — NULL is the pre-phase state")
    assert isinstance(col.type, SAString), "file_hash must be String(64)"
    assert col.type.length == 64, "file_hash must be String(64) — a sha256 hex digest"
    assert col.server_default is None, (
        "file_hash must carry NO server default — NULL is the meaningful"
        " pre-phase state"
    )
    # No index: neither the inline flag (None = unspecified) nor any
    # table-level index may cover file_hash (unlike content_hash — a
    # query/exposure surface).
    assert not col.index, ("file_hash must be UNINDEXED — compared in Python")
    covering = [
        idx
        for idx in documents.indexes
        if any(c.name == "file_hash" for c in idx.columns)
    ]
    assert not covering, "no table-level index may cover file_hash"
    uq_covering = [
        c
        for c in documents.constraints
        if isinstance(c, UniqueConstraint) and any(col.name == "file_hash" for col in c.columns)
    ]
    assert not uq_covering, "no unique constraint may cover file_hash"

    # content_hash is untouched: still NOT NULL and indexed.
    content = documents.c["content_hash"]
    assert content.nullable is False, ("content_hash must stay NOT NULL (0001)")
    assert content.index is True, "content_hash must stay INDEXED (0001)"
    content_covering = [
        idx for idx in documents.indexes if any(c.name == "content_hash" for c in idx.columns)
    ]
    assert content_covering, "the content_hash index must survive phase 147"


def test_document_file_hash_orm_defaults_to_none() -> None:
    """A freshly constructed ``Document`` (no DB) exposes ``file_hash``:
    the omitted kwarg and explicit ``None`` read NULL (the pre-phase
    shape — NULL is backfilled with a real digest only by the first
    post-upgrade sync), and a set digest round-trips as-is (the real
    DB round-trip is pinned in ``test_migration_0025.py``)."""
    from app.models import Document

    base = dict(
        source="s",
        path="p.md",
        full_path="/tmp/s/p.md",
        title="p",
        content="c",
        content_hash="1" * 64,
    )
    stamped = Document(**base, file_hash="a" * 64)
    assert stamped.file_hash == "a" * 64

    explicit_null = Document(**base, file_hash=None)
    assert explicit_null.file_hash is None

    omitted = Document(**base)
    assert omitted.file_hash is None, (
        "an omitted file_hash must read NULL (the pre-phase shape)"
    )


def test_doc_drafts_token_is_unique_not_null() -> None:
    """Phase 59: the draft's URL credential — an unguessable uuid4,
    UNIQUE + NOT NULL (no "un-drafted" state, unlike the NULLable
    ``saved_chats.share_token``)."""
    drafts = Base.metadata.tables["doc_drafts"]
    token = drafts.c["token"]
    assert token.nullable is False, "doc_drafts.token must be NOT NULL"
    uq = [
        c
        for c in drafts.constraints
        if isinstance(c, UniqueConstraint)
        and {col.name for col in c.columns} == {"token"}
    ]
    assert uq, "doc_drafts must be unique on (token) — the URL credential"


def test_git_sources_ignore_paths_column_contract() -> None:
    """Phase 89: every source row carries its ignore list — JSONB, NOT
    NULL, server default ``'[]'`` (a pre-phase-89 row reads ``[]``, so
    every existing source imports exactly as before)."""
    sources = Base.metadata.tables["git_sources"]
    assert "ignore_paths" in sources.c, (
        "git_sources must have the ignore_paths column (phase 89)"
    )
    col = sources.c["ignore_paths"]
    assert col.nullable is False, "git_sources.ignore_paths must be NOT NULL"
    sd = col.server_default
    assert isinstance(sd, DefaultClause), "ignore_paths needs a server default"
    assert isinstance(sd.arg, TextClause), (
        "the server default must be the literal SQL text '[]'"
    )
    assert sd.arg.text == "'[]'", "ignore_paths server default must be '[]'"


def test_git_source_python_default_empty_list() -> None:
    """A freshly constructed row (no DB) resolves to an empty ignore
    list via the Python-side default (``default=list``) — the ORM
    INSERT-time default, so an ORM insert that omits the column inserts
    ``[]`` rather than NULL (the server default ``'[]'`` independently
    covers non-ORM inserts)."""
    from app.models import GitSource

    row = GitSource(url="https://example.com/r.git", kind="git")
    col = row.__table__.c["ignore_paths"]
    assert col.default is not None, (
        "ignore_paths needs a Python-side (INSERT-time) default"
    )
    # Invoked with the (unused) execution context at INSERT time — a
    # freshly constructed row that omits the kwarg stores [] rather
    # than NULL (the real-DB behaviour is pinned against the dev DB
    # by the git-sources API integration tests).
    assert col.default.arg(None) == [], "the default must resolve to []"


def test_git_sources_include_hidden_column_contract() -> None:
    """Phase 105: every source row carries its hidden-folders flag —
    BOOLEAN, NOT NULL, server default ``false`` (a pre-phase-105 row
    reads ``False``, so every existing source imports byte-identically
    until the owner flips it — A4)."""
    sources = Base.metadata.tables["git_sources"]
    assert "include_hidden" in sources.c, (
        "git_sources must have the include_hidden column (phase 105)"
    )
    col = sources.c["include_hidden"]
    assert col.nullable is False, "git_sources.include_hidden must be NOT NULL"
    sd = col.server_default
    assert isinstance(sd, DefaultClause), "include_hidden needs a server default"
    assert isinstance(sd.arg, TextClause), (
        "the server default must be the literal SQL text false"
    )
    assert sd.arg.text == "false", "include_hidden server default must be false"


def test_git_source_python_default_include_hidden_false() -> None:
    """A freshly constructed row (no DB) resolves the flag to ``False``
    via the Python-side default (``default=False``) — the ORM INSERT-time
    default, so an ORM insert that omits the column inserts ``false``
    rather than NULL (the server default ``false`` independently covers
    non-ORM inserts — A4)."""
    from app.models import GitSource

    row = GitSource(url="https://example.com/r.git", kind="git")
    col = row.__table__.c["include_hidden"]
    assert col.default is not None, (
        "include_hidden needs a Python-side (INSERT-time) default"
    )
    # A scalar default: the arg IS the value (not a callable).
    assert col.default.arg is False, "the default must resolve to False"


def test_documents_page_strip_mode_column_contract() -> None:
    """Phase 163 (task 01): ``documents.page_strip_mode`` — the CLASS of
    the page strip a non-NULL ``page_count`` describes (``"primary"`` =
    the phase-133 fallback class, the strip IS the view; ``"additive"``
    = the phase-140 pictorial class, text + a Text/Pages toggle):
    VARCHAR(16), NULLABLE (NULL = no strip — and the pre-phase shape for
    a row whose strip stands), NO server default, and NO index (the
    content endpoint reads it from the ROW — a display fact, never a
    query surface, the ``file_hash`` / ``page_count`` convention).
    ``page_count`` (0026) stays exactly as phase 148 left it."""
    from sqlalchemy import String as SAString

    documents = Base.metadata.tables["documents"]
    assert "page_strip_mode" in documents.c, (
        "documents must have the page_strip_mode column (phase 163)"
    )
    col = documents.c["page_strip_mode"]
    assert col.nullable is True, (
        "page_strip_mode must be NULLABLE — NULL means no strip (and the"
        " pre-phase row whose strip stands)"
    )
    assert isinstance(col.type, SAString), "page_strip_mode must be a String column"
    assert col.type.length == 16, (
        "page_strip_mode must be String(16) — the two class names, bounded"
    )
    assert col.server_default is None, (
        "page_strip_mode must carry NO server default — NULL is the"
        " meaningful pre-phase state"
    )
    assert not col.index, (
        "page_strip_mode must be UNINDEXED — the content endpoint reads it"
        " from the row, no query filters on it"
    )
    covering = [
        idx
        for idx in documents.indexes
        if any(c.name == "page_strip_mode" for c in idx.columns)
    ]
    assert not covering, "no table-level index may cover page_strip_mode"

    # page_count is untouched: still the nullable, unindexed sibling.
    page_count = documents.c["page_count"]
    assert page_count.nullable is True, "page_count must stay NULLABLE (0026)"
    assert not page_count.index, "page_count must stay UNINDEXED (0026)"


def test_document_page_strip_mode_orm_defaults_to_none() -> None:
    """A freshly constructed ``Document`` (no DB) exposes
    ``page_strip_mode``: an omitted kwarg and an explicit ``None`` both
    read NULL (no strip — the shape every non-qualifying document keeps,
    and the shape a pre-phase-163 row keeps WHILE its ``page_count``
    stands, which the wire serializes ABSENT), and each class name
    round-trips verbatim as a plain string (the real DB round-trip is
    pinned in ``test_migration_0027.py``)."""
    from app.models import Document

    base = dict(
        source="s",
        path="p.pdf",
        full_path="/tmp/s/p.pdf",
        title="p",
        content="c",
        content_hash="1" * 64,
        page_count=36,
    )
    for mode in ("primary", "additive"):
        assert Document(**base, page_strip_mode=mode).page_strip_mode == mode

    assert Document(**base, page_strip_mode=None).page_strip_mode is None
    omitted = Document(**base)
    assert omitted.page_strip_mode is None, (
        "an omitted page_strip_mode must read NULL while page_count stands"
        " (the pre-phase strip the viewer keeps showing as 'primary')"
    )


def test_doc_drafts_column_contract() -> None:
    """Phase 59: the editable triple (title/path/body) + status +
    timestamps are NOT NULL; ``branch`` / ``commit_sha`` are NULL
    until the push endpoint records them."""
    drafts = Base.metadata.tables["doc_drafts"]
    assert set(drafts.c.keys()) == {
        "id", "token", "title", "path", "body", "status",
        "branch", "commit_sha", "created_at", "updated_at",
    }
    for name in ("title", "path", "body", "status", "created_at", "updated_at"):
        assert drafts.c[name].nullable is False, f"{name} must be NOT NULL"
    for name in ("branch", "commit_sha"):
        assert drafts.c[name].nullable is True, f"{name} must be NULL until pushed"
    assert drafts.c["status"].default is not None, "status needs an ORM default (draft)"
    assert drafts.c["token"].default is not None, "token needs an ORM default (uuid4)"