"""Init-order / construction regression tests for ``ArtifactsAPI`` / ``NotesAPI``. Before the fix, :class:`ArtifactsAPI` required ``notes_api=client.notes`` at construction time, so :class:`NotesAPI` had to be built first. The shared :mod:`_mind_map` module decouples the two APIs — these tests pin that invariant down so the load-bearing init order can't silently come back. The static AST reach-in / runtime-import boundary *gates* live in ``tests/_guardrails/test_no_facade_reach_in.py``. What remains here are the tests that *exercise* the wired client (constructor-DI seams, seam wiring, mind-map flows). The two boundary tests that still inspect production source with the AST visitors import those helpers from the shared non-test module ``tests/_guardrails/_ast_reach_in.py`` (issue #1431) — the same module the gate file imports them from — so neither test imports from the other. """ from __future__ import annotations import ast import json from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock import pytest from notebooklm._artifacts import ArtifactsAPI from notebooklm._notes import NotesAPI from notebooklm.auth import AuthTokens from notebooklm.client import NotebookLMClient from tests._fixtures.fake_core import FakeSession, make_fake_core from tests._guardrails._ast_reach_in import ( _assignment_value, _call_keyword_value, _facade_construction_lines, _module_function_body, _owned_attr_assignment, _owned_attr_name, _RuntimeImportVisitor, ) from tests._helpers.client_factory import build_client_shell_for_tests pytestmark = pytest.mark.repo_lint SRC_ROOT = Path(__file__).resolve().parents[2] / "src" / "notebooklm" # --------------------------------------------------------------------------- # Constructor-DI seams (``src/notebooklm/_runtime/init.py`` + docs/architecture.md) # # These pin tests guard the post-refactor wiring shape so a future # refactor cannot silently re-introduce the retired module-level # late-binding wrappers (``_decode_response_late_bound``, # ``_sleep_late_bound``, ``_live_is_auth_error``) or the retired # ``Kernel.http_client`` setter. # --------------------------------------------------------------------------- def test_compose_client_internals_exposes_constructor_di_seams() -> None: """``compose_client_internals`` MUST expose the four constructor-DI seams. Stage B1 PR 2 of the post-refactoring plan moved the composition root out of ``NotebookLMClient.__init__`` into ``notebooklm._runtime.init.compose_client_internals``. The seams live on the helper (and on the canonical test builder ``build_client_shell_for_tests``), NOT on ``NotebookLMClient.__init__`` (which preserves the production surface). The seams replace the retired module-level late-binding wrappers and the retired ``Kernel.http_client`` setter. Each must be keyword-only and default to ``None`` so the helper can resolve the canonical seam via a fresh module-attribute lookup at construction time (preserving pre-construction monkeypatch propagation). See ``docs/architecture.md`` for the ClientSeams and Kernel entries. """ import inspect from notebooklm._runtime.init import compose_client_internals sig = inspect.signature(compose_client_internals) for name in ("decode_response", "sleep", "is_auth_error", "async_client_factory"): assert name in sig.parameters, ( f"compose_client_internals must expose constructor-DI kwarg {name!r}" ) param = sig.parameters[name] assert param.kind == inspect.Parameter.KEYWORD_ONLY, ( f"{name!r} must be keyword-only; got {param.kind!r}" ) assert param.default is None, ( f"{name!r} must default to None (None-sentinel + fresh module " f"lookup); got default {param.default!r}" ) def test_session_wires_seam_attributes_for_executor_and_chain() -> None: """Constructor-injected seams MUST reach the executor and chain builder. The ``RpcExecutor`` resolves ``decode_response`` / ``is_auth_error`` / ``sleep`` through closures over ``ClientSeams`` etc., so that tests which rebind ``client._seams.decode_response = stub`` after ``NotebookLMClient.__init__`` (which binds ``client._rpc_executor`` through ``compose_client_internals`` during assembly) still take effect. This test pins both halves: constructor-injected callables reach the executor, AND post-construction rebinds also take effect. """ from notebooklm.auth import AuthTokens auth = AuthTokens( cookies={"SID": "x"}, csrf_token="csrf", session_id="sid", ) def custom_decode(*_a, **_kw): return ["custom"] async def custom_sleep(_seconds): return None def custom_is_auth_error(_exc): return True core = build_client_shell_for_tests( auth, decode_response=custom_decode, sleep=custom_sleep, is_auth_error=custom_is_auth_error, ) assert core._seams.decode_response is custom_decode assert core._seams.sleep is custom_sleep assert core._seams.is_auth_error is custom_is_auth_error executor = core._rpc_executor # Constructor-injected callables propagate through the closure. assert executor._decode_response() == ["custom"] assert executor._is_auth_error(object()) is True # Post-construction rebind takes effect (late-binding contract). def rebound_decode(*_a, **_kw): return ["rebound"] core._seams.decode_response = rebound_decode assert executor._decode_response() == ["rebound"] def test_kernel_http_client_is_read_only_property() -> None: """``Kernel.http_client`` MUST have no setter.""" from notebooklm._kernel import Kernel descriptor = Kernel.__dict__["http_client"] assert isinstance(descriptor, property) assert descriptor.fset is None, ( "Kernel.http_client must remain read-only; the retired setter was a " "test-injection seam that constructor-time async_client_factory " "injection now replaces (see docs/architecture.md Kernel wiring)." ) def test_phase8_source_listing_service_name_and_facade_wiring_are_current() -> None: """Downstream notebook-metadata work depends on the finalized lister name.""" from notebooklm._source.listing import SourceLister from notebooklm._sources import SourcesAPI core = MagicMock() api = SourcesAPI(core, uploader=MagicMock()) assert isinstance(api._lister, SourceLister) def test_phase7_artifact_download_patch_seams_are_current() -> None: """Artifact downloads must use canonical helpers and collaborators. Phase 5 (refactor-history.md Migration Plan steps 6-7) moves the mind-map create/list/extract paths off the ``_mind_map`` module-level seams and onto the injected ``NoteService`` + ``NoteBackedMindMapService`` instances. Downloads should now import their canonical helpers directly rather than resolving through ``notebooklm._artifacts`` at runtime or in type-checking-only imports. """ import notebooklm._artifact.downloads as artifact_downloads import notebooklm._artifact.formatters as artifact_formatters import notebooklm._artifacts as artifacts import notebooklm._mind_map as mind_map import notebooklm.auth as auth tree = ast.parse((SRC_ROOT / "_artifact" / "downloads.py").read_text(encoding="utf-8")) artifact_facade_imports: list[str] = [] artifact_facade_modules = {"_artifacts", "notebooklm._artifacts"} for node in ast.walk(tree): if isinstance(node, ast.Import): artifact_facade_imports.extend( alias.name for alias in node.names if alias.name in artifact_facade_modules ) elif isinstance(node, ast.ImportFrom): module = node.module or "" if module in artifact_facade_modules: artifact_facade_imports.extend(f"{module}.{alias.name}" for alias in node.names) elif module in {"", "notebooklm"}: artifact_facade_imports.extend( alias.name for alias in node.names if alias.name == "_artifacts" ) assert artifact_facade_imports == [] assert artifacts._mind_map is mind_map assert not hasattr(artifact_downloads, "_artifact_seams") assert artifact_downloads.load_httpx_cookies is auth.load_httpx_cookies assert artifact_downloads._extract_app_data is artifact_formatters._extract_app_data assert ( artifact_downloads._format_interactive_content is artifact_formatters._format_interactive_content ) assert artifact_downloads._parse_data_table is artifact_formatters._parse_data_table def test_notebooks_api_has_no_hidden_sources_api_runtime_dependency() -> None: """Notebook metadata must use a narrow lister, not hidden SourcesAPI construction.""" notebooks_tree = ast.parse((SRC_ROOT / "_notebooks.py").read_text(encoding="utf-8")) visitor = _RuntimeImportVisitor( forbidden_names={"SourcesAPI"}, forbidden_modules={"_sources", "notebooklm._sources"}, ) visitor.visit(notebooks_tree) assert visitor.forbidden == [] assert _facade_construction_lines(notebooks_tree, {"SourcesAPI"}) == {} metadata_tree = ast.parse((SRC_ROOT / "_notebook_metadata.py").read_text(encoding="utf-8")) metadata_visitor = _RuntimeImportVisitor( forbidden_names={"SourcesAPI"}, forbidden_modules={"_sources", "notebooklm._sources"}, ) metadata_visitor.visit(metadata_tree) assert metadata_visitor.forbidden == [] assert _facade_construction_lines(metadata_tree, {"SourcesAPI"}) == {} def test_client_constructs_sources_before_notebooks_and_injects_sources_api() -> None: """Client wiring must avoid hidden SourcesAPI construction inside NotebooksAPI. The wiring lives in :func:`notebooklm._client_assembly._assemble_client` (the single construction seam ``NotebookLMClient.__init__`` and the canonical test factory both run), where the client instance is bound to the ``client`` parameter — hence the ``owner="client"`` matchers. """ assembly_tree = ast.parse((SRC_ROOT / "_client_assembly.py").read_text(encoding="utf-8")) assembly_body = _module_function_body(assembly_tree, "_assemble_client") sources_index, sources_assignment = _owned_attr_assignment( assembly_body, "sources", owner="client" ) notebooks_index, notebook_assignment = _owned_attr_assignment( assembly_body, "notebooks", owner="client" ) assert sources_index < notebooks_index sources_value = _assignment_value(sources_assignment) assert isinstance(sources_value, ast.Call) assert isinstance(sources_value.func, ast.Name) assert sources_value.func.id == "SourcesAPI" notebooks_value = _assignment_value(notebook_assignment) assert isinstance(notebooks_value, ast.Call) notebooks_call = notebooks_value assert isinstance(notebooks_call.func, ast.Name) assert notebooks_call.func.id == "NotebooksAPI" assert ( _owned_attr_name(_call_keyword_value(notebooks_call, "sources_api"), owner="client") == "sources" ) @pytest.fixture def mock_auth() -> AuthTokens: return AuthTokens( cookies={"SID": "test"}, csrf_token="csrf", session_id="session", ) def test_client_exposes_artifacts_and_notes(mock_auth: AuthTokens) -> None: """The client should construct both APIs regardless of order.""" client = NotebookLMClient(mock_auth) assert isinstance(client.artifacts, ArtifactsAPI) assert isinstance(client.notes, NotesAPI) def test_artifacts_constructible_without_notes_api(mock_auth: AuthTokens) -> None: """``ArtifactsAPI`` no longer takes ``notes_api`` at all (per docs/refactor-history.md Step 4) — the parameter was removed in favor of explicit ``mind_maps`` + ``note_service`` (Phase 5). The mind-map decoupling is now structural.""" from notebooklm._mind_map import NoteBackedMindMapService from notebooklm._note_service import NoteService core = MagicMock() api = ArtifactsAPI( rpc=core, drain=core, lifecycle=core, notebooks=MagicMock(), mind_maps=MagicMock(spec=NoteBackedMindMapService), note_service=MagicMock(spec=NoteService), ) assert api is not None # The legacy private attribute must not leak back: code that depends on # ``self._notes`` would re-introduce the coupling. assert not hasattr(api, "_notes") def test_artifacts_rejects_legacy_notes_api_kwarg(mock_auth: AuthTokens) -> None: """The legacy ``notes_api=`` kwarg was removed in Phase 3 (docs/refactor-history.md Step 4). Passing it must raise ``TypeError``.""" from notebooklm._mind_map import NoteBackedMindMapService from notebooklm._note_service import NoteService core = MagicMock() notes = NotesAPI( notes=MagicMock(spec=NoteService), mind_maps=MagicMock(spec=NoteBackedMindMapService), ) with pytest.raises(TypeError): ArtifactsAPI( # type: ignore[call-arg] core, notes_api=notes, notebooks=MagicMock(), mind_maps=MagicMock(spec=NoteBackedMindMapService), note_service=MagicMock(spec=NoteService), ) def test_artifacts_before_notes_construction_order(mock_auth: AuthTokens) -> None: """Both construction orders must succeed and produce working APIs. ``ArtifactsAPI`` and ``NotesAPI`` have no construction-order dependency on each other; this test pins that building either one first still yields working APIs. """ from notebooklm._mind_map import NoteBackedMindMapService from notebooklm._note_service import NoteService core = MagicMock() def _make_artifacts() -> ArtifactsAPI: return ArtifactsAPI( rpc=core, drain=core, lifecycle=core, notebooks=MagicMock(), mind_maps=MagicMock(spec=NoteBackedMindMapService), note_service=MagicMock(spec=NoteService), ) def _make_notes() -> NotesAPI: return NotesAPI( notes=MagicMock(spec=NoteService), mind_maps=MagicMock(spec=NoteBackedMindMapService), ) artifacts_first = _make_artifacts() notes_first = _make_notes() # Build in the opposite order too, just to make the symmetry explicit. notes_then = _make_notes() artifacts_then = _make_artifacts() assert artifacts_first is not None assert notes_first is not None assert artifacts_then is not None assert notes_then is not None # --------------------------------------------------------------------------- # Mind-map regression — ``generate_mind_map`` + ``list`` + ``download_mind_map`` # must keep working without an explicit ``NotesAPI`` injection. # --------------------------------------------------------------------------- def _make_core_for_mind_map_flow() -> tuple[FakeSession, list[tuple[Any, Any]]]: """Build a :class:`FakeSession` core whose ``rpc_call`` returns canned mind-map responses keyed on the RPC method. The core is built via ``make_fake_core(rpc_call=AsyncMock(...))`` — the sanctioned constructor-injection substrate (ADR-0007). The factory wires the injected mock onto ``fake.rpc_executor.rpc_call`` (the ``RpcCaller`` surface the mind-map flow threads into ``ArtifactsAPI``) and supplies benign defaults for the ``assert_bound_loop`` / ``operation_scope`` / ``register_drain_hook`` surfaces the artifacts runtime touches. Returns ``(core, calls)`` where ``calls`` is a list of ``(method, params)`` tuples populated as the test exercises the API. """ calls: list[tuple[Any, Any]] = [] mind_map_payload = { "name": "Mind Map Title", "children": [{"name": "child"}], } mind_map_json = json.dumps(mind_map_payload) async def fake_rpc_call(method: Any, params: Any, **_: Any) -> Any: calls.append((method, params)) name = getattr(method, "name", str(method)) if name == "GENERATE_MIND_MAP": return [[mind_map_json]] if name == "CREATE_NOTE": return [["note_abc"]] if name == "UPDATE_NOTE": return None if name == "GET_NOTES_AND_MIND_MAPS": return [ [ [ "note_abc", ["note_abc", mind_map_json, [], None, "Mind Map Title"], ] ] ] if name == "LIST_ARTIFACTS": return [[]] return None core = make_fake_core(rpc_call=AsyncMock(side_effect=fake_rpc_call)) return core, calls def _build_artifacts_with_real_mind_map_service(core: FakeSession) -> ArtifactsAPI: """Build an ``ArtifactsAPI`` whose mind-map services are real instances backed by ``core.rpc_executor`` so the mind-map flow exercises the live RPC callbacks against the canned executor. """ from notebooklm._mind_map import NoteBackedMindMapService from notebooklm._note_service import NoteService note_service = NoteService(core.rpc_executor) mind_maps = NoteBackedMindMapService(note_service) return ArtifactsAPI( rpc=core.rpc_executor, drain=core, lifecycle=core, notebooks=MagicMock(get_source_ids=AsyncMock(return_value=["src_1"])), mind_maps=mind_maps, note_service=note_service, ) @pytest.mark.asyncio async def test_generate_mind_map_works_without_notes_injection() -> None: """``generate_mind_map`` must persist the mind map via ``_mind_map`` primitives, not via an injected ``NotesAPI``.""" core, calls = _make_core_for_mind_map_flow() api = _build_artifacts_with_real_mind_map_service(core) result = await api.generate_mind_map("nb_123", source_ids=["src_1"]) assert result.note_id == "note_abc" assert result.mind_map["name"] == "Mind Map Title" # The constructor-injected RPC mock was actually exercised (ADR-0007 # Form-1 bite-check: the injected collaborator is reached, not silently # bypassed by a stale auto-vivified attribute). core.rpc_executor.rpc_call.assert_awaited() # The flow must have gone GENERATE_MIND_MAP -> CREATE_NOTE -> UPDATE_NOTE method_names = [getattr(m, "name", str(m)) for m, _ in calls] assert "GENERATE_MIND_MAP" in method_names assert "CREATE_NOTE" in method_names assert "UPDATE_NOTE" in method_names @pytest.mark.asyncio async def test_artifacts_list_pulls_mind_maps_without_notes_injection( tmp_path: Any, ) -> None: """``ArtifactsAPI.list`` must read mind maps through ``_mind_map`` — no ``NotesAPI`` reference required.""" core, _ = _make_core_for_mind_map_flow() api = _build_artifacts_with_real_mind_map_service(core) artifacts = await api.list("nb_123") # One mind map should surface from GET_NOTES_AND_MIND_MAPS. assert any(a.kind.name == "MIND_MAP" for a in artifacts) @pytest.mark.asyncio async def test_download_mind_map_works_without_notes_injection( tmp_path: Any, ) -> None: """``download_mind_map`` reaches into mind-map storage via ``_mind_map`` rather than ``self._notes``.""" core, _ = _make_core_for_mind_map_flow() api = _build_artifacts_with_real_mind_map_service(core) output = tmp_path / "mm.json" returned = await api.download_mind_map("nb_123", str(output)) assert returned == str(output) saved = json.loads(output.read_text(encoding="utf-8")) assert saved["name"] == "Mind Map Title"