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
173 lines
5.8 KiB
Python
173 lines
5.8 KiB
Python
"""Unit tests for the headless arm of ``run_browser_capture`` (layer-3 re-auth).
|
|
|
|
Covers the landing-URL classifier and the no-human contract:
|
|
|
|
* authenticated landing (lands on the NotebookLM host) → capture / filter /
|
|
atomically persist ``storage_state.json`` (reusing the P1 path); NEVER waits
|
|
for a human.
|
|
* redirected to the Google login page → raise
|
|
:class:`HeadlessLoginRequiredError` loudly (the profile's session is also
|
|
dead); NEVER hangs on ``wait_for_url``.
|
|
* the ``(headless, interactive)`` mode guard: only the two sanctioned arms are
|
|
accepted.
|
|
|
|
The Playwright context is faked via ``patch("playwright.sync_api.sync_playwright")``
|
|
(the same shape the existing login-coverage suite uses), so no real browser /
|
|
network is required and ``playwright`` stays lazily imported.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from notebooklm._auth.browser_capture import (
|
|
BrowserCapturePlan,
|
|
_reject_unsupported_mode,
|
|
run_browser_capture,
|
|
)
|
|
from notebooklm.exceptions import HeadlessLoginRequiredError
|
|
|
|
|
|
class _RaisingCaptureIO:
|
|
"""``BrowserCaptureIO`` whose ``fail`` raises (mirrors the headless sink)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.emitted: list[Any] = []
|
|
|
|
def emit(self, *args: Any, **kwargs: Any) -> None:
|
|
self.emitted.append(args)
|
|
|
|
def fail(self, code: int) -> Any:
|
|
raise HeadlessLoginRequiredError(f"io.fail({code})")
|
|
|
|
def run_async(self, coro: Any) -> Any: # pragma: no cover - not reached
|
|
raise AssertionError("run_async not used in the headless arm")
|
|
|
|
|
|
class _FakeSyncPlaywright:
|
|
def __init__(self, playwright: Any) -> None:
|
|
self._playwright = playwright
|
|
|
|
def __enter__(self) -> Any:
|
|
return self._playwright
|
|
|
|
def __exit__(self, *exc: Any) -> bool:
|
|
return False
|
|
|
|
|
|
def _fake_playwright_landing(url: str, *, cookies: list[dict] | None = None) -> Any:
|
|
"""Build a fake playwright whose page lands on ``url``."""
|
|
page = MagicMock()
|
|
page.url = url
|
|
page.goto.return_value = None
|
|
page.content.return_value = "<html></html>"
|
|
context = MagicMock()
|
|
context.pages = [page]
|
|
context.storage_state.return_value = {
|
|
"cookies": cookies if cookies is not None else [],
|
|
"origins": [],
|
|
}
|
|
playwright = MagicMock()
|
|
playwright.chromium.launch_persistent_context.return_value = context
|
|
return playwright, context, page
|
|
|
|
|
|
def _run_headless(plan: BrowserCapturePlan, io: Any, playwright: Any) -> Any:
|
|
with patch(
|
|
"playwright.sync_api.sync_playwright",
|
|
side_effect=lambda: _FakeSyncPlaywright(playwright),
|
|
):
|
|
return run_browser_capture(plan, io, headless=True, interactive=False)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mode guard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mode_guard_accepts_both_sanctioned_arms() -> None:
|
|
io = _RaisingCaptureIO()
|
|
# interactive login arm
|
|
_reject_unsupported_mode(headless=False, interactive=True, io=io)
|
|
# headless re-auth arm
|
|
_reject_unsupported_mode(headless=True, interactive=False, io=io)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("headless", "interactive"),
|
|
[(True, True), (False, False)],
|
|
)
|
|
def test_mode_guard_rejects_contradictory_combos(headless: bool, interactive: bool) -> None:
|
|
with pytest.raises(NotImplementedError, match="Unsupported browser-capture mode"):
|
|
_reject_unsupported_mode(headless=headless, interactive=interactive, io=_RaisingCaptureIO())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Authenticated landing → capture + persist
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.requires_playwright
|
|
def test_headless_authenticated_landing_persists_storage(tmp_path: Path) -> None:
|
|
storage = tmp_path / "storage_state.json"
|
|
profile = tmp_path / "browser_profile"
|
|
profile.mkdir()
|
|
|
|
cookies = [
|
|
{"name": "SID", "value": "v", "domain": ".google.com", "path": "/"},
|
|
# A sibling-product cookie that the domain filter must DROP.
|
|
{"name": "X", "value": "y", "domain": "mail.google.com", "path": "/"},
|
|
]
|
|
playwright, _context, _page = _fake_playwright_landing(
|
|
"https://notebooklm.google.com/", cookies=cookies
|
|
)
|
|
io = _RaisingCaptureIO()
|
|
|
|
result = _run_headless(
|
|
BrowserCapturePlan(browser="chromium", browser_profile=profile, storage_path=storage),
|
|
io,
|
|
playwright,
|
|
)
|
|
|
|
# Persisted, and the domain allowlist filtered out the mail.google.com row.
|
|
assert storage.exists()
|
|
persisted = json.loads(storage.read_text(encoding="utf-8"))
|
|
names = {c["name"] for c in persisted["cookies"]}
|
|
assert "SID" in names
|
|
assert "X" not in names
|
|
assert result is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Redirected to login → loud failure, no hang
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.requires_playwright
|
|
def test_headless_redirected_to_login_raises_loudly(tmp_path: Path) -> None:
|
|
storage = tmp_path / "storage_state.json"
|
|
profile = tmp_path / "browser_profile"
|
|
profile.mkdir()
|
|
|
|
playwright, _context, page = _fake_playwright_landing(
|
|
"https://accounts.google.com/signin/v2/identifier"
|
|
)
|
|
io = _RaisingCaptureIO()
|
|
|
|
with pytest.raises(HeadlessLoginRequiredError, match="session is"):
|
|
_run_headless(
|
|
BrowserCapturePlan(browser="chromium", browser_profile=profile, storage_path=storage),
|
|
io,
|
|
playwright,
|
|
)
|
|
|
|
# The headless arm must NOT wait for a human.
|
|
page.wait_for_url.assert_not_called()
|
|
# Nothing was persisted on the dead-session path.
|
|
assert not storage.exists()
|