/tests/integration/test_workflow_tracer_vcr.py

"""Tracer-bullet full-workflow VCR cassette.

End-to-end VCR cassette covering the most common user journey through the
client API: **create-notebook → add-source → ask → generate → download**.
This is the integration-tier regression test for the golden path — if a
refactor breaks any of the five public-API entry points the test exercises,
the cassette replay will surface the regression before it ships.

The five phases captured in the cassette:

1. ``client.notebooks.create`` — fresh scratch notebook for this recording.
2. ``client.sources.add_url`` + ``client.sources.wait_until_ready`` — add a
   URL source and block until it is processed.
3. ``client.chat.ask`` — single chat turn against the new notebook.
4. ``client.artifacts.generate_report`` + ``client.artifacts.wait_for_completion``
   — generate a Briefing Doc and poll until completion.
5. ``client.artifacts.download_report`` — write the markdown to a tempfile
   and assert the bytes were produced.

A final ``client.notebooks.delete`` is also recorded so the cassette captures
the full lifecycle — including teardown — and replay does not leave any
orphan state assertions.

Recording
---------
Pre-condition: ``notebooklm login`` succeeds against the default Chromium
profile. The recorder creates its own scratch notebook so it cannot
collide with parallel agents that may be using the shared
``NOTEBOOKLM_GENERATION_NOTEBOOK_ID`` notebook. The scratch notebook is
deleted as the final recorded RPC.

To re-record this cassette::

    export NOTEBOOKLM_VCR_RECORD=1
    uv run pytest tests/integration/test_workflow_tracer_vcr.py -v -s

The ``-s`` flag lets the recorder log progress between phases (useful when
the report-generation poll takes a minute or two).

Replay
------
The default VCR matcher includes ``rpcids``. This cassette also
opts in to the ``freq`` body matcher because the chat-ask phase
hits the streaming endpoint, whose disambiguator lives in the
form-encoded ``f.req`` body rather than the URL query string. ``rpcids``
is a no-op for the streaming endpoint and ``freq`` is a no-op for
batchexecute endpoints, so the two matchers compose cleanly.

Size budget
-----------
The cassette is the regression test for the user journey, not a media
fixture — keep it ≤ 5 MB. If a future re-record blows past the budget
because the report response body has grown, the right fix is usually to
ask a shorter question or pick a smaller report format (e.g. swap
``BRIEFING_DOC`` for a custom prompt that targets a tight word count).
Do NOT skip any of the five phases to shrink the cassette — the full
chain is the test.
"""

from __future__ import annotations

import asyncio
import uuid
from pathlib import Path

import pytest
import yaml

from notebooklm import NotebookLMClient
from notebooklm.rpc import RPCMethod
from notebooklm.types import ReportFormat
from tests.integration.conftest import get_vcr_auth, skip_no_cassettes
from tests.vcr_config import _is_vcr_record_mode, notebooklm_vcr

pytestmark = [pytest.mark.vcr, skip_no_cassettes]

CASSETTE_NAME = "workflow_tracer_bullet.yaml"
CASSETTE_PATH = Path(__file__).parent.parent / "cassettes" / CASSETTE_NAME

# Wikipedia is stable, fast to process, and produces a short-but-non-trivial
# source. Picked because the page is text-only (no PDFs/embeds) and the title
# is easily asserted on.
_TRACER_URL = "https://en.wikipedia.org/wiki/Tracer_bullet"
_TRACER_QUESTION = "In one sentence, what is a tracer bullet?"


@pytest.fixture
def fast_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
    """Skip polling backoff during replay while preserving live record cadence."""
    if _is_vcr_record_mode():
        return

    async def instant_sleep(_seconds: float, result: object | None = None) -> object | None:
        return result

    monkeypatch.setattr(asyncio, "sleep", instant_sleep)


class TestWorkflowTracerBullet:
    """Records and replays the full create→add→ask→generate→download journey."""

    @pytest.mark.vcr
    @pytest.mark.asyncio
    @notebooklm_vcr.use_cassette(
        CASSETTE_NAME,
        # The chat-ask phase hits the streaming endpoint; ``freq`` decodes the
        # ``f.req`` body so the streaming POST is disambiguated by its param
        # shape rather than replay-order. ``rpcids`` already covers the
        # batchexecute phases via the default matcher list — re-listing it
        # here keeps the per-cassette ``match_on`` self-contained.
        match_on=["method", "scheme", "host", "port", "path", "rpcids", "freq"],
    )
    async def test_full_workflow(self, tmp_path: Path, fast_sleep: None) -> None:
        """End-to-end user journey produces a downloadable report.

        Asserts each phase's intermediate output:

        - Phase 1: ``notebooks.create`` returns a notebook with a non-empty
          UUID id.
        - Phase 2: ``sources.add_url`` returns a source with the same URL,
          and ``wait_until_ready`` produces a READY source.
        - Phase 3: ``chat.ask`` returns a non-empty answer and a
          conversation id.
        - Phase 4: ``artifacts.generate_report`` returns a task id and the
          subsequent ``wait_for_completion`` reports status=completed.
        - Phase 5: ``artifacts.download_report`` writes a non-empty file to
          the requested path.

        The final ``notebooks.delete`` call ensures the cassette captures
        the full lifecycle; we do not assert on its return value beyond
        "did not raise" since deletion semantics are exercised by other
        cassettes (``test_vcr_comprehensive``).
        """
        # Use a UUID-suffixed title so a hypothetical retry in record mode
        # cannot collide with a pre-existing notebook with the same title
        # (the ``idempotent_create`` probe would otherwise be ambiguous).
        # During replay, the cassette drives the response regardless of the
        # title we pass, so the UUID is purely a record-mode safety hatch.
        title = f"T8.E3 tracer-bullet {uuid.uuid4().hex[:8]}"

        auth = await get_vcr_auth()
        async with NotebookLMClient(auth) as client:
            # ---------------------------------------------------------------
            # Phase 1: create-notebook
            # ---------------------------------------------------------------
            notebook = await client.notebooks.create(title)
            assert notebook.id, "create-notebook must return a non-empty id"
            assert isinstance(notebook.id, str)
            notebook_id = notebook.id

            try:
                # -----------------------------------------------------------
                # Phase 2: add-source
                # -----------------------------------------------------------
                source = await client.sources.add_url(
                    notebook_id, _TRACER_URL, wait=True, wait_timeout=120.0
                )
                assert source.id, "add_url must return a source with a non-empty id"
                # ``url`` round-trip — what we sent comes back on the source.
                # (Replay drives the URL from the cassette body, so this is a
                #  replay-time invariant too.)
                assert source.url == _TRACER_URL
                # ``wait=True`` should leave the source ready by the time
                # ``add_url`` returns.
                assert source.is_ready, f"source not ready after wait: status={source.status}"

                # -----------------------------------------------------------
                # Phase 3: ask
                # -----------------------------------------------------------
                ask_result = await client.chat.ask(notebook_id, _TRACER_QUESTION)
                assert ask_result.answer, "ask must return a non-empty answer"
                assert isinstance(ask_result.answer, str)
                assert ask_result.conversation_id, "ask must return a conversation id"

                # -----------------------------------------------------------
                # Phase 4: generate (Briefing Doc — fastest, text-only artifact)
                # -----------------------------------------------------------
                gen = await client.artifacts.generate_report(
                    notebook_id, report_format=ReportFormat.BRIEFING_DOC
                )
                assert gen.task_id, "generate_report must return a task_id"
                assert gen.status != "failed", f"generation failed: {gen.error}"

                # Poll for completion. The report endpoint typically resolves
                # within 30–60 s; cap at 5 minutes to leave headroom for slow
                # backends during record-mode without bloating the cassette.
                final = await client.artifacts.wait_for_completion(
                    notebook_id,
                    gen.task_id,
                    initial_interval=2.0,
                    max_interval=10.0,
                    timeout=300.0,
                )
                assert final.status == "completed", (
                    f"report did not complete: status={final.status} error={final.error}"
                )

                # -----------------------------------------------------------
                # Phase 5: download
                # -----------------------------------------------------------
                output_path = tmp_path / "tracer_report.md"
                written_path = await client.artifacts.download_report(
                    notebook_id, str(output_path), artifact_id=gen.task_id
                )
                assert Path(written_path) == output_path
                assert output_path.exists(), "download_report must write the markdown file"
                content = output_path.read_text(encoding="utf-8")
                assert content, "downloaded report must be non-empty"
                # Briefing docs always have *some* markdown structure; a stray
                # binary-blob download would not produce any hash mark anywhere.
                assert "#" in content, (
                    "downloaded report should contain at least one markdown heading"
                )
            finally:
                # Teardown: delete the scratch notebook so the recording does
                # not leave orphan state in the user's account even if an
                # earlier assertion fails. Captured in the cassette so replay
                # also drives the delete RPC.
                deleted = await client.notebooks.delete(notebook_id)
                assert deleted is None

    def test_cassette_size_under_budget(self) -> None:
        """Cassette stays under the 5 MB budget documented in the module docstring.

        A re-record that blows past the budget is a signal to (a) re-ask a
        shorter question or (b) re-record with the recorder's
        ``before_record_response`` hook stripped, not to silently accept a
        ballooning fixture. Failing the check loudly in CI surfaces the
        issue at PR time.
        """
        assert CASSETTE_PATH.exists(), (
            f"cassette missing: {CASSETTE_PATH}. "
            "Re-record with NOTEBOOKLM_VCR_RECORD=1 — see module docstring."
        )
        size_bytes = CASSETTE_PATH.stat().st_size
        size_mb = size_bytes / (1024 * 1024)
        assert size_mb <= 5.0, (
            f"cassette exceeds 5 MB budget: {size_mb:.2f} MB. "
            "Tighten the recording or reduce response body sizes."
        )

    def test_cassette_captures_all_five_phases(self) -> None:
        """Cassette includes batchexecute RPCs for each of the five phases.

        Asserts the recorded batchexecute interactions contain the rpcids
        that prove each phase fired. We do not pin a strict ordered
        sequence (unlike ``test_mind_map_chain_vcr.py``) because the chat
        phase mixes a streaming POST with several batchexecute support
        RPCs whose order is implementation-defined — but the set of
        required rpcids is stable.

        Required RPC IDs (any order, may appear more than once):

        - ``CREATE_NOTEBOOK`` — phase 1
        - ``ADD_SOURCE`` — phase 2
        - ``GET_NOTEBOOK`` — phase 2 wait-until-ready polls (and chat
          ``get_source_ids``)
        - ``CREATE_ARTIFACT`` — phase 4
        - ``LIST_ARTIFACTS`` — phase 4 polling + phase 5 download lookup
        - ``DELETE_NOTEBOOK`` — teardown

        The streaming chat call (phase 3) does not appear in batchexecute
        rpcids; we assert its presence via the URI check below.
        """
        assert CASSETTE_PATH.exists(), f"cassette missing: {CASSETTE_PATH}"

        with CASSETTE_PATH.open(encoding="utf-8") as fh:
            cassette = yaml.safe_load(fh)

        interactions = cassette.get("interactions", [])
        assert len(interactions) >= 5, (
            f"cassette must capture at least 5 interactions; got {len(interactions)}"
        )

        from urllib.parse import parse_qs, urlparse

        recorded_rpcids: set[str] = set()
        streaming_chat_seen = False
        for interaction in interactions:
            uri = interaction.get("request", {}).get("uri", "")
            parsed = urlparse(uri)
            if "/batchexecute" in parsed.path:
                qs = parse_qs(parsed.query)
                for rpc_id in qs.get("rpcids", []):
                    recorded_rpcids.add(rpc_id)
            # Streaming chat lives on a non-batchexecute path under
            # ``/_/LabsTailwindUi/data/`` — its URI contains the service
            # name ``LabsTailwindOrchestrationService/GenerateFreeFormStreamed``
            # as a path component rather than a query param. Matching on the
            # method name (``GenerateFreeFormStreamed``) is sufficient and
            # keeps the assertion tolerant of host changes.
            if "GenerateFreeFormStreamed" in uri:
                streaming_chat_seen = True

        required = {
            RPCMethod.CREATE_NOTEBOOK.value,  # phase 1
            RPCMethod.ADD_SOURCE.value,  # phase 2
            RPCMethod.GET_NOTEBOOK.value,  # phase 2 polls + chat source-id lookup
            RPCMethod.CREATE_ARTIFACT.value,  # phase 4
            RPCMethod.LIST_ARTIFACTS.value,  # phase 4 polls + phase 5 download
            RPCMethod.DELETE_NOTEBOOK.value,  # teardown
        }
        missing = required - recorded_rpcids
        assert not missing, (
            f"cassette missing required rpcids: {sorted(missing)}. "
            f"Recorded: {sorted(recorded_rpcids)}"
        )

        assert streaming_chat_seen, (
            "cassette must include the streaming chat interaction (phase 3); "
            "no URI matched 'GenerateFreeFormStreamed'"
        )


# Skip the on-disk assertions when we are actively recording — the recorder
# writes the cassette after the test body returns, so the file does not yet
# exist at the time the shape tests run if collected in the same session.
# In replay mode these tests run normally.
if _is_vcr_record_mode():  # pragma: no cover — record-time only
    TestWorkflowTracerBullet.test_cassette_size_under_budget = pytest.mark.skip(  # type: ignore[method-assign]
        reason="Cassette not yet written during record run; replay verifies."
    )(TestWorkflowTracerBullet.test_cassette_size_under_budget)
    TestWorkflowTracerBullet.test_cassette_captures_all_five_phases = pytest.mark.skip(  # type: ignore[method-assign]
        reason="Cassette not yet written during record run; replay verifies."
    )(TestWorkflowTracerBullet.test_cassette_captures_all_five_phases)