library-desk/tests/test_tenant_isolation_live.py

"""
Live tenant-isolation test (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 a wiki page + ingest it as ``llm_tester``.
2. /query/hybrid as ``llm_tester``: the tenant's own content is reachable
   and ZERO results come from the ``jpmschweitzer`` tenant.
3. /query/hybrid as a THIRD, nonexistent tenant inside the reserved
   namespace (``llm_tester_void``): ZERO results entirely — without ever
   writing as ``jpmschweitzer``.
4. Teardown deletes the created page; 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_tenant_isolation_live.py -v
"""

import uuid

import httpx
import pytest

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

pytestmark = pytest.mark.integration

#: Third tenant: nonexistent, but still inside the reserved namespace so
#: even its side effects (persisted SearchQuery nodes) stay in test space.
GHOST_TENANT = f"{TEST_TENANT}_void"

# Disable the web leg so "zero results" is meaningful, and re-ranking so
# the test does not depend on LLM latency.
HYBRID_CONFIG = {
    "enable_web": False,
    "enable_reranking": False,
    "enable_vector": True,
    "enable_graph": True,
    "enable_volatile": True,
    "enable_documents": True,
    "final_result_count": 20,
}


@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 isolation 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(180.0, connect=10.0),
    )
    yield client
    client.close()


@pytest.fixture(scope="module")
def marker() -> str:
    """Unique content marker for this run."""
    return f"xylophone quantum walrus {uuid.uuid4().hex[:10]}"


@pytest.fixture(scope="module")
def ingested_page(api, marker):
    """Create + ingest a wiki page as the test tenant; delete afterwards."""
    assert_safe_test_tenant(TEST_TENANT)

    page_slug = f"isolation-probe-{uuid.uuid4().hex[:8]}"
    create_resp = api.post(
        "/wiki/pages",
        json={
            "title": f"Tenant Isolation Probe {marker}",
            "path": f"/isolation-tests/{page_slug}",
            "content": (
                f"# Tenant Isolation Probe\n\n"
                f"The secret marker phrase is: {marker}.\n"
                f"This page belongs exclusively to the {TEST_TENANT} tenant "
                f"and is deleted by the test teardown."
            ),
            "description": "Ephemeral tenant-isolation test page",
            "tags": ["isolation-test"],
            "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}")

    # Deterministic ingestion (vectors + graph) as the 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
    ingest = ingest_resp.json()
    assert ingest["success"] is True
    assert ingest["vector_chunks_created"] >= 1

    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


def _hybrid(api, user: str, query: str) -> dict:
    resp = api.post(
        "/query/hybrid",
        params={"user": user},
        json={"query": query, "config": HYBRID_CONFIG},
    )
    assert resp.status_code == 200, resp.text
    return resp.json()


class TestLiveTenantIsolation:
    def test_own_tenant_sees_own_content_and_nothing_from_production(
        self, api, marker, ingested_page
    ):
        assert_safe_test_tenant(TEST_TENANT)
        data = _hybrid(api, TEST_TENANT, f"secret marker phrase {marker}")

        # 1) Sanity: the tenant's own freshly-ingested page is retrievable,
        #    proving the pipeline works and the zero-assertions below are
        #    meaningful.
        own_hits = [
            r for r in data["results"]
            if r.get("page_id") == ingested_page["page_id"]
        ]
        assert own_hits, (
            f"Expected the ingested page {ingested_page['page_id']} in "
            f"results: {[(r.get('source_type'), r.get('title')) for r in data['results']]}"
        )

        # 2) ZERO results from the production tenant.
        for result in data["results"]:
            path = (result.get("page_path") or "")
            assert PRODUCTION_TENANT not in path, (
                f"PRODUCTION LEAK: result path {path!r} for tenant {TEST_TENANT}"
            )
            if result["source_type"] in ("wiki", "vector", "graph"):
                assert path.lstrip("/").startswith(f"users/{TEST_TENANT}"), (
                    f"Cross-tenant wiki result: {path!r}"
                )

        # 3) The formatted LLM context must not leak production paths either.
        assert PRODUCTION_TENANT not in data["context"]

    def test_nonexistent_tenant_gets_zero_results(self, api, marker, ingested_page):
        """The inverse check without writing as jpmschweitzer: a third,
        nonexistent tenant must see NOTHING - not llm_tester's page and
        not jpmschweitzer's corpus."""
        assert_safe_test_tenant(GHOST_TENANT)

        data = _hybrid(api, GHOST_TENANT, f"secret marker phrase {marker}")

        assert data["total_results"] == 0, (
            f"Nonexistent tenant {GHOST_TENANT} got results: "
            f"{[(r.get('source_type'), r.get('title'), r.get('page_path')) for r in data['results']]}"
        )
        assert data["results"] == []

    def test_nonexistent_tenant_gets_zero_results_on_generic_query(self, api):
        """Even a broad query over common homelab topics returns nothing
        for a tenant with no data."""
        assert_safe_test_tenant(GHOST_TENANT)

        data = _hybrid(api, GHOST_TENANT, "docker kubernetes home server setup")

        assert data["total_results"] == 0