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
1094 lines
43 KiB
Python
1094 lines
43 KiB
Python
"""Comprehensive VCR tests for all NotebookLM API operations.
|
|
|
|
This file records cassettes for ALL API operations.
|
|
Run with NOTEBOOKLM_VCR_RECORD=1 to record new cassettes.
|
|
|
|
Recording requires the same env vars as e2e tests:
|
|
- NOTEBOOKLM_READ_ONLY_NOTEBOOK_ID: For read-only operations
|
|
- NOTEBOOKLM_GENERATION_NOTEBOOK_ID: For mutable operations
|
|
|
|
Note: Notebook IDs only matter when RECORDING. During replay, VCR uses
|
|
recorded responses regardless of notebook ID.
|
|
|
|
Note: These tests are automatically skipped if cassettes are not available.
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from notebooklm import NotebookLMClient, ReportFormat
|
|
from notebooklm.types import Artifact, ArtifactType
|
|
from tests.integration.conftest import get_vcr_auth, skip_no_cassettes
|
|
from tests.vcr_config import notebooklm_vcr
|
|
|
|
# Skip all tests in this module if cassettes are not available
|
|
pytestmark = [pytest.mark.vcr, skip_no_cassettes]
|
|
|
|
# Use same env vars as e2e tests for consistency.
|
|
#
|
|
# These only matter during recording for endpoints whose matcher ignores the
|
|
# request body (most batchexecute calls). For body-aware matchers
|
|
# — notably ``freq`` on the streaming-chat endpoint — replay also
|
|
# needs to send the SAME notebook_id that was recorded, because the matcher
|
|
# compares slot 7 of the decoded ``f.req`` envelope. We therefore default
|
|
# ``MUTABLE_NOTEBOOK_ID`` to the canonical recording notebook UUID
|
|
# used to record the chat cassettes; recording-time runs override
|
|
# this with the real env var.
|
|
READONLY_NOTEBOOK_ID = os.environ.get(
|
|
"NOTEBOOKLM_READ_ONLY_NOTEBOOK_ID",
|
|
"c3f6285f-1709-44c4-9cd6-e95cf0ea4f5e",
|
|
)
|
|
MUTABLE_NOTEBOOK_ID = os.environ.get(
|
|
"NOTEBOOKLM_GENERATION_NOTEBOOK_ID",
|
|
"bb00c9e3-656c-4fd2-b890-2b71e1cf3814",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Helper for reducing boilerplate
|
|
# =============================================================================
|
|
|
|
|
|
@asynccontextmanager
|
|
async def vcr_client():
|
|
"""Context manager for creating authenticated VCR client."""
|
|
auth = await get_vcr_auth()
|
|
async with NotebookLMClient(auth) as client:
|
|
yield client
|
|
|
|
|
|
# =============================================================================
|
|
# Notebooks API
|
|
# =============================================================================
|
|
|
|
|
|
class TestNotebooksAPI:
|
|
"""Notebooks API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_list.yaml")
|
|
async def test_list(self):
|
|
"""List all notebooks."""
|
|
async with vcr_client() as client:
|
|
notebooks = await client.notebooks.list()
|
|
assert isinstance(notebooks, list)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_get.yaml")
|
|
async def test_get(self):
|
|
"""Get a specific notebook."""
|
|
async with vcr_client() as client:
|
|
notebook = await client.notebooks.get(READONLY_NOTEBOOK_ID)
|
|
assert notebook is not None
|
|
if READONLY_NOTEBOOK_ID:
|
|
assert notebook.id == READONLY_NOTEBOOK_ID
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_get_summary.yaml")
|
|
async def test_get_summary(self):
|
|
"""Get notebook summary."""
|
|
async with vcr_client() as client:
|
|
summary = await client.notebooks.get_summary(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(summary, str)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_get_description.yaml")
|
|
async def test_get_description(self):
|
|
"""Get notebook description."""
|
|
async with vcr_client() as client:
|
|
description = await client.notebooks.get_description(READONLY_NOTEBOOK_ID)
|
|
assert description is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_get_raw.yaml")
|
|
async def test_get_raw(self):
|
|
"""Get raw notebook data."""
|
|
async with vcr_client() as client:
|
|
raw = await client.notebooks.get_raw(READONLY_NOTEBOOK_ID)
|
|
assert raw is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_rename.yaml")
|
|
async def test_rename(self):
|
|
"""Rename a notebook (then rename back)."""
|
|
async with vcr_client() as client:
|
|
notebook = await client.notebooks.get(MUTABLE_NOTEBOOK_ID)
|
|
original_name = notebook.title
|
|
await client.notebooks.rename(MUTABLE_NOTEBOOK_ID, "VCR Test Renamed")
|
|
await client.notebooks.rename(MUTABLE_NOTEBOOK_ID, original_name)
|
|
|
|
|
|
# =============================================================================
|
|
# Sources API
|
|
# =============================================================================
|
|
|
|
|
|
class TestSourcesAPI:
|
|
"""Sources API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_list.yaml")
|
|
async def test_list(self):
|
|
"""List sources in a notebook."""
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(sources, list)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_get_guide.yaml")
|
|
async def test_get_guide(self):
|
|
"""Get source guide for a specific source."""
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(READONLY_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
guide = await client.sources.get_guide(READONLY_NOTEBOOK_ID, sources[0].id)
|
|
assert guide is not None
|
|
# Verify values are actually populated (catches parsing bugs like issue #70)
|
|
assert guide.summary, "Expected non-empty summary from source guide"
|
|
assert isinstance(guide.keywords, tuple)
|
|
assert len(guide.keywords) > 0, "Expected non-empty keywords from source guide"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_get_fulltext.yaml")
|
|
async def test_get_fulltext(self):
|
|
"""Get source fulltext content."""
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(READONLY_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
fulltext = await client.sources.get_fulltext(READONLY_NOTEBOOK_ID, sources[0].id)
|
|
assert fulltext is not None
|
|
assert fulltext.source_id == sources[0].id
|
|
# Verify content is actually populated (catches parsing bugs like issue #70)
|
|
assert fulltext.content, "Expected non-empty content from fulltext"
|
|
assert fulltext.title, "Expected non-empty title from fulltext"
|
|
assert fulltext.char_count > 0, "Expected positive char_count"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_add_text.yaml")
|
|
async def test_add_text(self):
|
|
"""Add a text source."""
|
|
async with vcr_client() as client:
|
|
source = await client.sources.add_text(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
title="VCR Test Source",
|
|
content="This is a test source created by VCR recording.",
|
|
)
|
|
assert source is not None
|
|
assert source.title == "VCR Test Source"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_add_url.yaml")
|
|
async def test_add_url(self):
|
|
"""Add a URL source."""
|
|
async with vcr_client() as client:
|
|
source = await client.sources.add_url(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
url="https://en.wikipedia.org/wiki/Artificial_intelligence",
|
|
)
|
|
assert source is not None
|
|
assert source.id, "Expected non-empty source ID"
|
|
# Title may be extracted from the page
|
|
assert source.title is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_add_drive.yaml")
|
|
async def test_add_drive(self):
|
|
"""Add a Google Drive document source.
|
|
|
|
Uses a public Google Doc for testing. The add_drive() function
|
|
uses single-wrapped params [source_data] (not double-wrapped).
|
|
"""
|
|
async with vcr_client() as client:
|
|
source = await client.sources.add_drive(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
file_id="1oAk_INJHbIPsIh49jgNqj3FESSGHZrzxFY7t05Lvvl0",
|
|
title="VCR Test Drive Doc",
|
|
mime_type="application/vnd.google-apps.document",
|
|
wait=False, # Don't wait for processing during VCR recording
|
|
)
|
|
assert source is not None
|
|
assert source.id, "Expected non-empty source ID"
|
|
# Drive sources use the actual document title, not the passed title
|
|
assert source.title, "Expected non-empty source title"
|
|
|
|
|
|
# =============================================================================
|
|
# Notes API
|
|
# =============================================================================
|
|
|
|
|
|
class TestNotesAPI:
|
|
"""Notes API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notes_list.yaml")
|
|
async def test_list(self):
|
|
"""List notes in a notebook."""
|
|
async with vcr_client() as client:
|
|
notes = await client.notes.list(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(notes, list)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notes_list_mind_maps.yaml")
|
|
async def test_list_mind_maps(self):
|
|
"""List mind maps in a notebook."""
|
|
async with vcr_client() as client:
|
|
mind_maps = await client.notes.list_mind_maps(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(mind_maps, list)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notes_create.yaml")
|
|
async def test_create(self):
|
|
"""Create a note."""
|
|
async with vcr_client() as client:
|
|
note = await client.notes.create(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
title="VCR Test Note",
|
|
content="This is a test note created by VCR recording.",
|
|
)
|
|
assert note is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notes_create_and_update.yaml")
|
|
async def test_create_and_update(self):
|
|
"""Create and update a note."""
|
|
async with vcr_client() as client:
|
|
note = await client.notes.create(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
title="VCR Update Test",
|
|
content="Original content.",
|
|
)
|
|
assert note is not None
|
|
# v0.8.0 (#1362): update() runs an existence preflight; stub the
|
|
# just-created note as a hit so the cassette (recorded pre-flip,
|
|
# without the extra GET_NOTES_AND_MIND_MAPS round-trip) still
|
|
# replays the create+update interactions only.
|
|
client.notes.get_or_none = AsyncMock(return_value=note)
|
|
await client.notes.update(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
note.id,
|
|
title="VCR Update Test - Updated",
|
|
content="Updated content.",
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Artifacts API - Read Operations
|
|
# =============================================================================
|
|
|
|
|
|
# Artifact list method configurations: (method_name, cassette_name)
|
|
ARTIFACT_LIST_METHODS = [
|
|
("list", "artifacts_list.yaml"),
|
|
("list_audio", "artifacts_list_audio.yaml"),
|
|
("list_video", "artifacts_list_video.yaml"),
|
|
("list_reports", "artifacts_list_reports.yaml"),
|
|
("list_quizzes", "artifacts_list_quizzes.yaml"),
|
|
("list_flashcards", "artifacts_list_flashcards.yaml"),
|
|
("list_infographics", "artifacts_list_infographics.yaml"),
|
|
("list_slide_decks", "artifacts_list_slide_decks.yaml"),
|
|
("list_data_tables", "artifacts_list_data_tables.yaml"),
|
|
]
|
|
|
|
|
|
class TestArtifactsListAPI:
|
|
"""Artifacts API list operations - parametrized to reduce duplication."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("method_name,cassette", ARTIFACT_LIST_METHODS)
|
|
async def test_list_artifacts(self, method_name, cassette):
|
|
"""Test artifact list methods."""
|
|
with notebooklm_vcr.use_cassette(cassette):
|
|
async with vcr_client() as client:
|
|
method = getattr(client.artifacts, method_name)
|
|
if method_name == "list":
|
|
result = await method(READONLY_NOTEBOOK_ID)
|
|
else:
|
|
result = await method(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(result, list)
|
|
# Every element the decoder hands back must be a fully-formed
|
|
# Artifact with a non-empty id/title and a known status. This
|
|
# rejects "the call replays but the parser silently returned
|
|
# garbage" — the failure mode the original ``isinstance(list)``
|
|
# check would not catch.
|
|
for art in result:
|
|
assert isinstance(art, Artifact)
|
|
assert isinstance(art.id, str) and art.id
|
|
assert isinstance(art.title, str)
|
|
assert isinstance(art.status, int)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("method_name", "cassette", "expected_kind", "expected_type_code"),
|
|
[
|
|
(
|
|
"list_infographics",
|
|
"artifacts_list_infographics.yaml",
|
|
ArtifactType.INFOGRAPHIC,
|
|
7,
|
|
),
|
|
(
|
|
"list_data_tables",
|
|
"artifacts_list_data_tables.yaml",
|
|
ArtifactType.DATA_TABLE,
|
|
9,
|
|
),
|
|
],
|
|
)
|
|
async def test_list_artifacts_kind_parsing(
|
|
self, method_name, cassette, expected_kind, expected_type_code
|
|
):
|
|
"""Parser turns INFOGRAPHIC (7) and DATA_TABLE (9) rows into the right kind.
|
|
|
|
The two cassettes were already wired
|
|
into :data:`ARTIFACT_LIST_METHODS`, but the surrounding assertion only
|
|
proved the call replayed — not that the decoder mapped the integer
|
|
type code to the user-facing :class:`ArtifactType` enum. This test
|
|
asserts the full parser contract: at least one artifact is returned,
|
|
every artifact carries the expected ``_artifact_type`` integer, and
|
|
every artifact's ``.kind`` property resolves to the expected enum.
|
|
"""
|
|
with notebooklm_vcr.use_cassette(cassette):
|
|
async with vcr_client() as client:
|
|
method = getattr(client.artifacts, method_name)
|
|
result = await method(READONLY_NOTEBOOK_ID)
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) >= 1, f"{cassette} should contain at least one artifact"
|
|
for art in result:
|
|
assert isinstance(art, Artifact)
|
|
assert art._artifact_type == expected_type_code, (
|
|
f"Expected raw type code {expected_type_code}, "
|
|
f"got {art._artifact_type} for artifact {art.id!r}"
|
|
)
|
|
assert art.kind == expected_kind, (
|
|
f"Expected .kind == {expected_kind!r}, got {art.kind!r} for artifact {art.id!r}"
|
|
)
|
|
# The .kind property is a str-enum so equality holds both ways.
|
|
assert art.kind == expected_kind.value
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_suggest_reports.yaml")
|
|
async def test_suggest_reports(self):
|
|
"""Get report suggestions."""
|
|
async with vcr_client() as client:
|
|
suggestions = await client.artifacts.suggest_reports(READONLY_NOTEBOOK_ID)
|
|
assert isinstance(suggestions, list)
|
|
|
|
|
|
class TestArtifactsDownloadAPI:
|
|
"""Artifacts API download operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_report.yaml")
|
|
async def test_download_report(self, tmp_path):
|
|
"""Download a report as markdown."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "report.md"
|
|
try:
|
|
path = await client.artifacts.download_report(
|
|
READONLY_NOTEBOOK_ID, str(output_path)
|
|
)
|
|
assert os.path.exists(path)
|
|
content = output_path.read_text(encoding="utf-8")
|
|
assert len(content) > 0 and "#" in content
|
|
except ValueError as e:
|
|
if "No completed report" in str(e):
|
|
pytest.skip("No completed report artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_mind_map.yaml")
|
|
async def test_download_mind_map(self, tmp_path):
|
|
"""Download a mind map as JSON."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "mindmap.json"
|
|
try:
|
|
path = await client.artifacts.download_mind_map(
|
|
READONLY_NOTEBOOK_ID, str(output_path)
|
|
)
|
|
assert os.path.exists(path)
|
|
data = json.loads(output_path.read_text(encoding="utf-8"))
|
|
assert "name" in data
|
|
except ValueError as e:
|
|
if "No mind maps found" in str(e):
|
|
pytest.skip("No mind map artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_data_table.yaml")
|
|
async def test_download_data_table(self, tmp_path):
|
|
"""Download a data table as CSV."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "data.csv"
|
|
try:
|
|
path = await client.artifacts.download_data_table(
|
|
READONLY_NOTEBOOK_ID, str(output_path)
|
|
)
|
|
assert os.path.exists(path)
|
|
with open(output_path, encoding="utf-8-sig") as f:
|
|
rows = list(csv.reader(f))
|
|
assert len(rows) >= 1
|
|
except ValueError as e:
|
|
if "No completed data table" in str(e):
|
|
pytest.skip("No completed data table artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_quiz.yaml")
|
|
async def test_download_quiz(self, tmp_path):
|
|
"""Download a quiz as JSON."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "quiz.json"
|
|
try:
|
|
path = await client.artifacts.download_quiz(READONLY_NOTEBOOK_ID, str(output_path))
|
|
assert os.path.exists(path)
|
|
data = json.loads(output_path.read_text(encoding="utf-8"))
|
|
assert "title" in data
|
|
assert "questions" in data
|
|
except ValueError as e:
|
|
if "No completed quiz" in str(e):
|
|
pytest.skip("No completed quiz artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_quiz_markdown.yaml")
|
|
async def test_download_quiz_markdown(self, tmp_path):
|
|
"""Download a quiz as markdown."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "quiz.md"
|
|
try:
|
|
path = await client.artifacts.download_quiz(
|
|
READONLY_NOTEBOOK_ID, str(output_path), output_format="markdown"
|
|
)
|
|
assert os.path.exists(path)
|
|
content = output_path.read_text(encoding="utf-8")
|
|
assert "# " in content # Should have a heading
|
|
assert "Question" in content or "##" in content
|
|
except ValueError as e:
|
|
if "No completed quiz" in str(e):
|
|
pytest.skip("No completed quiz artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_flashcards.yaml")
|
|
async def test_download_flashcards(self, tmp_path):
|
|
"""Download flashcards as JSON."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "flashcards.json"
|
|
try:
|
|
path = await client.artifacts.download_flashcards(
|
|
READONLY_NOTEBOOK_ID, str(output_path)
|
|
)
|
|
assert os.path.exists(path)
|
|
data = json.loads(output_path.read_text(encoding="utf-8"))
|
|
assert "title" in data
|
|
assert "cards" in data
|
|
# Verify normalized format (front/back, not f/b)
|
|
if data["cards"]:
|
|
assert "front" in data["cards"][0]
|
|
assert "back" in data["cards"][0]
|
|
except ValueError as e:
|
|
if "No completed flashcard" in str(e):
|
|
pytest.skip("No completed flashcard artifact available")
|
|
raise
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_download_flashcards_markdown.yaml")
|
|
async def test_download_flashcards_markdown(self, tmp_path):
|
|
"""Download flashcards as markdown."""
|
|
async with vcr_client() as client:
|
|
output_path = tmp_path / "flashcards.md"
|
|
try:
|
|
path = await client.artifacts.download_flashcards(
|
|
READONLY_NOTEBOOK_ID, str(output_path), output_format="markdown"
|
|
)
|
|
assert os.path.exists(path)
|
|
content = output_path.read_text(encoding="utf-8")
|
|
assert "# " in content # Should have a heading
|
|
assert "**Q:**" in content or "Card" in content
|
|
except ValueError as e:
|
|
if "No completed flashcard" in str(e):
|
|
pytest.skip("No completed flashcard artifact available")
|
|
raise
|
|
|
|
|
|
# =============================================================================
|
|
# Artifacts API - Generation Operations (use mutable notebook)
|
|
# =============================================================================
|
|
|
|
|
|
class TestArtifactsGenerateAPI:
|
|
"""Artifacts API generation operations.
|
|
|
|
These tests generate artifacts which may take time and consume quota.
|
|
They use the mutable notebook to avoid polluting the read-only one.
|
|
"""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_generate_report.yaml")
|
|
async def test_generate_report(self):
|
|
"""Generate a briefing doc report."""
|
|
async with vcr_client() as client:
|
|
result = await client.artifacts.generate_report(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
report_format=ReportFormat.BRIEFING_DOC,
|
|
)
|
|
assert result is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_generate_study_guide.yaml")
|
|
async def test_generate_study_guide(self):
|
|
"""Generate a study guide."""
|
|
async with vcr_client() as client:
|
|
result = await client.artifacts.generate_study_guide(MUTABLE_NOTEBOOK_ID)
|
|
assert result is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_generate_quiz.yaml")
|
|
async def test_generate_quiz(self):
|
|
"""Generate a quiz."""
|
|
async with vcr_client() as client:
|
|
result = await client.artifacts.generate_quiz(MUTABLE_NOTEBOOK_ID)
|
|
assert result is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_generate_flashcards.yaml")
|
|
async def test_generate_flashcards(self):
|
|
"""Generate flashcards."""
|
|
async with vcr_client() as client:
|
|
result = await client.artifacts.generate_flashcards(MUTABLE_NOTEBOOK_ID)
|
|
assert result is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_retry_failed.yaml")
|
|
async def test_retry_failed(self):
|
|
"""Retry a failed artifact in place — the same id comes back in_progress."""
|
|
artifact_id = "11111111-2222-3333-4444-555555555555"
|
|
async with vcr_client() as client:
|
|
result = await client.artifacts.retry_failed(MUTABLE_NOTEBOOK_ID, artifact_id)
|
|
assert result.task_id == artifact_id
|
|
assert result.status == "in_progress"
|
|
|
|
|
|
# =============================================================================
|
|
# Chat API
|
|
# =============================================================================
|
|
|
|
|
|
class TestChatAPI:
|
|
"""Chat API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette(
|
|
"chat_ask.yaml",
|
|
# Opt this streaming-chat test in to the ``freq`` body matcher.
|
|
# The matcher decodes the form-encoded ``f.req`` payload so two
|
|
# otherwise-identical POSTs (same method/scheme/host/port/path) can be
|
|
# disambiguated by their param shape. ``freq`` is opt-in per-cassette
|
|
# because most endpoints do not send ``f.req``.
|
|
match_on=["method", "scheme", "host", "port", "path", "freq"],
|
|
)
|
|
async def test_ask(self):
|
|
"""Ask a question."""
|
|
async with vcr_client() as client:
|
|
result = await client.chat.ask(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
"What is this notebook about?",
|
|
)
|
|
assert result is not None
|
|
assert result.answer is not None
|
|
assert result.conversation_id is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette(
|
|
"chat_ask_with_references.yaml",
|
|
# Opt-in to the ``freq`` body matcher so the streaming-chat
|
|
# POST is disambiguated by its decoded ``f.req`` payload rather than
|
|
# by replay-order. See ``test_ask`` above for the full rationale.
|
|
match_on=["method", "scheme", "host", "port", "path", "freq"],
|
|
)
|
|
async def test_ask_with_references(self):
|
|
"""Ask a question that generates references."""
|
|
async with vcr_client() as client:
|
|
result = await client.chat.ask(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
"Summarize the key points with specific citations from the sources.",
|
|
)
|
|
assert result is not None
|
|
assert result.answer is not None
|
|
# References may or may not be present depending on the answer
|
|
assert isinstance(result.references, list)
|
|
# If references exist, verify structure
|
|
for ref in result.references:
|
|
assert ref.source_id is not None
|
|
assert ref.citation_number is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("chat_get_history.yaml")
|
|
async def test_get_conversation_id(self):
|
|
"""Get conversation ID."""
|
|
async with vcr_client() as client:
|
|
conv_id = await client.chat.get_conversation_id(MUTABLE_NOTEBOOK_ID)
|
|
# May be None if no conversations, or a string UUID
|
|
assert conv_id is None or isinstance(conv_id, str)
|
|
|
|
|
|
# =============================================================================
|
|
# Settings API
|
|
# =============================================================================
|
|
|
|
|
|
class TestSettingsAPI:
|
|
"""Settings API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("settings_get_output_language.yaml")
|
|
async def test_get_output_language(self):
|
|
"""Get current output language setting."""
|
|
async with vcr_client() as client:
|
|
language = await client.settings.get_output_language()
|
|
# Language may be None if not set, or a string like "en", "ja", "zh_Hans"
|
|
assert language is None or isinstance(language, str)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("settings_set_output_language.yaml")
|
|
async def test_set_output_language(self):
|
|
"""Set output language (then restore original)."""
|
|
async with vcr_client() as client:
|
|
# Get current language to restore later
|
|
original = await client.settings.get_output_language()
|
|
# Set to English
|
|
result = await client.settings.set_output_language("en")
|
|
assert result == "en" or result is None
|
|
# Restore original if it was set
|
|
if original:
|
|
await client.settings.set_output_language(original)
|
|
|
|
|
|
# =============================================================================
|
|
# Sharing API
|
|
# =============================================================================
|
|
|
|
|
|
class TestSharingAPI:
|
|
"""Sharing API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sharing_get_status.yaml")
|
|
async def test_get_status(self):
|
|
"""Get sharing status for a notebook."""
|
|
async with vcr_client() as client:
|
|
status = await client.sharing.get_status(READONLY_NOTEBOOK_ID)
|
|
assert status is not None
|
|
assert status.notebook_id == READONLY_NOTEBOOK_ID
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sharing_set_public.yaml")
|
|
async def test_set_public(self):
|
|
"""Toggle public sharing (restore original state)."""
|
|
async with vcr_client() as client:
|
|
# Get current status
|
|
original = await client.sharing.get_status(MUTABLE_NOTEBOOK_ID)
|
|
# Toggle to opposite
|
|
new_status = await client.sharing.set_public(
|
|
MUTABLE_NOTEBOOK_ID, not original.is_public
|
|
)
|
|
assert new_status.is_public != original.is_public
|
|
# Restore original state
|
|
await client.sharing.set_public(MUTABLE_NOTEBOOK_ID, original.is_public)
|
|
|
|
|
|
# =============================================================================
|
|
# Sources API - Additional Operations
|
|
# =============================================================================
|
|
|
|
|
|
class TestSourcesAdditionalAPI:
|
|
"""Additional sources API operations not covered in main TestSourcesAPI."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_add_file.yaml")
|
|
async def test_add_file(self, tmp_path):
|
|
"""Add a file source."""
|
|
# Create a test file
|
|
test_file = tmp_path / "vcr_test_document.txt"
|
|
test_file.write_text("This is a test document for VCR cassette recording.")
|
|
|
|
async with vcr_client() as client:
|
|
source = await client.sources.add_file(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
str(test_file),
|
|
)
|
|
assert source is not None
|
|
assert source.id is not None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_check_freshness.yaml")
|
|
async def test_check_freshness(self):
|
|
"""Check source freshness."""
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(READONLY_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
is_fresh = await client.sources.check_freshness(READONLY_NOTEBOOK_ID, sources[0].id)
|
|
assert isinstance(is_fresh, bool)
|
|
# The cassette shows API returns [] which should be interpreted as fresh
|
|
assert is_fresh is True, "Source in cassette should be fresh (API returned [])"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_check_freshness_drive.yaml")
|
|
async def test_check_freshness_drive(self):
|
|
"""Check freshness for Drive source (different response format)."""
|
|
from notebooklm import SourceType
|
|
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(MUTABLE_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
# Find a GOOGLE_DOCS source
|
|
drive_source = next((s for s in sources if s.kind == SourceType.GOOGLE_DOCS), None)
|
|
if not drive_source:
|
|
pytest.skip("No GOOGLE_DOCS source available")
|
|
is_fresh = await client.sources.check_freshness(MUTABLE_NOTEBOOK_ID, drive_source.id)
|
|
assert isinstance(is_fresh, bool)
|
|
# Drive sources return [[null, true, [source_id]]] when fresh
|
|
assert is_fresh is True, "Drive source should be fresh"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_refresh.yaml")
|
|
async def test_refresh(self):
|
|
"""Refresh a source."""
|
|
from notebooklm import SourceType
|
|
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(MUTABLE_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
# Find a WEB_PAGE source (text sources can't be refreshed)
|
|
url_source = next((s for s in sources if s.kind == SourceType.WEB_PAGE), None)
|
|
if not url_source:
|
|
pytest.skip("No WEB_PAGE source available for refresh")
|
|
result = await client.sources.refresh(MUTABLE_NOTEBOOK_ID, url_source.id)
|
|
# v0.8.0 (#1290): refresh() returns None on success (no exception)
|
|
assert result is None, "refresh() should return None on success"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_rename.yaml")
|
|
async def test_rename(self):
|
|
"""Rename a source (then restore original name)."""
|
|
async with vcr_client() as client:
|
|
sources = await client.sources.list(MUTABLE_NOTEBOOK_ID)
|
|
if not sources:
|
|
pytest.skip("No sources available")
|
|
source = sources[0]
|
|
original_title = source.title
|
|
# Rename
|
|
renamed = await client.sources.rename(
|
|
MUTABLE_NOTEBOOK_ID, source.id, "VCR Test Renamed Source"
|
|
)
|
|
assert renamed.title == "VCR Test Renamed Source"
|
|
# Restore
|
|
await client.sources.rename(MUTABLE_NOTEBOOK_ID, source.id, original_title)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("sources_delete.yaml")
|
|
async def test_delete(self):
|
|
"""Delete a source (creates one first to delete)."""
|
|
async with vcr_client() as client:
|
|
# Create a source to delete
|
|
source = await client.sources.add_text(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
title="VCR Delete Test Source",
|
|
content="This source will be deleted.",
|
|
)
|
|
assert source is not None
|
|
# Delete it
|
|
result = await client.sources.delete(MUTABLE_NOTEBOOK_ID, source.id)
|
|
assert result is None
|
|
|
|
|
|
# =============================================================================
|
|
# Notebooks API - Additional Operations
|
|
# =============================================================================
|
|
|
|
|
|
class TestNotebooksAdditionalAPI:
|
|
"""Additional notebooks API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_create.yaml")
|
|
async def test_create(self):
|
|
"""Create a new notebook."""
|
|
async with vcr_client() as client:
|
|
notebook = await client.notebooks.create("VCR Test Notebook")
|
|
assert notebook is not None
|
|
assert notebook.title == "VCR Test Notebook"
|
|
# Note: We don't delete it here to keep the cassette simple
|
|
# A separate delete test will clean up
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_delete.yaml")
|
|
async def test_delete(self):
|
|
"""Delete a notebook (creates one first)."""
|
|
async with vcr_client() as client:
|
|
# Create a notebook to delete
|
|
notebook = await client.notebooks.create("VCR Delete Test Notebook")
|
|
assert notebook is not None
|
|
# Delete it
|
|
result = await client.notebooks.delete(notebook.id)
|
|
assert result is None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notebooks_remove_from_recent.yaml")
|
|
async def test_remove_from_recent(self):
|
|
"""Remove a notebook from recently viewed."""
|
|
async with vcr_client() as client:
|
|
# This just removes from the recent list, doesn't delete
|
|
await client.notebooks.remove_from_recent(MUTABLE_NOTEBOOK_ID)
|
|
# No return value to check - if it doesn't raise, it worked
|
|
|
|
|
|
# =============================================================================
|
|
# Notes API - Additional Operations
|
|
# =============================================================================
|
|
|
|
|
|
class TestNotesAdditionalAPI:
|
|
"""Additional notes API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("notes_delete.yaml")
|
|
async def test_delete(self):
|
|
"""Delete a note (creates one first)."""
|
|
async with vcr_client() as client:
|
|
# Create a note to delete
|
|
note = await client.notes.create(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
title="VCR Delete Test Note",
|
|
content="This note will be deleted.",
|
|
)
|
|
assert note is not None
|
|
# Delete it
|
|
result = await client.notes.delete(MUTABLE_NOTEBOOK_ID, note.id)
|
|
assert result is None
|
|
|
|
|
|
# =============================================================================
|
|
# Artifacts API - Additional Operations
|
|
# =============================================================================
|
|
|
|
|
|
class TestArtifactsAdditionalAPI:
|
|
"""Additional artifacts API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_rename.yaml")
|
|
async def test_rename(self):
|
|
"""Rename an artifact."""
|
|
async with vcr_client() as client:
|
|
# List artifacts to find one to rename
|
|
artifacts = await client.artifacts.list(MUTABLE_NOTEBOOK_ID)
|
|
if not artifacts:
|
|
pytest.skip("No artifacts available")
|
|
artifact = artifacts[0]
|
|
original_title = artifact.title
|
|
# v0.8.0 (#1362): return_object=False now runs the existence
|
|
# preflight too. Stub it as a hit (the artifact came from the list
|
|
# above, so it exists) so no extra LIST_ARTIFACTS round-trip fires
|
|
# and the existing cassette (rename RPC only) keeps replaying.
|
|
client.artifacts._listing.get_studio_only = AsyncMock(return_value=artifact)
|
|
# Rename, then restore the original name.
|
|
await client.artifacts.rename(
|
|
MUTABLE_NOTEBOOK_ID, artifact.id, "VCR Renamed Artifact", return_object=False
|
|
)
|
|
await client.artifacts.rename(
|
|
MUTABLE_NOTEBOOK_ID, artifact.id, original_title, return_object=False
|
|
)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_delete.yaml")
|
|
async def test_delete(self):
|
|
"""Delete an artifact."""
|
|
async with vcr_client() as client:
|
|
# List existing artifacts
|
|
artifacts = await client.artifacts.list(MUTABLE_NOTEBOOK_ID)
|
|
if not artifacts:
|
|
pytest.skip("No artifacts available to delete")
|
|
# Delete the first one
|
|
artifact_id = artifacts[0].id
|
|
deleted = await client.artifacts.delete(MUTABLE_NOTEBOOK_ID, artifact_id)
|
|
assert deleted is None
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("artifacts_export_report.yaml")
|
|
async def test_export_report(self):
|
|
"""Export a report to Google Docs."""
|
|
async with vcr_client() as client:
|
|
# Find a completed report artifact
|
|
reports = await client.artifacts.list_reports(MUTABLE_NOTEBOOK_ID)
|
|
completed_reports = [r for r in reports if r.is_completed]
|
|
if not completed_reports:
|
|
pytest.skip("No completed report artifact available")
|
|
report = completed_reports[0]
|
|
# Export it to Google Docs
|
|
result = await client.artifacts.export_report(
|
|
MUTABLE_NOTEBOOK_ID, report.id, title="VCR Export Test"
|
|
)
|
|
assert result is not None
|
|
|
|
|
|
# =============================================================================
|
|
# Research API
|
|
# =============================================================================
|
|
|
|
|
|
class TestResearchAPI:
|
|
"""Research API operations."""
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("research_start_fast.yaml")
|
|
async def test_start_fast(self):
|
|
"""Start fast web research."""
|
|
async with vcr_client() as client:
|
|
result = await client.research.start(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
query="Python programming best practices",
|
|
source="web",
|
|
mode="fast",
|
|
)
|
|
assert result is not None
|
|
assert result.task_id
|
|
assert result.mode == "fast"
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("research_poll.yaml")
|
|
async def test_poll(self):
|
|
"""Poll research status."""
|
|
async with vcr_client() as client:
|
|
# Start research first
|
|
start_result = await client.research.start(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
query="Machine learning fundamentals",
|
|
source="web",
|
|
mode="fast",
|
|
)
|
|
if not start_result or not start_result.task_id:
|
|
pytest.skip("Could not start research")
|
|
|
|
# Poll for results
|
|
result = await client.research.poll(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
task_id=start_result.task_id,
|
|
)
|
|
assert result is not None
|
|
assert result.status
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("research_import_sources.yaml")
|
|
async def test_import_sources(self):
|
|
"""Import research sources."""
|
|
async with vcr_client() as client:
|
|
# Start research
|
|
start_result = await client.research.start(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
query="Data science tutorials",
|
|
source="web",
|
|
mode="fast",
|
|
)
|
|
if not start_result or not start_result.task_id:
|
|
pytest.skip("Could not start research")
|
|
|
|
# Poll until we have sources (with timeout via cassette)
|
|
poll_result = await client.research.poll(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
task_id=start_result.task_id,
|
|
)
|
|
if not poll_result.sources:
|
|
pytest.skip("No research sources found")
|
|
|
|
# Import first source
|
|
imported = await client.research.import_sources(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
start_result.task_id,
|
|
poll_result.sources[:1],
|
|
)
|
|
assert isinstance(imported, list)
|
|
|
|
@pytest.mark.vcr
|
|
@pytest.mark.asyncio
|
|
@notebooklm_vcr.use_cassette("research_start_deep.yaml")
|
|
async def test_start_deep(self):
|
|
"""Start deep web research."""
|
|
async with vcr_client() as client:
|
|
result = await client.research.start(
|
|
MUTABLE_NOTEBOOK_ID,
|
|
query="Artificial intelligence history",
|
|
source="web",
|
|
mode="deep",
|
|
)
|
|
assert result is not None
|
|
assert result.task_id
|
|
assert result.mode == "deep"
|