teng-lin--notebooklm-py/tests/unit/cli/test_generate_characterization.py

"""Characterization (golden-snapshot) tests for ``notebooklm generate`` commands.

These tests lock in the exact byte-for-byte output of every ``generate``
subcommand before the service-layer extraction, so extraction work can prove
handler-regression-free behavior. They cover:

* happy-path text + JSON output for each of the 10 leaf generate commands
* the rate-limited (retry-exhausted) error path
* the ``None``-result generation-failed error path
* video / cinematic-video usage-error message text
* report's "smart custom" format coercion and ``--append`` warning text

Characterization-test discipline:

* This file MUST pass on the PR's branch base (main) **before** the
  extraction commit lands. Run with ``uv run pytest
  tests/unit/cli/test_generate_characterization.py -q`` against pre-T1
  main to confirm; the same suite must remain byte-stable after T1.
* Snapshots are stored as module-level string constants. Any drift in
  Click usage messages, JSON formatting, or status-line wording will
  surface here.
* JSON snapshots are compared as parsed dicts (order/whitespace
  resilient where contract is structural, not formatting); text
  snapshots are compared as exact strings.
"""

from __future__ import annotations

import json
from collections.abc import Callable, Iterator
from typing import Any
from unittest.mock import AsyncMock, patch

import pytest
from click.testing import CliRunner, Result

import notebooklm.auth as auth_module
import notebooklm.cli.helpers as helpers_module
from notebooklm.notebooklm_cli import cli
from notebooklm.types import GenerationStatus

from .conftest import create_mock_client, inject_client

# ---------------------------------------------------------------------------
# Fixtures and helpers
# ---------------------------------------------------------------------------

_AUTH_PAYLOAD = {
    "SID": "t",
    "__Secure-1PSIDTS": "t",
    "HSID": "t",
    "SSID": "t",
    "APISID": "t",
    "SAPISID": "t",
}


@pytest.fixture
def runner() -> CliRunner:
    return CliRunner()


@pytest.fixture
def authed_invoke(
    runner: CliRunner,
) -> Iterator[Callable[..., Result]]:
    """Yield an ``invoke(args, *, configure=None)`` helper with auth seams mocked.

    ``configure`` (when supplied) receives a fresh mock client (already
    attached to the patched ``NotebookLMClient`` constructor) and is
    expected to attach the artifact-generation AsyncMock attributes the
    test needs. When ``configure`` is ``None`` (usage-error tests that
    never reach the API call), the mock client is still installed so the
    auth bootstrap path succeeds; the API methods stay as default
    ``MagicMock``s and are never invoked.
    """

    def _invoke(
        args: list[str],
        *,
        configure: Callable[[Any], None] | None = None,
    ) -> Result:
        mock_client = create_mock_client()
        if configure is not None:
            configure(mock_client)
        with (
            patch.object(
                helpers_module,
                "load_auth_from_storage",
                return_value=_AUTH_PAYLOAD,
            ),
            patch.object(
                auth_module,
                "fetch_tokens_with_domains",
                new_callable=AsyncMock,
            ) as mock_fetch,
        ):
            mock_fetch.return_value = ("csrf", "session")
            return runner.invoke(cli, args, obj=inject_client(mock_client))

    yield _invoke


def _attach_async_return(method_name: str, value: Any) -> Callable[[Any], None]:
    """Build a ``configure`` callable that sets ``artifacts.<method>`` mock."""

    def _apply(mock_client: Any) -> None:
        setattr(mock_client.artifacts, method_name, AsyncMock(return_value=value))

    return _apply


# ---------------------------------------------------------------------------
# Happy-path golden output: text and JSON
# ---------------------------------------------------------------------------

# Mapping (CLI subcommand, ``client.artifacts.*`` method name, extra args, task_id).
# The 10 leaf generate commands covered by this characterization suite.
HAPPY_PATH_CASES: list[tuple[str, str, list[str], str]] = [
    ("audio", "generate_audio", [], "task_audio"),
    ("video", "generate_video", [], "task_video"),
    ("cinematic-video", "generate_cinematic_video", [], "task_cv"),
    ("slide-deck", "generate_slide_deck", [], "task_slides"),
    (
        "revise-slide",
        "revise_slide",
        ["Move title up", "--artifact", "art_1", "--slide", "0"],
        "task_revise",
    ),
    ("quiz", "generate_quiz", [], "task_quiz"),
    ("flashcards", "generate_flashcards", [], "task_flash"),
    ("infographic", "generate_infographic", [], "task_info"),
    ("data-table", "generate_data_table", ["Compare key concepts"], "task_dtable"),
    ("report", "generate_report", [], "task_report"),
]


@pytest.mark.parametrize("cmd,method,extra_args,task_id", HAPPY_PATH_CASES)
def test_generate_happy_path_json_snapshot(
    cmd: str,
    method: str,
    extra_args: list[str],
    task_id: str,
    authed_invoke: Callable[..., Result],
) -> None:
    """JSON output for the 10 generate commands is exactly
    ``{"task_id": <id>, "status": "pending"}`` on a no-wait happy path."""
    result = authed_invoke(
        ["generate", cmd, "--json", "-n", "nb_123", *extra_args],
        configure=_attach_async_return(method, {"task_id": task_id, "status": "processing"}),
    )
    assert result.exit_code == 0, result.output
    payload = json.loads(result.output)
    assert payload == {"task_id": task_id, "status": "pending"}


@pytest.mark.parametrize("cmd,method,extra_args,task_id", HAPPY_PATH_CASES)
def test_generate_happy_path_text_snapshot(
    cmd: str,
    method: str,
    extra_args: list[str],
    task_id: str,
    authed_invoke: Callable[..., Result],
) -> None:
    """Text output for the 10 generate commands is exactly ``Started: <task_id>\\n``
    on a no-wait happy path."""
    result = authed_invoke(
        ["generate", cmd, "-n", "nb_123", *extra_args],
        configure=_attach_async_return(method, {"task_id": task_id, "status": "processing"}),
    )
    assert result.exit_code == 0, result.output
    assert result.output == f"Started: {task_id}\n"


# ---------------------------------------------------------------------------
# Mind-map happy path (its own output shape: full result echoed)
# ---------------------------------------------------------------------------


def test_generate_mind_map_json_snapshot(authed_invoke: Callable[..., Result]) -> None:
    """Mind-map JSON output is the converged {mind_map, note_id, kind} payload.

    The additive ``kind`` key keeps note-backed consumers reading ``mind_map`` /
    ``note_id`` working while marking the backing (issue #1256). ``--kind
    note-backed`` pins the note-backed shape (the interactive default is covered
    by the routing tests in ``test_generate.py``).
    """
    payload = {
        "note_id": "n1",
        "mind_map": {"name": "Root", "children": [{"a": 1}, {"b": 2}]},
    }
    result = authed_invoke(
        ["generate", "mind-map", "--kind", "note-backed", "--json", "-n", "nb_123"],
        configure=_attach_async_return("generate_mind_map", payload),
    )
    assert result.exit_code == 0, result.output
    assert json.loads(result.output) == {**payload, "kind": "note_backed"}


def test_generate_mind_map_text_snapshot(authed_invoke: Callable[..., Result]) -> None:
    """Mind-map text output is the kind-agnostic ``ID / Kind / Root / Children`` block.

    Passing ``--kind note-backed`` explicitly pins the note-backed shape (the
    interactive default is covered by the routing tests in ``test_generate.py``).
    """
    payload = {
        "note_id": "n1",
        "mind_map": {"name": "Root", "children": [{"a": 1}, {"b": 2}]},
    }
    result = authed_invoke(
        ["generate", "mind-map", "--kind", "note-backed", "-n", "nb_123"],
        configure=_attach_async_return("generate_mind_map", payload),
    )
    assert result.exit_code == 0, result.output
    assert result.output == (
        "Mind map generated:\n  ID: n1\n  Kind: note_backed\n  Root: Root\n  Children: 2 nodes\n"
    )


# ---------------------------------------------------------------------------
# Error paths
# ---------------------------------------------------------------------------


def test_generate_audio_failure_none_result_json(
    authed_invoke: Callable[..., Result],
) -> None:
    """When the generation API returns ``None`` the CLI emits a structured JSON
    error and exits non-zero."""
    result = authed_invoke(
        ["generate", "audio", "--json", "-n", "nb_123"],
        configure=_attach_async_return("generate_audio", None),
    )
    assert result.exit_code == 1
    assert json.loads(result.output) == {
        "error": True,
        "code": "GENERATION_FAILED",
        "message": "Audio generation failed",
    }


def test_generate_audio_failure_none_result_text(
    authed_invoke: Callable[..., Result],
) -> None:
    """When the generation API returns ``None`` text mode prints the
    failure message to stderr and exits non-zero."""
    result = authed_invoke(
        ["generate", "audio", "-n", "nb_123"],
        configure=_attach_async_return("generate_audio", None),
    )
    assert result.exit_code == 1
    # ``output_error`` writes via ``safe_echo(err=True)``. Assert against
    # ``result.stderr`` directly so a regression that accidentally writes
    # the failure message to stdout (e.g. wrong ``err=`` flag) is caught.
    # Click 8.2+ keeps ``stdout``/``stderr``/``output`` as independent
    # streams; ``output`` is a merge and would mask the leak.
    assert "Audio generation failed" in result.stderr
    assert result.stdout == ""


def test_generate_audio_rate_limit_retry_exhausted_json(
    authed_invoke: Callable[..., Result],
) -> None:
    """Retry-exhausted rate-limit (retry=0) returns the RATE_LIMITED JSON
    payload and exits non-zero."""
    rate_limited = GenerationStatus(
        task_id="task_rl",
        status="failed",
        error_code="RATE_LIMIT_EXCEEDED",
        error="rate limited",
    )
    result = authed_invoke(
        ["generate", "audio", "--json", "-n", "nb_123"],
        configure=_attach_async_return("generate_audio", rate_limited),
    )
    assert result.exit_code == 1
    assert json.loads(result.output) == {
        "error": True,
        "code": "RATE_LIMITED",
        "message": "Audio generation rate limited by Google.",
    }


def test_generate_audio_rate_limit_retry_exhausted_text(
    authed_invoke: Callable[..., Result],
) -> None:
    """Retry-exhausted rate-limit (retry=0) text mode prints the rate-limit
    message AND the hint, then exits non-zero."""
    rate_limited = GenerationStatus(
        task_id="task_rl",
        status="failed",
        error_code="RATE_LIMIT_EXCEEDED",
        error="rate limited",
    )
    result = authed_invoke(
        ["generate", "audio", "-n", "nb_123"],
        configure=_attach_async_return("generate_audio", rate_limited),
    )
    assert result.exit_code == 1
    # Rate-limit message + hint both go to stderr via ``output_error``
    # (``safe_echo(err=True)``). Pin them on ``result.stderr`` so a leak
    # to stdout would surface; ``result.stdout`` stays empty.
    assert result.stderr == (
        "Audio generation rate limited by Google.\n"
        "Daily quota may be exceeded. Try again in 1-24 hours, "
        "or use --retry N to retry automatically.\n"
    )
    assert result.stdout == ""


# ---------------------------------------------------------------------------
# Video / cinematic-video usage-error text
# ---------------------------------------------------------------------------


def test_video_cinematic_rejects_style_prompt(
    authed_invoke: Callable[..., Result],
) -> None:
    """``--style-prompt`` is rejected when the video format is cinematic.

    Per ADR-0015, the post-parse validation routes through ``output_error``:
    exit 1 (VALIDATION_ERROR), message on stderr, no usage footer.
    """
    result = authed_invoke(
        [
            "generate",
            "video",
            "--format",
            "cinematic",
            "--style-prompt",
            "foo",
            "-n",
            "nb_123",
        ],
    )
    assert result.exit_code == 1
    # ``output_error`` writes the message to stderr in text mode — pin the
    # assertion there so a stdout leak would surface.
    assert "--style-prompt cannot be used with cinematic video" in result.stderr


def test_video_style_custom_requires_style_prompt(
    authed_invoke: Callable[..., Result],
) -> None:
    """``--style custom`` requires ``--style-prompt`` for non-cinematic video."""
    result = authed_invoke(["generate", "video", "--style", "custom", "-n", "nb_123"])
    assert result.exit_code == 1
    # ``output_error`` writes the message to stderr in text mode.
    assert "--style custom requires --style-prompt" in result.stderr


def test_video_style_prompt_requires_style_custom(
    authed_invoke: Callable[..., Result],
) -> None:
    """``--style-prompt`` requires ``--style custom`` for non-cinematic video."""
    result = authed_invoke(["generate", "video", "--style-prompt", "foo", "-n", "nb_123"])
    assert result.exit_code == 1
    # ``output_error`` writes the message to stderr in text mode.
    assert "--style-prompt requires --style custom" in result.stderr


def test_cinematic_video_alias_rejects_non_cinematic_format(
    authed_invoke: Callable[..., Result],
) -> None:
    """``generate cinematic-video --format explainer`` is rejected with a
    targeted message that points users at ``generate video``."""
    result = authed_invoke(
        [
            "generate",
            "cinematic-video",
            "--format",
            "explainer",
            "-n",
            "nb_123",
        ],
    )
    assert result.exit_code == 1
    # ``output_error`` writes the message to stderr in text mode.
    assert "--format must be 'cinematic' for the cinematic-video subcommand" in result.stderr


# ---------------------------------------------------------------------------
# Report "smart-custom" coercion and --append warning
# ---------------------------------------------------------------------------


def test_report_description_triggers_custom_format(
    authed_invoke: Callable[..., Result],
) -> None:
    """A bare description on ``generate report`` (with default --format
    briefing-doc) is smart-detected as a custom report — the API call gets
    ``ReportFormat.CUSTOM`` and the description as ``custom_prompt``."""
    captured: dict[str, Any] = {}

    async def _capture(*args: Any, **kwargs: Any) -> Any:
        captured.update(kwargs)
        return {"task_id": "task_report_custom", "status": "processing"}

    def _configure(mock_client: Any) -> None:
        mock_client.artifacts.generate_report = AsyncMock(side_effect=_capture)

    result = authed_invoke(
        ["generate", "report", "My custom prompt", "-n", "nb_123"],
        configure=_configure,
    )
    assert result.exit_code == 0, result.output
    # ReportFormat.CUSTOM == "custom" by enum string identity.
    assert captured["report_format"].value == "custom"
    assert captured["custom_prompt"] == "My custom prompt"
    assert captured["extra_instructions"] is None


def test_report_custom_with_append_emits_warning(
    authed_invoke: Callable[..., Result],
) -> None:
    """``--append`` is a no-op when ``--format custom`` is in effect; the CLI
    emits a stderr warning and clears the append payload before calling the
    API."""
    captured: dict[str, Any] = {}

    async def _capture(*args: Any, **kwargs: Any) -> Any:
        captured.update(kwargs)
        return {"task_id": "task_report_custom_w", "status": "processing"}

    def _configure(mock_client: Any) -> None:
        mock_client.artifacts.generate_report = AsyncMock(side_effect=_capture)

    result = authed_invoke(
        [
            "generate",
            "report",
            "--format",
            "custom",
            "--append",
            "extra",
            "desc",
            "-n",
            "nb_123",
        ],
        configure=_configure,
    )
    assert result.exit_code == 0, result.output
    # ``_emit_warnings`` writes via ``click.echo(..., err=True)`` — pin
    # the warning text on stderr so a stdout leak (which would also break
    # the success-text snapshot below) would be caught here too.
    assert (
        "Warning: --append has no effect with --format custom. "
        "Use the description argument instead." in result.stderr
    )
    # Warning side-effect: append is suppressed (CLI sets it to None).
    assert captured["extra_instructions"] is None


# ---------------------------------------------------------------------------
# Group-level help (regression guard on subcommand listing)
# ---------------------------------------------------------------------------


def test_generate_help_lists_all_ten_kinds(runner: CliRunner) -> None:
    """``notebooklm generate --help`` lists every leaf subcommand by name.

    Lock-in protects against accidental command drop during refactor.
    """
    result = runner.invoke(cli, ["generate", "--help"])
    assert result.exit_code == 0
    expected_subcommands = [
        "audio",
        "video",
        "cinematic-video",
        "slide-deck",
        "revise-slide",
        "quiz",
        "flashcards",
        "infographic",
        "data-table",
        "mind-map",
        "report",
    ]
    for name in expected_subcommands:
        assert name in result.output, f"missing subcommand {name!r}: {result.output}"