library-desk/tests/test_quality_report_live.py

"""
Live end-to-end test for the weekly quality report (integration, guard-gated).

Runs against the LOCAL wakeup server (./wakeup.sh, port 8778 — never the
production container on 8089) with the shared backing services, entirely
under the reserved llm_tester tenant namespace:

1. Create + ingest a wiki page as ``llm_tester`` (deliberately without tags
   so the report has something to flag).
2. POST /maintenance/quality-report {user: llm_tester}: asserts the report
   is generated, mentions the created page in the missing-metadata section,
   folds in integrity results, and writes the dated report page under
   users/llm_tester/system/quality-reports/YYYY-MM-DD.
3. Fetches the written report page back from the wiki.
4. Teardown deletes the created pages; the session-scoped teardown in
   conftest purges every remaining llm_tester artifact.

Run with:
    RUN_INTEGRATION_TESTS=1 TEST_HOST=<shared-host> \\
        LIBRARY_DESK_URL=http://localhost:8778 \\
        .venv/bin/python -m pytest tests/test_quality_report_live.py -v
"""

import time
import uuid
from datetime import datetime, timezone

import httpx
import pytest

from tests.conftest import (
    LIBRARY_DESK_URL,
    PRODUCTION_TENANT,
    TEST_TENANT,
    assert_safe_test_tenant,
)

pytestmark = pytest.mark.integration


@pytest.fixture(scope="module")
def api():
    """HTTP client for the local dev server, with bearer auth."""
    from src.config import get_settings

    assert ":8089" not in LIBRARY_DESK_URL, (
        "Refusing to run the live quality-report test against the "
        "production container (port 8089); point LIBRARY_DESK_URL at ./wakeup.sh"
    )
    settings = get_settings()
    client = httpx.Client(
        base_url=LIBRARY_DESK_URL,
        headers={"Authorization": f"Bearer {settings.library_api_key}"},
        timeout=httpx.Timeout(300.0, connect=10.0),
    )
    yield client
    client.close()


@pytest.fixture(scope="module")
def ingested_page(api):
    """Create + ingest a metadata-poor wiki page as the test tenant."""
    assert_safe_test_tenant(TEST_TENANT)

    slug = f"quality-probe-{uuid.uuid4().hex[:8]}"
    create_resp = api.post(
        "/wiki/pages",
        json={
            "title": f"Quality Probe {slug}",
            "path": f"/quality-tests/{slug}",
            "content": (
                "# Quality Probe\n\n"
                "Ephemeral page used to verify the weekly quality report. "
                "It intentionally has no tags so the report flags it."
            ),
            "description": "",
            "tags": [],
            "user": TEST_TENANT,
        },
    )
    assert create_resp.status_code == 201, create_resp.text
    page = create_resp.json()
    page_id = page["id"]
    assert page["path"].lstrip("/").startswith(f"users/{TEST_TENANT}")

    ingest_resp = api.post(
        "/ingest/page",
        json={"page_id": page_id, "user": TEST_TENANT, "force_refresh": True},
    )
    assert ingest_resp.status_code == 200, ingest_resp.text

    # Wiki.js updates its page-listing index asynchronously after creation;
    # the quality report relies on that listing, so wait until the new page
    # is visible (up to ~30s) before running the report.
    deadline = time.time() + 30
    while time.time() < deadline:
        listing = api.get("/wiki/pages", params={"user": TEST_TENANT})
        assert listing.status_code == 200, listing.text
        if any(p["id"] == page_id for p in listing.json().get("pages", [])):
            break
        time.sleep(2)
    else:
        pytest.fail(f"Page {page_id} never appeared in the Wiki.js listing")

    yield {"page_id": page_id, "path": page["path"]}

    delete_resp = api.delete(f"/wiki/pages/{page_id}", params={"user": TEST_TENANT})
    assert delete_resp.status_code == 200, delete_resp.text


class TestLiveQualityReport:
    def test_quality_report_end_to_end(self, api, ingested_page):
        assert_safe_test_tenant(TEST_TENANT)

        resp = api.post(
            "/maintenance/quality-report",
            json={"user": TEST_TENANT, "stale_days": 30},
        )
        assert resp.status_code == 200, resp.text
        data = resp.json()

        assert data["success"] is True
        assert data["user"] == TEST_TENANT
        assert data["duration_ms"] > 0

        # The dated report page was written inside the tenant's namespace
        today = datetime.now(timezone.utc).date().isoformat()
        expected_path = f"users/{TEST_TENANT}/system/quality-reports/{today}"
        assert data["page_path"] == expected_path
        assert data["page_id"]
        assert PRODUCTION_TENANT not in data["page_path"]

        # The metadata-poor probe page is flagged
        flagged_ids = {p["page_id"] for p in data["pages_missing_metadata"]}
        assert ingested_page["page_id"] in flagged_ids

        # Integrity results are folded in (cached or inline)
        assert data["integrity"] is not None
        assert data["integrity"]["source"] in ("cached", "inline")
        assert "counts" in data["integrity"]

        # Report content includes the summary + the probe page path
        assert "## Summary" in data["report"]
        assert ingested_page["path"].lstrip("/") in data["report"]

        # The written page is retrievable from the wiki via the API
        page_resp = api.get(
            f"/wiki/pages/{data['page_id']}", params={"user": TEST_TENANT}
        )
        assert page_resp.status_code == 200, page_resp.text
        page = page_resp.json()
        assert page["path"].lstrip("/") == expected_path
        assert "## Summary" in page["content"]

        # Same-day rerun must update the same page, not create a duplicate
        rerun = api.post(
            "/maintenance/quality-report",
            json={"user": TEST_TENANT, "stale_days": 30},
        )
        assert rerun.status_code == 200, rerun.text
        assert rerun.json()["page_id"] == data["page_id"]