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).

Phase 97 adaptation: the catalog is the DRILL-DOWN TREE — the rows
live at their folder levels (``extension_kb`` → ``homelab`` →
``scripts`` / ``notes``); the drill is the only change, the asserted
rows/links/modal are unchanged.
"""
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()


def _drill(page: Page, *names: str) -> None:
    """Drill one level at a time (phase 97 — client-side, no fetch,
    no URL change): each name is the EXACT text of the source/folder
    link at the current level."""
    for name in names:
        page.click(f'#folders-tbody a.folder-link:text-is("{name}")')


def _go_top(page: Page) -> None:
    """Back to the top level: the breadcrumb's top-level link (call
    between drills only — the breadcrumb is hidden at the top)."""
    page.locator("#kb-crumb a.kb-crumb-link").first.click()


@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).
    # Phase 97: the rows live at their folder levels — drill source →
    # homelab → scripts first.
    _drill(page, SOURCE, "homelab", "scripts")
    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 (its OWN level — never
    # asserted by count: other suites' documents may share the shared
    # E2E database).
    _go_top(page)
    _drill(page, SOURCE, "homelab", "notes")
    expect(page.locator("#docs-tbody tr", has_text=MD_REL)).to_have_count(1)
    # Back to the .sh row's level — the modal asserts below click its
    # path link.
    _go_top(page)
    _drill(page, SOURCE, "homelab", "scripts")
    link = page.locator("#docs-tbody tr", has_text=SH_REL).locator(
        "td:nth-child(2) a.doc-link"
    )

    # 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}"