Files
wehub-resource-sync 09e9f3545f
Test / Code Quality (push) Has been cancelled
Test / Test (macos-latest, Python 3.10) (push) Has been cancelled
Test / Test (macos-latest, Python 3.11) (push) Has been cancelled
Test / Test (macos-latest, Python 3.12) (push) Has been cancelled
Test / Test (macos-latest, Python 3.13) (push) Has been cancelled
Test / Test (macos-latest, Python 3.14) (push) Has been cancelled
Test / Test (ubuntu-latest, Python 3.10) (push) Has been cancelled
Test / Test (ubuntu-latest, Python 3.11) (push) Has been cancelled
Test / Test (ubuntu-latest, Python 3.12) (push) Has been cancelled
Test / Test (ubuntu-latest, Python 3.13) (push) Has been cancelled
Test / Test (ubuntu-latest, Python 3.14) (push) Has been cancelled
Test / Test (windows-latest, Python 3.10) (push) Has been cancelled
Test / Test (windows-latest, Python 3.11) (push) Has been cancelled
Test / Test (windows-latest, Python 3.12) (push) Has been cancelled
Test / Test (windows-latest, Python 3.13) (push) Has been cancelled
Test / Test (windows-latest, Python 3.14) (push) Has been cancelled
CodeQL / Analyze (push) Has been cancelled
dependency-audit / pip-audit (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:30:13 +08:00

1043 lines
44 KiB
Python

"""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.<x>`` / ``from ..rpc`` /
``from ..rpc.<x>`` / ``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.<public...> import _symbol`` /
``from ..<public...> 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 <pkg>._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 <readiness symbol>``.
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.<x>`` 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 <readiness symbol>``
# — 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]] = [("<module>", 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))