"""AST lint: enforce the CLI -> client/types boundary. Block-list rules applied to every ``src/notebooklm/cli/**/*.py`` file: 1. No imports of private modules anywhere in the ``notebooklm`` tree: any path segment that starts with a single underscore (and is not a dunder) is rejected. Catches ``from notebooklm._foo import ...``, ``from notebooklm.pkg._bar import ...``, ``from .._foo import ...``, ``from notebooklm import _foo``, ``from .. import _foo``, etc. 2. No imports from the RPC layer: ``from notebooklm.rpc`` / ``from notebooklm.rpc.`` / ``from ..rpc`` / ``from ..rpc.`` / ``import notebooklm.rpc`` / ``from .. import rpc`` are all rejected. The CLI must consume RPC enums via the public ``notebooklm.types`` re-export. 3. No private-name leakage from a public module: ``from notebooklm. import _symbol`` / ``from .. import _symbol`` is rejected when no segment of the source path is itself underscored. This stops the CLI from reaching into a public module's internals (e.g. ``from notebooklm.auth import _internal_helper``). Dunders (``__version__``) remain allowed. Allowed: - Intra-cli imports (level == 1): ``from ._encoding import ...``, including underscored siblings — those are the CLI's own private modules. - Imports of non-underscored siblings/parents: ``from ..types import ...``, ``from ..research import ...``, etc. """ from __future__ import annotations import ast import pathlib from collections.abc import Iterator import pytest CLI_ROOT = pathlib.Path(__file__).resolve().parents[2] / "src" / "notebooklm" / "cli" HELPERS_PATH = CLI_ROOT / "helpers.py" OPTIONS_PATH = CLI_ROOT / "options.py" SERVICES_ROOT = CLI_ROOT / "services" RENDERING_PATH = CLI_ROOT / "rendering.py" CONTEXT_PATH = CLI_ROOT / "context.py" RUNTIME_PATH = CLI_ROOT / "runtime.py" AUTH_RUNTIME_PATH = CLI_ROOT / "auth_runtime.py" RESOLVE_PATH = CLI_ROOT / "resolve.py" COMPLETION_CALLBACKS = { "_complete_artifacts", "_complete_notebooks", "_complete_sources", "_resolve_notebook_for_completion", } COMPLETION_FORBIDDEN_SYMBOLS = { "NotebookLMClient", "get_auth_tokens", "run_async", } FUNCTION_DEF_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef) BLOCK_DEF_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef) CLI_COMMAND_MODULES = { "agent", "artifact", "chat", "doctor", "download", "generate", "label", "language", "note", "notebook", "profile", "research", "session", "share", "skill", "source", } RENDERING_FORBIDDEN_MODULES = CLI_COMMAND_MODULES | { "auth_runtime", "completion", "context", "input", "resolve", "runtime", } CONTEXT_FORBIDDEN_MODULES = CLI_COMMAND_MODULES | { "auth_runtime", "completion", "input", "rendering", "resolve", "runtime", } AUTH_RUNTIME_ALLOWED_MODULES = {"error_handler", "helpers", "services"} # Within ``services``, only the AuthSource resolver is allowed — auth_runtime # is not a Click handler and must not reach into command-rendering services. # rev-1 CodeRabbit feedback on #962 (#auth_runtime_imports tightening). AUTH_RUNTIME_ALLOWED_SERVICE_MODULES = {"notebooklm.cli.services.auth_source"} RESOLVE_FORBIDDEN_MODULES = CLI_COMMAND_MODULES | { "auth_runtime", "completion", "helpers", "runtime", } HELPERS_IMPORT_ALLOWED_FILES = { "__init__.py", # compatibility re-export surface "auth_runtime.py", # call-time patch seam for auth/runtime wrappers "completion.py", # call-time patch seam for completion callbacks } HELPERS_FACADE_ALLOWED_DEFS = { "_current_storage_override", "_display_cited_import_selection", "_get_context_value", "_resolve_partial_id", "_set_context_value", "build_cookie_jar", "clear_context", "cli_name_to_artifact_type", "display_report", "display_research_sources", "emit_status", "get_artifact_type_display", "get_auth_tokens", "get_client", "get_current_conversation", "get_current_notebook", "get_source_type_display", "handle_auth_error", "handle_error", "import_research_sources", "import_with_retry", "json_error_response", "json_output_response", "load_auth_from_storage", "read_stdin_text", "require_notebook", "resolve_artifact_id", "resolve_note_id", "resolve_notebook_id", "resolve_prompt", "resolve_source_id", "resolve_source_ids", "run_async", "set_current_conversation", "set_current_notebook", "validate_id", "with_auth_and_errors", "with_client", } def _is_private_segment(seg: str) -> bool: """True if ``seg`` is a single-underscore-prefixed name (not a dunder). Empty strings and dunders (``__version__``) are not private. """ return bool(seg) and seg.startswith("_") and not seg.startswith("__") def _is_dunder_name(name: str) -> bool: """True for double-underscore names that are intentionally public.""" return name.startswith("__") and name.endswith("__") def _has_private_segment(parts: list[str]) -> bool: """True if any segment in ``parts`` is private (per Rule 1).""" return any(_is_private_segment(p) for p in parts) def _is_rpc_path(parts: list[str]) -> bool: """True if ``parts`` is the RPC layer or a sub-path (per Rule 2). ``parts`` is the path *below* the ``notebooklm`` prefix, e.g. ``["rpc"]`` or ``["rpc", "types"]``. """ return bool(parts) and parts[0] == "rpc" def _is_browser_capture_path(parts: list[str]) -> bool: """True if ``parts`` targets the ``notebooklm._auth.browser_capture`` core. The transport-neutral browser launch -> capture -> filter -> persist primitive (``_auth/browser_capture.py``) is a **sanctioned** exception to the no-private-module rule, in the same spirit as ``_app``: it is the neutral core the CLI Playwright-login adapter (``cli/services/playwright_login.py``) sits over (ADR-0021 keeps the interactive presentation in ``cli/`` while the launch/capture/persist core moves down to ``_auth``, reachable by the client runtime and the future headless re-auth layer). The CLI adapter is allowed to import this single module even though the leading ``_auth`` would otherwise flag it as private. Only this one module is exempted — the rest of ``_auth.*`` stays behind the ``auth.py`` facade. ``parts`` is the path *below* the ``notebooklm`` prefix, e.g. ``["_auth", "browser_capture"]`` (matches the dotted-module forms ``from notebooklm._auth.browser_capture import …`` and ``from ..._auth.browser_capture import …``). """ return parts[:2] == ["_auth", "browser_capture"] def _is_browser_capture_alias_import(parent_parts: list[str], names: list[ast.alias]) -> bool: """True for the ``from ._auth import browser_capture`` shape. Complements :func:`_is_browser_capture_path` (which matches the dotted-module forms) so the sanctioned module stays exempt across *every* import shape: ``from notebooklm._auth import browser_capture`` and ``from .._auth import browser_capture`` resolve the module via the imported *name* rather than the module path, so they are checked here. ``parent_parts`` is the package path below ``notebooklm`` (e.g. ``["_auth"]``); the exemption only applies when every imported name is exactly ``browser_capture``. """ return parent_parts == ["_auth"] and all(alias.name == "browser_capture" for alias in names) # The ONLY ``_auth.headless_reauth`` names the CLI may import. Deliberately # narrow: the credential-free, browser-free readiness probe + its typed return, # consumed by the ``doctor`` diagnostic. The layer-3 *drive* path # (``attempt_headless_reauth`` and friends) is wired through the client runtime, # NOT the CLI, and stays behind the boundary so the CLI can never become an # auth-minting surface. Keep this set minimal. _HEADLESS_REAUTH_ALLOWED_NAMES = frozenset({"headless_reauth_readiness", "HeadlessReauthReadiness"}) def _is_headless_reauth_symbol_import(parts: list[str], names: list[ast.alias]) -> bool: """True ONLY for ``from ..._auth.headless_reauth import ``. A **second, deliberately name-level** sanctioned exception alongside :func:`_is_browser_capture_path`, in the same spirit as ``_app`` and the browser-capture core — but narrower. ``_auth/headless_reauth.py`` owns the layer-3 headless re-auth feature; the CLI ``doctor`` command consumes its ``headless_reauth_readiness()`` probe — a **credential-free, browser-free** readiness snapshot (profile present + playwright installed). ``doctor`` cannot route this through ``_app`` because the ``_app`` boundary forbids ``_app`` from importing ``_auth`` (the probe lives in ``_auth``), so the adapter imports the symbol directly. Unlike the module-level ``browser_capture`` carve-out, this exemption is keyed on the imported *names*: ONLY :data:`_HEADLESS_REAUTH_ALLOWED_NAMES` pass. A module-form import (``import notebooklm._auth.headless_reauth`` / ``from .._auth import headless_reauth``) binds the whole module — including the L3 drive path ``attempt_headless_reauth`` — so it is deliberately NOT exempted and stays a boundary violation. ``parts`` is the path *below* the ``notebooklm`` prefix, i.e. exactly ``["_auth", "headless_reauth"]`` for the ``from`` target. """ return parts == ["_auth", "headless_reauth"] and all( alias.name in _HEADLESS_REAUTH_ALLOWED_NAMES for alias in names ) def _is_app_path(parts: list[str]) -> bool: """True if ``parts`` targets the ``notebooklm._app`` business-logic layer. ``_app`` is the **sanctioned** exception to the no-private-module rule: it is the transport-neutral business-logic package every adapter (the CLI, the FastMCP server, future HTTP) is designed to consume (relocation plan §1/§2; boundary enforced the other way by ``tests/_guardrails/test_app_boundary.py``). The CLI is allowed to import ``notebooklm._app`` / ``notebooklm._app.`` even though the leading underscore would otherwise flag it as private. ``parts`` is the path *below* the ``notebooklm`` prefix, e.g. ``["_app"]`` or ``["_app", "download"]``. """ return bool(parts) and parts[0] == "_app" def _cli_module_imports(path: pathlib.Path) -> set[str]: """Return direct ``notebooklm.cli`` module imports used by a CLI file.""" tree = ast.parse(path.read_text(encoding="utf-8")) imports: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" mod_parts = mod.split(".") if mod else [] if node.level == 1: if mod: imports.add(mod_parts[0]) else: imports.update(alias.name.split(".")[0] for alias in node.names) elif node.level >= 2 and mod_parts[:1] == ["cli"]: if len(mod_parts) > 1: imports.add(mod_parts[1]) else: imports.update(alias.name.split(".")[0] for alias in node.names) elif node.level == 0 and mod_parts[:2] == ["notebooklm", "cli"]: if len(mod_parts) > 2: imports.add(mod_parts[2]) elif len(mod_parts) == 2: imports.update(alias.name.split(".")[0] for alias in node.names) elif isinstance(node, ast.Import): for alias in node.names: parts = alias.name.split(".") if parts[:2] == ["notebooklm", "cli"] and len(parts) > 2: imports.add(parts[2]) return imports def _imports_notebooklm_auth(tree: ast.AST) -> list[str]: """Return imports that bind CLI code directly to notebooklm.auth.""" offenders: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" mod_parts = mod.split(".") if mod else [] if node.level == 0: if mod_parts[:2] == ["notebooklm", "auth"]: offenders.append(f"from {mod} import ...") elif mod_parts == ["notebooklm"]: offenders.extend( f"from notebooklm import {alias.name}" for alias in node.names if alias.name == "auth" ) elif node.level >= 2: if mod_parts[:1] == ["auth"]: offenders.append(f"from {'.' * node.level}{mod} import ...") elif not mod: offenders.extend( f"from {'.' * node.level} import {alias.name}" for alias in node.names if alias.name == "auth" ) elif isinstance(node, ast.Import): for alias in node.names: parts = alias.name.split(".") if parts[:2] == ["notebooklm", "auth"]: offenders.append(f"import {alias.name}") return offenders def _imports_helpers_facade(tree: ast.AST, relative_parts: tuple[str, ...]) -> list[str]: """Return imports that bind a CLI module back to ``cli.helpers``.""" offenders: list[str] = [] cli_package_level = len(relative_parts) for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" mod_parts = mod.split(".") if mod else [] if node.level > 0: if node.level == cli_package_level and mod_parts[:1] == ["helpers"]: offenders.append(f"from {'.' * node.level}{mod} import ...") elif node.level == cli_package_level and not mod: offenders.extend( f"from {'.' * node.level} import {alias.name}" for alias in node.names if alias.name == "helpers" ) elif node.level > cli_package_level and mod_parts[:1] == ["cli"]: if len(mod_parts) > 1 and mod_parts[1] == "helpers": offenders.append(f"from {'.' * node.level}{mod} import ...") elif len(mod_parts) == 1: offenders.extend( f"from {'.' * node.level}{mod} import {alias.name}" for alias in node.names if alias.name == "helpers" ) elif node.level == 0: if mod_parts[:3] == ["notebooklm", "cli", "helpers"]: offenders.append(f"from {mod} import ...") elif mod_parts[:2] == ["notebooklm", "cli"]: offenders.extend( f"from {mod} import {alias.name}" for alias in node.names if alias.name == "helpers" ) elif isinstance(node, ast.Import): for alias in node.names: if alias.name.split(".")[:3] == ["notebooklm", "cli", "helpers"]: offenders.append(f"import {alias.name}") return offenders @pytest.mark.parametrize( ("source", "expected"), [ ("from notebooklm.cli import completion\n", {"completion"}), ("from notebooklm.cli.runtime import with_client\n", {"runtime"}), ("from ..cli import rendering\n", {"rendering"}), ("from ..cli.context import get_current_notebook\n", {"context"}), ], ) def test_cli_module_imports_detects_cli_import_forms( tmp_path: pathlib.Path, source: str, expected: set[str] ) -> None: path = tmp_path / "sample.py" path.write_text(source, encoding="utf-8") assert _cli_module_imports(path) == expected @pytest.mark.parametrize( ("relative_parts", "source", "expected"), [ (("source.py",), "from .helpers import console\n", ["from .helpers import ..."]), (("source.py",), "from . import helpers\n", ["from . import helpers"]), ( ("services", "source.py"), "from ..helpers import console\n", ["from ..helpers import ..."], ), ( ("services", "source.py"), "from .. import helpers\n", ["from .. import helpers"], ), (("services", "source.py"), "from . import helpers\n", []), ( ("source.py",), "from notebooklm.cli.helpers import console\n", ["from notebooklm.cli.helpers import ..."], ), ( ("source.py",), "from notebooklm.cli import helpers\n", ["from notebooklm.cli import helpers"], ), ( ("source.py",), "import notebooklm.cli.helpers as helpers\n", ["import notebooklm.cli.helpers"], ), ], ) def test_imports_helpers_facade_detects_cli_helpers_forms( relative_parts: tuple[str, ...], source: str, expected: list[str] ) -> None: tree = ast.parse(source) assert _imports_helpers_facade(tree, relative_parts) == expected def _violations(tree: ast.AST) -> list[str]: # noqa: C901 - flat dispatch on import shape bad: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): mod = node.module or "" mod_parts = mod.split(".") if mod else [] if node.level == 0: # Absolute import: only inspect notebooklm.* roots. if mod_parts and mod_parts[0] == "notebooklm": if len(mod_parts) >= 2: sub_parts = mod_parts[1:] # ``notebooklm._app`` is the sanctioned shared layer; # ``notebooklm._auth.browser_capture`` is the sanctioned # neutral browser-capture core (ADR-0021); the # ``notebooklm._auth.headless_reauth`` readiness symbols # are sanctioned for the doctor diagnostic (name-level). if ( _is_app_path(sub_parts) or _is_browser_capture_path(sub_parts) or _is_browser_capture_alias_import(sub_parts, node.names) or _is_headless_reauth_symbol_import(sub_parts, node.names) ): continue # Rule 1 (any private segment) or Rule 2 (rpc layer). if _has_private_segment(sub_parts) or _is_rpc_path(sub_parts): bad.append(f"from {mod} import ...") else: # Rule 3: private-name leakage from a public module. for alias in node.names: if _is_private_segment(alias.name): bad.append(f"from {mod} import {alias.name}") else: # ``from notebooklm import X`` — inspect each name. # Rule 1 (private name) or Rule 2 (``rpc`` sub-package), # excluding the sanctioned ``_app`` package. for alias in node.names: if alias.name == "_app": continue if _is_private_segment(alias.name) or alias.name == "rpc": bad.append(f"from notebooklm import {alias.name}") elif node.level >= 2: # Relative parent-package import (cli reaches into notebooklm/*). if mod: # ``from .._app...`` / ``from ..._app...`` — sanctioned layer; # ``from ..._auth.browser_capture import …`` — sanctioned # neutral browser-capture core (ADR-0021); # ``from ..._auth.headless_reauth import `` # — sanctioned for the doctor diagnostic (name-level only). if ( _is_app_path(mod_parts) or _is_browser_capture_path(mod_parts) or _is_browser_capture_alias_import(mod_parts, node.names) or _is_headless_reauth_symbol_import(mod_parts, node.names) ): continue # Rule 1 (any private segment) or Rule 2 (rpc layer). if _has_private_segment(mod_parts) or _is_rpc_path(mod_parts): bad.append(f"from {'.' * node.level}{mod} import ...") continue # Rule 3: private-name leakage from a public source module. for alias in node.names: if _is_private_segment(alias.name): bad.append(f"from {'.' * node.level}{mod} import {alias.name}") else: # ``from .. import X`` — inspect each imported name. # Rule 1 (private name) or Rule 2 (``rpc`` sub-package), # excluding the sanctioned ``_app`` package. for alias in node.names: if alias.name == "_app": continue if _is_private_segment(alias.name) or alias.name == "rpc": bad.append(f"from {'.' * node.level} import {alias.name}") else: # level == 1 (intra-cli). Inspect ``from . import X`` only for # the explicit ``rpc`` name — siblings starting with ``_`` are # cli's own private modules and remain allowed. if not mod: for alias in node.names: if alias.name == "rpc": bad.append(f"from . import {alias.name}") elif isinstance(node, ast.Import): for alias in node.names: parts = alias.name.split(".") if not (len(parts) >= 2 and parts[0] == "notebooklm"): continue sub_parts = parts[1:] # ``import notebooklm._app[.x]`` is the sanctioned shared layer; # ``import notebooklm._auth.browser_capture`` is the sanctioned # neutral browser-capture core (ADR-0021). NOTE: a *module-form* # ``import notebooklm._auth.headless_reauth`` is deliberately NOT # exempted — it binds the whole module incl. the L3 drive path; # only name-level readiness-symbol imports are sanctioned. if _is_app_path(sub_parts) or _is_browser_capture_path(sub_parts): continue # Rule 1 (any private segment) or Rule 2 (rpc layer). if _has_private_segment(sub_parts) or _is_rpc_path(sub_parts): bad.append(f"import {alias.name}") return bad def _iter_block_body_nodes(block: ast.AST) -> Iterator[ast.AST]: """Yield nodes from a block signature/body without descending into nested blocks.""" def walk(node: ast.AST, *, is_root: bool = False) -> Iterator[ast.AST]: if not is_root and isinstance(node, BLOCK_DEF_TYPES): return if not is_root: yield node for child in ast.iter_child_nodes(node): yield from walk(child) yield from walk(block, is_root=True) def _has_forbidden_completion_boundary(parts: list[str]) -> bool: """Match forbidden names on exact dotted prefixes or final symbol names.""" dotted_matches = ( ".".join(parts[:index]) in COMPLETION_FORBIDDEN_SYMBOLS for index in range(1, len(parts) + 1) ) return any(dotted_matches) or bool(parts and parts[-1] in COMPLETION_FORBIDDEN_SYMBOLS) def _completion_boundary_violations(tree: ast.AST) -> tuple[set[str], list[str]]: forbidden_names = set(COMPLETION_FORBIDDEN_SYMBOLS) import_offenders: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): module_parts = (node.module or "").split(".") if node.module else [] module_forbidden = _has_forbidden_completion_boundary(module_parts) for alias in node.names: if _is_dunder_name(alias.name): continue if not module_forbidden and alias.name not in COMPLETION_FORBIDDEN_SYMBOLS: continue alias_name = alias.asname or alias.name forbidden_names.add(alias_name) import_offenders.append(f"import: {alias.name}") elif isinstance(node, ast.Import): for alias in node.names: parts = alias.name.split(".") if any(_is_dunder_name(part) for part in parts): continue if not _has_forbidden_completion_boundary(parts): continue if alias.asname: forbidden_names.add(alias.asname) elif len(parts) == 1: forbidden_names.add(alias.name) import_offenders.append(f"import: {alias.name}") check_targets: list[tuple[str, ast.AST]] = [("", tree)] for node in ast.walk(tree): if isinstance(node, FUNCTION_DEF_TYPES): check_targets.append((node.name, node)) elif isinstance(node, ast.ClassDef): check_targets.append((f"class {node.name}", node)) top_level_functions = {node.name for node in tree.body if isinstance(node, FUNCTION_DEF_TYPES)} missing_callbacks = COMPLETION_CALLBACKS - top_level_functions offenders = list(import_offenders) for context_name, block_node in sorted(check_targets, key=lambda item: item[0]): for node in _iter_block_body_nodes(block_node): if isinstance(node, ast.Name) and node.id in forbidden_names: offenders.append(f"{context_name}: {node.id}") elif ( isinstance(node, ast.Attribute) and node.attr in COMPLETION_FORBIDDEN_SYMBOLS and not _is_dunder_name(node.attr) ): offenders.append(f"{context_name}: .{node.attr}") return missing_callbacks, offenders def test_no_private_module_imports_in_cli(): offenders: list[tuple[str, list[str]]] = [] for path in sorted(CLI_ROOT.rglob("*.py")): tree = ast.parse(path.read_text(encoding="utf-8")) bad = _violations(tree) if bad: offenders.append((str(path.relative_to(CLI_ROOT.parent)), bad)) assert not offenders, ( "CLI must not import notebooklm._* (private modules), notebooklm.rpc.*, " "or `_private` names out of public notebooklm modules. " "Promote needed symbols to a public module (config/urls/log/research/types) " f"and import from there.\nOffenders: {offenders}" ) def test_cli_services_stay_on_public_library_boundary() -> None: """Keep service-layer modules from binding directly to private library/RPC APIs.""" offenders: list[tuple[str, list[str]]] = [] for path in sorted(SERVICES_ROOT.rglob("*.py")): tree = ast.parse(path.read_text(encoding="utf-8")) bad = _violations(tree) if bad: offenders.append((str(path.relative_to(CLI_ROOT.parent)), bad)) assert not offenders, ( "notebooklm.cli.services.* must not import notebooklm._* private modules, " "notebooklm.rpc.*, or `_private` names from public notebooklm modules. " "Route service collaborators through public library APIs or CLI facades.\n" f"Offenders: {offenders}" ) def test_rendering_stays_on_low_level_cli_import_boundary() -> None: imports = _cli_module_imports(RENDERING_PATH) assert not (imports & RENDERING_FORBIDDEN_MODULES), ( "cli.rendering must not import runtime/auth/context/resolve/input/completion " f"or command modules. Offenders: {sorted(imports & RENDERING_FORBIDDEN_MODULES)}" ) def test_context_stays_on_low_level_cli_import_boundary() -> None: imports = _cli_module_imports(CONTEXT_PATH) assert not (imports & CONTEXT_FORBIDDEN_MODULES), ( "cli.context must not import runtime/auth/rendering/resolve/input/completion " "or command modules. " f"Offenders: {sorted(imports & CONTEXT_FORBIDDEN_MODULES)}" ) def test_runtime_stays_leaf_module() -> None: imports = _cli_module_imports(RUNTIME_PATH) assert imports == set(), f"cli.runtime must not import other cli modules. Offenders: {imports}" def test_auth_runtime_imports_only_runtime_facade_collaborators() -> None: imports = _cli_module_imports(AUTH_RUNTIME_PATH) assert imports <= AUTH_RUNTIME_ALLOWED_MODULES, ( "cli.auth_runtime must not import command, rendering, context, resolve, " "input, or completion modules directly. The ``services`` subpackage is " "allowed because P3.T3 consolidated the auth-source precedence chain into " ":class:`notebooklm.cli.services.auth_source.AuthSource`. " f"Offenders: {sorted(imports - AUTH_RUNTIME_ALLOWED_MODULES)}" ) # Tighter check (rev-1 CodeRabbit feedback on #962): within ``services``, # only ``services.auth_source`` is on the allowlist. Reaching into command # rendering services from auth_runtime would re-create the layering bug # P3.T3 fixed. tree = ast.parse(AUTH_RUNTIME_PATH.read_text(encoding="utf-8")) service_imports = { node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and (node.module or "").startswith("notebooklm.cli.services") } # Also catch absolute ``import notebooklm.cli.services.X`` forms # (rev-2 CodeRabbit feedback on #962 — without this branch a bare # ``import`` would silently bypass the layering guard). for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name.startswith("notebooklm.cli.services"): service_imports.add(alias.name) # Also catch relative imports of the form ``from .services.X import ...`` for node in ast.walk(tree): if ( isinstance(node, ast.ImportFrom) and node.level == 1 and (node.module or "").startswith("services") ): service_imports.add("notebooklm.cli." + (node.module or "")) assert service_imports <= AUTH_RUNTIME_ALLOWED_SERVICE_MODULES, ( "cli.auth_runtime may only import from ``notebooklm.cli.services.auth_source`` " "(the AuthSource resolver). Other service modules are command-rendering " "collaborators and must stay out of the auth-runtime layer. " f"Offenders: {sorted(service_imports - AUTH_RUNTIME_ALLOWED_SERVICE_MODULES)}" ) def test_resolve_stays_off_helpers_runtime_auth_and_commands() -> None: """Partial-ID resolution stays below helpers/runtime/auth and command modules.""" imports = _cli_module_imports(RESOLVE_PATH) auth_imports = _imports_notebooklm_auth(ast.parse(RESOLVE_PATH.read_text(encoding="utf-8"))) assert not (imports & RESOLVE_FORBIDDEN_MODULES), ( "cli.resolve must not import cli.helpers, cli.runtime/auth_runtime, " f"or command modules. Offenders: {sorted(imports & RESOLVE_FORBIDDEN_MODULES)}" ) assert not auth_imports, ( "cli.resolve must not import notebooklm.auth; keep auth/runtime work outside " f"the resolver layer. Offenders: {auth_imports}" ) def test_command_modules_do_not_import_helpers_facade_for_moved_symbols() -> None: """Production CLI modules import moved helpers from their owning modules.""" offenders: list[tuple[str, list[str]]] = [] for path in sorted(CLI_ROOT.rglob("*.py")): if path.name == "helpers.py": continue relative = str(path.relative_to(CLI_ROOT)) tree = ast.parse(path.read_text(encoding="utf-8")) helper_imports = _imports_helpers_facade(tree, tuple(path.relative_to(CLI_ROOT).parts)) if helper_imports and relative not in HELPERS_IMPORT_ALLOWED_FILES: offenders.append((relative, helper_imports)) assert not offenders, ( "Production CLI modules must not import moved symbols from cli.helpers. " "Import rendering/context/runtime/auth/resolve/input/research helpers from " "their owning modules instead. Only the compatibility export surface and " "documented call-time patch seams may bind to cli.helpers.\n" f"Offenders: {offenders}" ) def test_helpers_remains_compatibility_facade() -> None: """Keep ``cli.helpers`` from silently regaining broad command responsibilities.""" tree = ast.parse(HELPERS_PATH.read_text(encoding="utf-8")) defs = {node.name for node in tree.body if isinstance(node, BLOCK_DEF_TYPES)} imports = _cli_module_imports(HELPERS_PATH) assert defs <= HELPERS_FACADE_ALLOWED_DEFS, ( "cli.helpers should stay a compatibility facade over moved helper modules. " "Add compatibility re-exports to HELPERS_FACADE_ALLOWED_DEFS only after " "the implementation lives in an owning module. " f"Unexpected defs: {sorted(defs - HELPERS_FACADE_ALLOWED_DEFS)}" ) assert not (imports & CLI_COMMAND_MODULES), ( "cli.helpers must not import command modules; keep it below commands. " f"Offenders: {sorted(imports & CLI_COMMAND_MODULES)}" ) def test_options_completion_callbacks_stay_on_completion_provider_boundary() -> None: """Keep live completion auth/client/runtime work out of ``cli.options``.""" tree = ast.parse(OPTIONS_PATH.read_text(encoding="utf-8")) missing_callbacks, offenders = _completion_boundary_violations(tree) assert not missing_callbacks, ( "Expected top-level completion callbacks missing from cli.options: " f"{sorted(missing_callbacks)}" ) assert not offenders, ( "cli.options must delegate completion live auth/client/runtime work to " "cli.completion instead of constructing clients, loading auth, or running " f"async work directly: {offenders}" ) @pytest.mark.parametrize( ("source", "expected"), [ ( "from notebooklm import NotebookLMClient as Client\n" "def _complete_artifacts():\n" " return Client\n", ["import: NotebookLMClient", "_complete_artifacts: Client"], ), ( "import run_async as runner\ndef _complete_notebooks():\n return runner\n", ["import: run_async", "_complete_notebooks: runner"], ), ( "import pkg.run_async\ndef _complete_sources():\n return None\n", ["import: pkg.run_async"], ), ( "from .runtime import run_async as runner\n" "def _complete_sources():\n" " return runner\n", ["import: run_async", "_complete_sources: runner"], ), ( "class Provider:\n" " client = NotebookLMClient\n" "def _resolve_notebook_for_completion():\n" " return None\n", ["class Provider: NotebookLMClient"], ), ( "@run_async\n" "def _complete_artifacts(client: NotebookLMClient = None) -> NotebookLMClient:\n" " return client\n", [ "_complete_artifacts: run_async", "_complete_artifacts: NotebookLMClient", ], ), ( "class Provider(NotebookLMClient):\n" " pass\n" "def _resolve_notebook_for_completion():\n" " return None\n", ["class Provider: NotebookLMClient"], ), ], ) def test_completion_boundary_detects_import_and_block_shapes( source: str, expected: list[str] ) -> None: """Self-check the AST guardrail paths used by the live options.py test.""" _, offenders = _completion_boundary_violations(ast.parse(source)) assert set(offenders) == set(expected), f"Expected {expected}, got {offenders}" def test_completion_boundary_reports_missing_callbacks() -> None: missing_callbacks, _ = _completion_boundary_violations(ast.parse("")) assert missing_callbacks == COMPLETION_CALLBACKS def test_completion_boundary_ignores_dunder_imports() -> None: _, offenders = _completion_boundary_violations( ast.parse("from notebooklm import __version__\n") ) assert offenders == [] def test_completion_boundary_allows_standard_library_and_safe_relative_imports() -> None: _, offenders = _completion_boundary_violations( ast.parse("import pathlib\nfrom collections import abc\nfrom .completion import complete\n") ) assert offenders == [] @pytest.mark.parametrize( ("source", "expected"), [ ( "from notebooklm._auth.tokens import AuthTokens", "from notebooklm._auth.tokens import ...", ), ("import notebooklm._auth.tokens", "import notebooklm._auth.tokens"), ("from .._auth.tokens import AuthTokens", "from .._auth.tokens import ..."), ("from .. import _auth", "from .. import _auth"), ( "from notebooklm._types.sources import Source", "from notebooklm._types.sources import ...", ), ("import notebooklm._types.sources", "import notebooklm._types.sources"), ("from notebooklm import _types", "from notebooklm import _types"), ("from .._types.sources import Source", "from .._types.sources import ..."), ("from .. import _types", "from .. import _types"), ], ) def test_cli_boundary_blocks_private_project_import_shapes( source: str, expected: str, ) -> None: """CLI imports must stay on public notebooklm modules, including moved _types.""" assert expected in _violations(ast.parse(source)) @pytest.mark.parametrize( "source", [ "from notebooklm._app import to_jsonable\n", "from notebooklm._app.download import build_download_plan\n", "from notebooklm._app.events import ProgressSink\n", "from .._app import to_jsonable\n", "from .._app.download import execute_download\n", "from ..._app.download import execute_download\n", "from notebooklm import _app\n", "from .. import _app\n", "from ... import _app\n", "import notebooklm._app\n", "import notebooklm._app.download\n", ], ) def test_cli_boundary_allows_sanctioned_app_layer_imports(source: str) -> None: """``notebooklm._app`` is the shared business-logic layer adapters consume. Every import shape that targets ``_app`` must be allowed even though the leading underscore would otherwise flag it private — this is the seam the CLI/MCP/HTTP adapters are built on (relocation plan §1/§2). """ assert _violations(ast.parse(source)) == [] @pytest.mark.parametrize( ("source", "expected"), [ # A different private package next to ``_app`` is still blocked. ("from notebooklm._appendix import x\n", "from notebooklm._appendix import ..."), ("from .._appendix import x\n", "from .._appendix import ..."), ("from notebooklm import _appendix\n", "from notebooklm import _appendix"), ], ) def test_cli_boundary_app_allowlist_is_exact_not_prefix(source: str, expected: str) -> None: """The ``_app`` allowlist must not leak to other ``_app``-prefixed packages.""" assert expected in _violations(ast.parse(source)) @pytest.mark.parametrize( "source", [ "from notebooklm._auth.browser_capture import run_browser_capture\n", "from .._auth.browser_capture import run_browser_capture\n", "from ..._auth.browser_capture import run_browser_capture\n", "from notebooklm._auth import browser_capture\n", "from .._auth import browser_capture\n", "from ..._auth import browser_capture\n", "import notebooklm._auth.browser_capture\n", ], ) def test_cli_boundary_allows_sanctioned_browser_capture_imports(source: str) -> None: """``notebooklm._auth.browser_capture`` is the sanctioned neutral capture core. Every import shape that targets this single module must be allowed even though the leading ``_auth`` would otherwise flag it private — it is the transport-neutral launch/capture/persist core the CLI Playwright-login adapter sits over (ADR-0021). The rest of ``_auth.*`` stays behind the ``auth.py`` facade. """ assert _violations(ast.parse(source)) == [] @pytest.mark.parametrize( ("source", "expected"), [ # Sibling ``_auth`` modules are NOT sanctioned — only browser_capture is. ( "from notebooklm._auth.cookie_policy import build_cookie_domain_allowlist\n", "from notebooklm._auth.cookie_policy import ...", ), ("from .._auth.tokens import AuthTokens\n", "from .._auth.tokens import ..."), ("from notebooklm._auth import tokens\n", "from notebooklm._auth import ..."), ("from .._auth import storage\n", "from .._auth import ..."), ("import notebooklm._auth.cookie_policy\n", "import notebooklm._auth.cookie_policy"), # ``browser_capture`` alongside a sibling name is not the sanctioned shape # (the exemption requires every imported name to be browser_capture). ( "from notebooklm._auth import browser_capture, tokens\n", "from notebooklm._auth import ...", ), ], ) def test_cli_boundary_browser_capture_allowlist_is_exact_not_prefix( source: str, expected: str ) -> None: """The browser_capture exemption must not leak to other ``_auth.*`` modules.""" assert expected in _violations(ast.parse(source)) @pytest.mark.parametrize( "source", [ "from notebooklm._auth.headless_reauth import headless_reauth_readiness\n", "from .._auth.headless_reauth import headless_reauth_readiness\n", "from ..._auth.headless_reauth import headless_reauth_readiness\n", "from notebooklm._auth.headless_reauth import HeadlessReauthReadiness\n", "from .._auth.headless_reauth import (\n" " headless_reauth_readiness,\n" " HeadlessReauthReadiness,\n" ")\n", ], ) def test_cli_boundary_allows_headless_reauth_readiness_symbols(source: str) -> None: """The CLI ``doctor`` may import the credential-free L3 readiness probe. ONLY the readiness symbols (``headless_reauth_readiness`` / ``HeadlessReauthReadiness``) are sanctioned, across the ``from``-import shapes. The L3 *drive* path stays behind the boundary (covered below). """ assert _violations(ast.parse(source)) == [] @pytest.mark.parametrize( ("source", "expected"), [ # The L3 *drive* path must stay blocked — this is the security boundary. ( "from notebooklm._auth.headless_reauth import attempt_headless_reauth\n", "from notebooklm._auth.headless_reauth import ...", ), ( "from .._auth.headless_reauth import attempt_headless_reauth\n", "from .._auth.headless_reauth import ...", ), # A readiness symbol imported ALONGSIDE the drive path is not sanctioned # (the exemption requires EVERY imported name to be readiness-only). ( "from notebooklm._auth.headless_reauth import " "headless_reauth_readiness, attempt_headless_reauth\n", "from notebooklm._auth.headless_reauth import ...", ), # Module-form imports bind the whole module (incl. the drive path) and # are deliberately NOT exempted. ( "import notebooklm._auth.headless_reauth\n", "import notebooklm._auth.headless_reauth", ), ("from notebooklm._auth import headless_reauth\n", "from notebooklm._auth import ..."), ("from .._auth import headless_reauth\n", "from .._auth import ..."), ], ) def test_cli_boundary_headless_reauth_carveout_blocks_drive_path( source: str, expected: str ) -> None: """The headless_reauth carve-out is readiness-only; the L3 drive stays blocked.""" assert expected in _violations(ast.parse(source))