teng-lin--notebooklm-py/tests/unit/app/test_app_notebooks.py

"""Unit tests for the transport-neutral ``notebooklm._app.notebooks`` core.

These pin the Click-free notebook workflows at the ``_app`` boundary with a
``MagicMock`` client + an injected partial-id resolver (the CLI normally
injects ``cli.resolve.resolve_notebook_id``):

* ``create`` / ``delete`` / ``rename`` / ``describe`` (summary) / ``metadata``
  executors delegate to the right ``client.notebooks`` RPC and project the typed
  result dataclasses,
* the resolver is threaded through ``rename`` / ``describe`` / ``metadata`` and
  the resolved id flows into the downstream RPC,
* ``describe`` tolerates a ``None`` description (the CLI renders both views from
  the typed field).

The CLI tests keep ownership of the ``--use`` context side effect, the
serializers, the ``--json`` envelopes, and the error-category exit codes (the
generic error classification is covered by ``app/test_app_errors.py``).
"""

from __future__ import annotations

from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock

import pytest

from notebooklm._app.notebooks import (
    NotebookCreateResult,
    NotebookDescribeResult,
    NotebookMetadataResult,
    NotebookRenameResult,
    execute_notebook_create,
    execute_notebook_delete,
    execute_notebook_describe,
    execute_notebook_metadata,
    execute_notebook_rename,
)
from notebooklm.exceptions import NotebookNotFoundError
from notebooklm.types import (
    Notebook,
    NotebookDescription,
    NotebookMetadata,
    SuggestedTopic,
)


def _client() -> MagicMock:
    client = MagicMock()
    client.notebooks = MagicMock()
    return client


async def _resolve_nb(_client, nb_id, *, json_output=False):
    """Identity resolver that prefixes to verify the *resolved* id flows downstream."""
    return f"full_{nb_id}"


_CREATED_AT = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
_MODIFIED_AT = datetime(2026, 1, 3, 4, 5, 6, tzinfo=timezone.utc)


# ---------------------------------------------------------------------------
# create
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_execute_notebook_create_projects_notebook() -> None:
    client = _client()
    # Both timestamps already populated → the best-effort backfill is a no-op
    # (no wasted GET re-read), so this stays a pure projection assertion.
    notebook = Notebook(
        id="nb_new",
        title="My notebook",
        created_at=_CREATED_AT,
        modified_at=_MODIFIED_AT,
    )
    client.notebooks.create = AsyncMock(return_value=notebook)

    result = await execute_notebook_create(client, "My notebook")

    assert isinstance(result, NotebookCreateResult)
    assert result.notebook is notebook
    client.notebooks.create.assert_awaited_once_with("My notebook")
    client.notebooks.get.assert_not_called()


# The create RPC (CREATE_NOTEBOOK / CCqFvf) returns null created_at/modified_at;
# the core does ONE best-effort GET re-read to backfill just those two slots so
# every adapter (CLI/REST/MCP) surfaces populated timestamps on create (#1705,
# lifting the MCP-only #1699 fix into the transport-neutral home).


@pytest.mark.asyncio
async def test_execute_notebook_create_backfills_null_timestamps() -> None:
    client = _client()
    created = Notebook(id="nb_new", title="New", sources_count=0, is_owner=True)
    client.notebooks.create = AsyncMock(return_value=created)
    # The GET diverges on every non-timestamp field to prove ONLY the two
    # timestamp slots are taken from it; the create stays authoritative.
    client.notebooks.get = AsyncMock(
        return_value=Notebook(
            id="nb_new",
            title="Stale",
            created_at=_CREATED_AT,
            sources_count=9,
            is_owner=False,
            modified_at=_MODIFIED_AT,
        )
    )

    nb = (await execute_notebook_create(client, "New")).notebook

    assert nb.created_at == _CREATED_AT  # backfilled from GET
    assert nb.modified_at == _MODIFIED_AT  # backfilled from GET
    assert nb.title == "New"  # from create, not the divergent GET
    assert nb.sources_count == 0  # from create
    assert nb.is_owner is True  # from create
    client.notebooks.get.assert_awaited_once_with("nb_new")


@pytest.mark.asyncio
async def test_execute_notebook_create_fills_only_the_null_slot() -> None:
    """Per-key + additive: a populated create timestamp is never overwritten."""
    client = _client()
    created = Notebook(id="nb_new", title="New", created_at=_CREATED_AT)  # modified_at None
    client.notebooks.create = AsyncMock(return_value=created)
    client.notebooks.get = AsyncMock(
        return_value=Notebook(
            id="nb_new",
            title="New",
            created_at=datetime(2099, 1, 1, tzinfo=timezone.utc),  # must NOT win
            modified_at=_MODIFIED_AT,
        )
    )

    nb = (await execute_notebook_create(client, "New")).notebook

    assert nb.created_at == _CREATED_AT  # create's value preserved, GET ignored
    assert nb.modified_at == _MODIFIED_AT  # only the null slot filled from GET
    client.notebooks.get.assert_awaited_once_with("nb_new")


@pytest.mark.asyncio
async def test_execute_notebook_create_reread_failure_falls_back() -> None:
    """A failed GET re-read must not fail a successful create (best-effort)."""
    client = _client()
    client.notebooks.create = AsyncMock(return_value=Notebook(id="nb_new", title="New"))
    client.notebooks.get = AsyncMock(side_effect=NotebookNotFoundError("nb_new"))

    nb = (await execute_notebook_create(client, "New")).notebook  # must not raise

    assert nb.created_at is None
    assert nb.modified_at is None
    client.notebooks.get.assert_awaited_once_with("nb_new")


@pytest.mark.asyncio
async def test_execute_notebook_create_reread_still_null_stays_null() -> None:
    """If the GET itself still has null timestamps (propagation lag), stay null."""
    client = _client()
    client.notebooks.create = AsyncMock(return_value=Notebook(id="nb_new", title="New"))
    client.notebooks.get = AsyncMock(return_value=Notebook(id="nb_new", title="New"))

    nb = (await execute_notebook_create(client, "New")).notebook

    assert nb.created_at is None
    assert nb.modified_at is None
    client.notebooks.get.assert_awaited_once_with("nb_new")


@pytest.mark.asyncio
async def test_execute_notebook_create_empty_id_skips_reread() -> None:
    """No id → no ``get("")``; the create result is returned untouched."""
    client = _client()
    client.notebooks.create = AsyncMock(return_value=Notebook(id="", title="New"))
    client.notebooks.get = AsyncMock()

    nb = (await execute_notebook_create(client, "New")).notebook

    assert nb.created_at is None
    client.notebooks.get.assert_not_awaited()


# ---------------------------------------------------------------------------
# delete
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_execute_notebook_delete_delegates_to_client() -> None:
    client = _client()
    client.notebooks.delete = AsyncMock(return_value=None)

    # ``delete()`` returns None and raises on failure (issue #1211); reaching
    # here without an exception is the success contract.
    await execute_notebook_delete(client, "nb_1")

    client.notebooks.delete.assert_awaited_once_with("nb_1")


@pytest.mark.asyncio
async def test_execute_notebook_delete_propagates_failure() -> None:
    """``delete()`` raises on real failure (issue #1211); the core does not swallow it."""
    client = _client()
    client.notebooks.delete = AsyncMock(side_effect=RuntimeError("boom"))

    with pytest.raises(RuntimeError, match="boom"):
        await execute_notebook_delete(client, "nb_1")


# ---------------------------------------------------------------------------
# rename — resolver threading
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_execute_notebook_rename_resolves_then_renames() -> None:
    client = _client()
    client.notebooks.rename = AsyncMock(return_value=None)

    result = await execute_notebook_rename(
        client, "nb_part", "New title", resolve_notebook_id=_resolve_nb
    )

    assert isinstance(result, NotebookRenameResult)
    assert result.notebook_id == "full_nb_part"
    assert result.new_title == "New title"
    # The *resolved* id flows into the rename RPC, not the partial input.
    client.notebooks.rename.assert_awaited_once_with("full_nb_part", "New title")


# ---------------------------------------------------------------------------
# describe (summary)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_execute_notebook_describe_returns_description() -> None:
    client = _client()
    description = NotebookDescription(
        summary="A summary",
        suggested_topics=[SuggestedTopic(question="Q?", prompt="Ask Q")],
    )
    client.notebooks.get_description = AsyncMock(return_value=description)

    result = await execute_notebook_describe(client, "nb_part", resolve_notebook_id=_resolve_nb)

    assert isinstance(result, NotebookDescribeResult)
    assert result.notebook_id == "full_nb_part"
    assert result.description is description
    client.notebooks.get_description.assert_awaited_once_with("full_nb_part")


@pytest.mark.asyncio
async def test_execute_notebook_describe_tolerates_none_description() -> None:
    client = _client()
    client.notebooks.get_description = AsyncMock(return_value=None)

    result = await execute_notebook_describe(client, "nb_1", resolve_notebook_id=_resolve_nb)

    assert result.description is None


# ---------------------------------------------------------------------------
# metadata
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_execute_notebook_metadata_returns_metadata() -> None:
    client = _client()
    notebook = Notebook(id="full_nb_part", title="My notebook")
    metadata = NotebookMetadata(notebook=notebook, sources=[])
    client.notebooks.get_metadata = AsyncMock(return_value=metadata)

    result = await execute_notebook_metadata(client, "nb_part", resolve_notebook_id=_resolve_nb)

    assert isinstance(result, NotebookMetadataResult)
    assert result.notebook_id == "full_nb_part"
    assert result.metadata is metadata
    client.notebooks.get_metadata.assert_awaited_once_with("full_nb_part")