brain-of-reese/tests/e2e/test_import_extensions_env.py

"""Phase 56 E2E (Playwright): a NOVEL extension (``.sh``) flows config →
import → chunks → mock summary → Sources page.

TODO.md L6: "Allow the user to specify extensions to be read in .env,
don't hard-code working extensions." The subject is the env-driven
extension scope (``import_extensions="md,sh"``); the story-dedicated
fixture (``tests/fixtures/extension_kb/``) is seeded in-process against
the deterministic mock LLM — the phase-02 seeding-thread pattern, the
fixture, not the subject of the tests.

Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_import_extensions_env.py -v --no-cov

DB isolation: the fixture's source name (``extension_kb``) is
distinctive — the suite never asserts on absolute row counts and
deletes the rows it creates in a ``finally`` (other suites' documents
stay untouched in the shared E2E database).
"""
from __future__ import annotations

import asyncio
from collections.abc import Iterator
from pathlib import Path
from threading import Thread
from typing import Any

import pytest
from playwright.sync_api import Page, expect
from sqlalchemy import select

from app.config import Settings
from app.db import SessionLocal
from app.models import Document
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import login

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "extension_kb"
SOURCE = FIXTURES.name  # "extension_kb" — distinctive, never asserted by count
SH_REL = "homelab/scripts/uptime.sh"
MD_REL = "homelab/notes/note.md"
SENTINEL = "UPTIME-PROBE-SENTINEL-9c2f"


async def _import_fixtures(mock_port: int, extensions: str) -> ImportSummary:
    kwargs: dict[str, Any] = {
        "_env_file": None,
        "llm_base_url": f"http://127.0.0.1:{mock_port}/v1",
        "import_extensions": extensions,
    }
    settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return await import_sources([FIXTURES], LLMClient(settings))


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread.

    Playwright's sync API keeps an asyncio loop running on the test
    thread, so ``asyncio.run`` cannot be called directly from a test
    body.
    """
    box: dict[str, Any] = {}

    def runner() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as e:  # noqa: BLE001 — re-raised on the test thread
            box["error"] = e

    t = Thread(target=runner)
    t.start()
    t.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _delete_source_rows() -> None:
    """Delete every row of this suite's distinctive source (chunks
    cascade with the document rows)."""
    with SessionLocal() as db:
        for doc in db.scalars(select(Document).where(Document.source == SOURCE)).all():
            db.delete(doc)
        db.commit()


@pytest.fixture(autouse=True)
def extension_kb(mock_llm: int, db_ready: None) -> Iterator[ImportSummary]:
    """Seed the fixture with the NOVEL scope (``md,sh``) for one test
    and delete every row it creates afterwards (DB isolation — see the
    module docstring)."""
    _delete_source_rows()  # idempotent: leftovers from a crashed run
    summary = _run_in_thread(_import_fixtures(mock_llm, "md,sh"))
    try:
        yield summary
    finally:
        _delete_source_rows()


def test_admin_sources_lists_the_novel_extension(
    page: Page, app_url: str, extension_kb: ImportSummary
) -> None:
    # The seed saw exactly the two fixture files in their formats — the
    # novel .sh extension walked, chunked, and summarized.
    assert extension_kb.formats == {"sh": 1, "md": 1}
    assert (extension_kb.added, extension_kb.errors) == (2, 0)

    login(page, app_url)  # phase 16: the catalog is admin-only
    # The novel .sh document is listed; the path cell carries the full
    # path (the column is ellipsized — the title attribute is the pin).
    row = page.locator("#docs-tbody tr", has_text=SH_REL)
    expect(row).to_have_count(1)
    link = row.locator("td:nth-child(2) a.doc-link")
    expect(link).to_have_count(1)
    expect(link).to_have_attribute("title", SH_REL)
    # The markdown control doc is listed too (never asserted by count —
    # other suites' documents may share the shared E2E database).
    expect(page.locator("#docs-tbody tr", has_text=MD_REL)).to_have_count(1)

    # Format badge: the row's path link opens the same-page modal and
    # its meta row shows the .sh format (house assertion style —
    # test_document_viewer.py asserts the same locator for yaml/md).
    before = len(page.context.pages)
    link.click()
    assert len(page.context.pages) == before, "clicking a row link must not open a new tab"
    expect(page.locator("#doc-modal-meta .doc-source-badge")).to_have_text(SOURCE)
    expect(page.locator("#doc-modal-meta .format-badge")).to_have_text("sh")
    # Non-markdown content renders as escaped monospace text in a pre —
    # the sentinel proves it is THIS document's content.
    pre = page.locator("#doc-modal-content pre.doc-raw")
    expect(pre).to_have_count(1)
    expect(pre).to_contain_text(SENTINEL)
    # Still on the Sources page: no navigation happened.
    assert page.url == app_url + "/sources.html", f"navigated away: {page.url}"


def test_anonymous_sources_gate_and_no_api_docs(
    page: Page, app_url: str, extension_kb: ImportSummary
) -> None:
    """A fresh anonymous context (function-scoped ``page`` = new
    browser context, no cookies): the sign-in gate renders and the page
    never calls ``/api/docs`` — the phase-16 pin, regression-checked
    with the novel-extension KB seeded."""
    api_docs_calls: list[str] = []
    page.on(
        "request",
        lambda r: api_docs_calls.append(r.url) if "/api/docs" in r.url else None,
    )

    page.goto(f"{app_url}/sources.html")
    # The gate, with its sign-in link — not a redirect.
    gate = page.locator("#sources-gate")
    expect(gate).to_be_visible()
    expect(gate).to_contain_text("Sign in to view the full catalog")
    expect(gate.locator("a[href='/login.html?next=/sources.html']")).to_have_count(1)
    # Stat cards + table hidden…
    expect(page.locator("#stat-cards")).to_be_hidden()
    expect(page.locator("#docs-table")).to_be_hidden()
    expect(page.locator("#sources-empty")).to_be_hidden()
    # …and NO /api/docs call was ever made.
    assert api_docs_calls == [], f"anonymous sources page called /api/docs: {api_docs_calls}"