wehub/teng-lin--notebooklm-py
1
0
Fork 0
Code Issues Pull Requests Actions 11 Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
teng-lin--notebooklm-py/tests/unit/test_chat_characterization.py
T
wehub-resource-sync 09e9f3545f
CodeQL / Analyze (push) Waiting to run
Details
dependency-audit / pip-audit (push) Waiting to run
Details
Test / Code Quality (push) Has been cancelled
Details
Test / Test (macos-latest, Python 3.10) (push) Has been cancelled
Details
Test / Test (macos-latest, Python 3.11) (push) Has been cancelled
Details
Test / Test (macos-latest, Python 3.12) (push) Has been cancelled
Details
Test / Test (macos-latest, Python 3.13) (push) Has been cancelled
Details
Test / Test (macos-latest, Python 3.14) (push) Has been cancelled
Details
Test / Test (ubuntu-latest, Python 3.10) (push) Has been cancelled
Details
Test / Test (ubuntu-latest, Python 3.11) (push) Has been cancelled
Details
Test / Test (ubuntu-latest, Python 3.12) (push) Has been cancelled
Details
Test / Test (ubuntu-latest, Python 3.13) (push) Has been cancelled
Details
Test / Test (ubuntu-latest, Python 3.14) (push) Has been cancelled
Details
Test / Test (windows-latest, Python 3.10) (push) Has been cancelled
Details
Test / Test (windows-latest, Python 3.11) (push) Has been cancelled
Details
Test / Test (windows-latest, Python 3.12) (push) Has been cancelled
Details
Test / Test (windows-latest, Python 3.13) (push) Has been cancelled
Details
Test / Test (windows-latest, Python 3.14) (push) Has been cancelled
Details
chore: import upstream snapshot with attribution
2026-07-13 13:30:13 +08:00

2217 lines
88 KiB
Python
Raw Permalink Blame History

"""Characterization tests for ChatAPI.
These tests pin current observable behavior of the ChatAPI surface against
hand-rolled streaming stubs and synthetic ``build_rpc_response`` payloads.
They intentionally stay in ``tests/unit/`` (not ``tests/integration/``)
because the streaming flows here are too synthetic to back with a real
``vcrpy`` cassette — the assertions exercise specific stub-shaped chunk
boundaries and AsyncMock-driven retries that no real RPC recording would
faithfully reproduce. An earlier rename moved this file from
``test_chat_integration.py`` and labeled it with the ``characterization``
marker; see ``pyproject.toml`` markers list for the rationale.
"""
from unittest.mock import AsyncMock, patch
import pytest
from pytest_httpx import HTTPXMock
from notebooklm import NotebookLMClient
from notebooklm.rpc import ChatGoal, ChatResponseLength, RPCMethod
from notebooklm.types import ChatMode, ChatSettings
pytestmark = pytest.mark.characterization
class TestChatAPI:
"""Characterization tests for the ChatAPI."""
def test_chat_cache_is_distinct_per_client_instance(self, auth_tokens):
# Pins the per-tenant isolation contract: two NotebookLMClient
# instances must never share a conversation cache, or follow-up
# turns from one tenant could leak into the other's outgoing
# history payload. The default ``ChatAPI(conversation_cache=None)``
# factory path constructs a fresh ``ConversationCache`` per
# instance — this test would have caught a regression where
# someone made the default a class-level singleton.
client_a = NotebookLMClient(auth_tokens)
client_b = NotebookLMClient(auth_tokens)
assert client_a.chat._cache is not client_b.chat._cache
@pytest.mark.asyncio
async def test_get_conversation_id(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_id returns the most recent conversation ID."""
response = build_rpc_response(
RPCMethod.GET_LAST_CONVERSATION_ID,
[[["conv_001"]]],
)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_id("nb_123")
assert result == "conv_001"
request = httpx_mock.get_request()
assert RPCMethod.GET_LAST_CONVERSATION_ID in str(request.url)
@pytest.mark.asyncio
async def test_get_history(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history returns flat Q&A pairs from the most recent conversation."""
# First call: get_conversation_id
id_response = build_rpc_response(
RPCMethod.GET_LAST_CONVERSATION_ID,
[[["conv_001"]]],
)
# Second call: get_conversation_turns
# API returns individual turns newest-first: A2, Q2, A1, Q1
turns_response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[
[
[None, None, 2, None, [["Answer to second question."]]],
[None, None, 1, "Second question?"],
[None, None, 2, None, [["Answer to first question."]]],
[None, None, 1, "First question?"],
]
],
)
httpx_mock.add_response(content=id_response.encode())
httpx_mock.add_response(content=turns_response.encode())
async with NotebookLMClient(auth_tokens) as client:
qa_pairs = await client.chat.get_history("nb_123")
# get_history reverses API order to return oldest-first
assert len(qa_pairs) == 2
assert qa_pairs[0] == ("First question?", "Answer to first question.")
assert qa_pairs[1] == ("Second question?", "Answer to second question.")
@pytest.mark.asyncio
async def test_get_conversation_turns(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test getting conversation turns for a specific conversation.
The khqZz RPC returns Q&A turns for a conversation:
turn[2] == 1: user question, text at turn[3]
turn[2] == 2: AI answer, text at turn[4][0][0]
Turns are returned newest-first; limit=2 yields the latest Q&A pair.
"""
response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[
[
[None, None, 1, "What is machine learning?"],
[None, None, 2, None, [["Machine learning is a branch of AI."]]],
]
],
)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_turns("nb_123", "conv_001", limit=2)
assert result is not None
turns = result[0]
assert len(turns) == 2
# Turn type 1: user question
assert turns[0][2] == 1
assert turns[0][3] == "What is machine learning?"
# Turn type 2: AI answer
assert turns[1][2] == 2
assert turns[1][4][0][0] == "Machine learning is a branch of AI."
request = httpx_mock.get_request()
assert RPCMethod.GET_CONVERSATION_TURNS in str(request.url)
@pytest.mark.asyncio
async def test_get_conversation_turns_empty(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_turns handles empty turn list gracefully."""
response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[[]],
)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_turns("nb_123", "conv_001")
assert result is not None
assert result[0] == []
@pytest.mark.asyncio
async def test_get_history_empty(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test getting empty conversation history when no conversations exist."""
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
qa_pairs = await client.chat.get_history("nb_123")
assert qa_pairs == []
@pytest.mark.asyncio
async def test_configure_default_mode(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test configuring chat with default settings."""
response = build_rpc_response(RPCMethod.RENAME_NOTEBOOK, None)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
await client.chat.configure("nb_123")
request = httpx_mock.get_request()
assert RPCMethod.RENAME_NOTEBOOK in str(request.url)
@pytest.mark.asyncio
async def test_configure_learning_guide_mode(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test configuring chat as learning guide."""
response = build_rpc_response(RPCMethod.RENAME_NOTEBOOK, None)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
await client.chat.configure(
"nb_123",
goal=ChatGoal.LEARNING_GUIDE,
response_length=ChatResponseLength.LONGER,
)
request = httpx_mock.get_request()
assert RPCMethod.RENAME_NOTEBOOK in str(request.url)
@pytest.mark.asyncio
async def test_configure_custom_mode_without_prompt_raises(
self,
auth_tokens,
httpx_mock: HTTPXMock,
):
"""Test that CUSTOM mode without prompt raises ValidationError."""
from notebooklm.exceptions import ValidationError
async with NotebookLMClient(auth_tokens) as client:
with pytest.raises(ValidationError, match="custom_prompt is required"):
await client.chat.configure("nb_123", goal=ChatGoal.CUSTOM)
@pytest.mark.asyncio
async def test_configure_custom_mode_with_prompt(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test configuring chat with custom prompt."""
response = build_rpc_response(RPCMethod.RENAME_NOTEBOOK, None)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
await client.chat.configure(
"nb_123",
goal=ChatGoal.CUSTOM,
custom_prompt="You are a helpful tutor.",
)
request = httpx_mock.get_request()
assert RPCMethod.RENAME_NOTEBOOK in str(request.url)
@pytest.mark.asyncio
async def test_set_mode(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test setting chat mode with predefined config."""
response = build_rpc_response(RPCMethod.RENAME_NOTEBOOK, None)
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
await client.chat.set_mode("nb_123", ChatMode.CONCISE)
request = httpx_mock.get_request()
assert RPCMethod.RENAME_NOTEBOOK in str(request.url)
@pytest.mark.asyncio
async def test_get_settings_decodes_custom_persona(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""get_settings reads GET_NOTEBOOK and decodes the nb_info[7] block (#1751)."""
nb_info = [0, 1, 2, 3, 4, 5, 6, [[2, "chemistry tutor"], [4]]]
response = build_rpc_response(RPCMethod.GET_NOTEBOOK, [nb_info])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
settings = await client.chat.get_settings("nb_123")
assert settings == ChatSettings(
goal=ChatGoal.CUSTOM,
response_length=ChatResponseLength.LONGER,
custom_prompt="chemistry tutor",
)
request = httpx_mock.get_request()
assert RPCMethod.GET_NOTEBOOK in str(request.url)
@pytest.mark.asyncio
async def test_get_settings_never_configured_is_defaults(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""A never-configured notebook returns null at nb_info[7] → DEFAULT/DEFAULT."""
nb_info = [0, 1, 2, 3, 4, 5, 6, None]
response = build_rpc_response(RPCMethod.GET_NOTEBOOK, [nb_info])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
settings = await client.chat.get_settings("nb_123")
assert settings == ChatSettings(
goal=ChatGoal.DEFAULT,
response_length=ChatResponseLength.DEFAULT,
custom_prompt=None,
)
@pytest.mark.asyncio
async def test_get_settings_unknown_enum_code_raises_drift(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""An unknown goal/length code (new server enum) surfaces as decode drift."""
from notebooklm.exceptions import UnknownRPCMethodError
# goal_code 7 is not a known ChatGoal member.
nb_info = [0, 1, 2, 3, 4, 5, 6, [[7], [1]]]
response = build_rpc_response(RPCMethod.GET_NOTEBOOK, [nb_info])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
with pytest.raises(UnknownRPCMethodError):
await client.chat.get_settings("nb_123")
def test_get_cached_turns_empty(self, auth_tokens):
"""Test getting cached turns for new conversation."""
client = NotebookLMClient(auth_tokens)
turns = client.chat.get_cached_turns("nonexistent_conv")
assert turns == []
def test_clear_cache(self, auth_tokens):
"""Test clearing conversation cache."""
client = NotebookLMClient(auth_tokens)
result = client.chat.clear_cache("some_conv")
assert result is False
def test_clear_all_cache(self, auth_tokens):
"""Test clearing all conversation caches."""
client = NotebookLMClient(auth_tokens)
result = client.chat.clear_cache()
assert result is True
class TestChatReferences:
"""Characterization tests for chat references and citations."""
@pytest.mark.asyncio
async def test_ask_with_citations_returns_references(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() returns references when citations are present."""
import json
import re
# Build a realistic response with citations
# Structure discovered via API analysis:
# cite[1][4] = [[passage_wrapper]] where passage_wrapper[0] = [start, end, nested]
# nested = [[inner]] where inner = [start2, end2, text]
inner_data = [
[
"Machine learning is a subset of AI [1]. It uses algorithms to learn from data [2].",
None,
["chunk-001", "chunk-002", 987654],
None,
[
[],
None,
None,
[
# First citation
[
["chunk-001"],
[
None,
None,
0.95,
[[None]],
[ # cite[1][4] - text passages
[ # passage_wrapper
[ # passage_data
100, # start_char
250, # end_char
[ # nested passages
[ # nested_group
[ # inner
50,
120,
"Machine learning is a branch of artificial intelligence.",
]
]
],
]
]
],
[[[["11111111-1111-1111-1111-111111111111"]]]],
["chunk-001"],
],
],
# Second citation
[
["chunk-002"],
[
None,
None,
0.88,
[[None]],
[
[
[
300,
450,
[
[
[
280,
380,
"Algorithms learn patterns from training data.",
]
]
],
]
]
],
[[[["22222222-2222-2222-2222-222222222222"]]]],
["chunk-002"],
],
],
],
1,
],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id() # issue #659 post-ask round-trip
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
notebook_id="test_nb",
question="What is machine learning?",
source_ids=["src_001"],
)
# Verify answer
assert "Machine learning" in result.answer
assert "[1]" in result.answer
assert "[2]" in result.answer
# Verify references
assert len(result.references) == 2
# First reference
ref1 = result.references[0]
assert ref1.source_id == "11111111-1111-1111-1111-111111111111"
assert ref1.citation_number == 1
assert "artificial intelligence" in ref1.cited_text
# Second reference
ref2 = result.references[1]
assert ref2.source_id == "22222222-2222-2222-2222-222222222222"
assert ref2.citation_number == 2
assert "training data" in ref2.cited_text
@pytest.mark.asyncio
async def test_ask_without_citations(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() works when no citations are in the response."""
import json
import re
inner_data = [
[
"This is a simple answer without any source citations.",
None,
["server-simple-conv", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
notebook_id="test_nb",
question="Simple question",
source_ids=["src_001"],
)
assert result.answer == "This is a simple answer without any source citations."
assert len(result.references) == 0
@pytest.mark.asyncio
async def test_references_include_char_positions(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test that references include character position information."""
import json
import re
inner_data = [
[
"Answer with citation [1].",
None,
["chunk-001", 12345],
None,
[
[],
None,
None,
[
[
["chunk-001"],
[
None,
None,
0.9,
[[None]],
[
[
[
1000, # start_char
1500, # end_char
[[[[950, 1100, "Cited passage text."]]]],
]
]
],
[[[["aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"]]]],
["chunk-001"],
],
],
],
1,
],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
notebook_id="test_nb",
question="Question",
source_ids=["src_001"],
)
assert len(result.references) == 1
ref = result.references[0]
assert ref.start_char == 1000
assert ref.end_char == 1500
assert ref.chunk_id == "chunk-001"
@pytest.mark.asyncio
async def test_ask_returns_answer_when_marker_absent(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() extracts answer when API response lacks type_info[-1]==1 marker.
Regression test for issue #118: Google's API may change or omit the answer
marker, causing the parser to fall back to the longest unmarked text chunk.
"""
import json
import re
# Response with no trailing `1` marker in type_info — simulates changed API format
inner_data = [
[
"This is a valid answer returned without the answer marker.",
None,
["chunk-001", 12345],
None,
[[], None, None, []], # type_info has no trailing 1
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
notebook_id="test_nb",
question="What does this say?",
source_ids=["src_001"],
)
assert result.answer == "This is a valid answer returned without the answer marker."
assert result.conversation_id is not None
assert result.is_follow_up is False
@pytest.mark.asyncio
async def test_ask_prefers_marked_over_unmarked_in_streaming_response(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() picks the marked answer when response has both marked and unmarked chunks.
Streaming responses can contain multiple chunks. The marked answer chunk
(type_info[-1]==1) must win even when an unmarked chunk has longer text.
"""
import json
import re
# Streaming response: first chunk is a longer unmarked preamble,
# second chunk is the shorter but marked real answer.
preamble = [
[
"This is a long preamble or status message that is not the real answer to the question at all.",
None,
["chunk-001", 11111],
None,
[[], None, None, []], # no marker
]
]
answer = [
[
"The real answer.",
None,
["chunk-002", 22222],
None,
[[], None, None, [], 1], # marked
]
]
def make_chunk(inner_data):
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
return f"{len(chunk_json)}\n{chunk_json}"
response_body = f")]}}'\n{make_chunk(preamble)}\n{make_chunk(answer)}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
notebook_id="test_nb",
question="What is the answer?",
source_ids=["src_001"],
)
assert result.answer == "The real answer."
class TestChatAskErrorHandling:
"""Tests for ask() HTTP error handling ."""
@pytest.mark.asyncio
async def test_ask_timeout_raises_network_error(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() raises NetworkError on httpx.TimeoutException."""
import re
import httpx
from notebooklm.exceptions import NetworkError
# A null ask resolves the notebook's current conversation via hPTbtc
# before the POST (issue #1875); mock it so the POST is what fails.
mock_get_conversation_id()
httpx_mock.add_exception(
httpx.TimeoutException("timed out"),
url=re.compile(r".*GenerateFreeFormStreamed.*"),
)
# ``server_error_max_retries=0`` pins the original immediate-raise
# contract; the default retries 5xx + RequestError 3x.
async with NotebookLMClient(auth_tokens, server_error_max_retries=0) as client:
with pytest.raises(NetworkError, match="timed out"):
await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
@pytest.mark.asyncio
async def test_ask_http_status_error_raises_chat_error(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() raises ChatError on httpx.HTTPStatusError.
After the chat-path refactor, the chat path uses
:func:`_chat.transport.chat_aware_authed_post` which routes
through the shared transport pipeline. Auth-shaped statuses
(400/401/403) go through the refresh path before surfacing; this
test uses 500 to exercise the plain
``HTTPStatusError → ChatError`` mapping without entangling the
refresh machinery.
"""
import re
from notebooklm.exceptions import ChatError
# A null ask resolves the notebook's current conversation via hPTbtc
# before the POST (issue #1875); mock it so the POST is what fails.
mock_get_conversation_id()
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
status_code=500,
method="POST",
)
async with NotebookLMClient(auth_tokens, server_error_max_retries=0) as client:
with pytest.raises(ChatError, match="500"):
await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
@pytest.mark.asyncio
async def test_ask_request_error_raises_network_error(
self,
auth_tokens,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() raises NetworkError on httpx.RequestError."""
import re
import httpx
from notebooklm.exceptions import NetworkError
# A null ask resolves the notebook's current conversation via hPTbtc
# before the POST (issue #1875); mock it so the POST is what fails.
mock_get_conversation_id()
httpx_mock.add_exception(
httpx.ConnectError("connection refused"),
url=re.compile(r".*GenerateFreeFormStreamed.*"),
)
async with NotebookLMClient(auth_tokens, server_error_max_retries=0) as client:
with pytest.raises(NetworkError, match="connection refused"):
await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
@pytest.mark.asyncio
async def test_ask_without_csrf_token_skips_at_param(
self,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() without csrf_token omits the 'at' param."""
import json
import re
from notebooklm.auth import AuthTokens
# Auth tokens without csrf_token
auth_no_csrf = AuthTokens(
cookies={"SID": "sid"},
csrf_token=None,
session_id=None,
)
inner_data = [
[
"Answer without csrf.",
None,
["server-no-csrf-conv", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_no_csrf) as client:
result = await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
assert result.answer == "Answer without csrf."
@pytest.mark.asyncio
async def test_ask_with_session_id_adds_fsid_param(
self,
httpx_mock: HTTPXMock,
mock_get_conversation_id,
):
"""Test ask() with session_id adds f.sid param."""
import json
import re
from notebooklm.auth import AuthTokens
auth_with_session = AuthTokens(
cookies={"SID": "sid"},
csrf_token="test_token",
session_id="my_session_id",
)
inner_data = [
[
"Answer with session.",
None,
["server-with-session-conv", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_with_session) as client:
result = await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
assert result.answer == "Answer with session."
# Two HTTP calls now fire (chat-ask + post-ask hPTbtc, issue #659);
# assert f.sid is on the chat-ask leg specifically.
chat_request = next(
r for r in httpx_mock.get_requests() if "GenerateFreeFormStreamed" in str(r.url)
)
assert "f.sid" in str(chat_request.url)
@pytest.mark.asyncio
async def test_ask_empty_answer_does_not_cache_turn_for_follow_up(
self,
auth_tokens,
httpx_mock: HTTPXMock,
):
"""Empty answer on a follow-up must not append a turn to the cache.
Two paths reach this assertion under the current contract:
* **Unparseable response body** (e.g. just the XSSI prefix, garbage,
or wire-drift): now raises ``ChatResponseParseError`` instead of
silently returning an empty answer. The "no cache poisoning"
guarantee falls out of the raise — no turn is appended because
no ``AskResult`` is constructed.
* **Parseable ``wrb.fr`` envelope with empty answer text** (a
legitimate empty answer from the model): still returns
``answer == ""``, preserves the caller-supplied conversation_id,
and skips the cache append.
This test covers the second path — the first is covered in
``tests/unit/test_chat.py::test_streaming_empty_raises``.
"""
import json
import re
# Parseable wrb.fr envelope with an empty answer text. The chunk
# decodes (so parseable_chunk_count > 0 and the parser returns)
# but extracts no answer — exactly the "real empty answer" case
# the contract preserves against ``ChatResponseParseError``.
inner_data = [["", None, None, None, [[], None, None, [], 1]]]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
conversation_id="existing-conv-id",
)
# Empty answer: turn_number equals len(turns) (0), not len(turns)+1
assert result.answer == ""
assert result.turn_number == 0
# Caller-supplied conversation_id is preserved across the empty response.
assert result.conversation_id == "existing-conv-id"
@pytest.mark.asyncio
async def test_ask_follow_up_sets_is_follow_up(
self,
auth_tokens,
httpx_mock: HTTPXMock,
):
"""Test ask() with existing conversation_id sets is_follow_up=True ."""
import json
import re
inner_data = [
[
"Follow-up answer.",
None,
[12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
"nb_123",
"Follow-up question?",
source_ids=["src_001"],
conversation_id="existing-conv-id",
)
assert result.is_follow_up is True
assert result.conversation_id == "existing-conv-id"
class TestAskServerAssignedConversationId:
"""Regression tests for issue #659.
CLI-created conversations must be visible in the NotebookLM web UI. Live
API investigation revealed two facts that drive this contract:
1. ``params[4]`` (the conversation_id slot in the streamed-chat request)
must be ``null`` for new conversations. A client-minted UUID there
orphans the turn from the web UI conversation list.
2. The streamed-chat response's ``first[2][0]`` field is a per-stream
query id, **not** a real conversation_id (querying ``khqZz`` with it
returns 0 turns). The real conversation_id can only be obtained from
``hPTbtc`` (``ChatAPI.get_conversation_id``) after the ask completes.
So ``ask()`` for a new conversation must:
- send ``null`` at ``params[4]`` (verified at the wire level here)
- call ``hPTbtc`` after the ask and surface that id as
``AskResult.conversation_id`` (so follow-ups using it actually work)
- raise ``ChatError`` if ``hPTbtc`` returns ``None`` (the server
failed to record the turn, or the API shape drifted)
"""
@staticmethod
def _decode_params(request) -> list:
"""Decode the params list from a chat POST request body."""
import json
from urllib.parse import parse_qs, unquote
body = request.content.decode("utf-8")
body_qs = parse_qs(body, keep_blank_values=True)
f_req = json.loads(unquote(body_qs["f.req"][0]))
return json.loads(f_req[1])
@pytest.mark.asyncio
async def test_new_conversation_sends_null_conversation_id_in_request(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""New conversations: ``params[4] == null`` AND the post-ask
``hPTbtc`` result becomes ``AskResult.conversation_id``.
"""
import json
import re
# The "stream id" the parser used to mislabel as server_conv_id.
# It must NOT end up as AskResult.conversation_id.
stream_id = "stream-aaaa-bbbb-cccc-dddddddddddd"
inner_data = [
[
"Server-assigned answer.",
None,
[stream_id, 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
chat_response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=chat_response_body.encode(),
method="POST",
)
# Post-ask hPTbtc returns the REAL conversation_id. AskResult
# must adopt this, NOT first[2][0].
real_conv = "real-1111-2222-3333-444444444444"
hptbtc_response = build_rpc_response(
RPCMethod.GET_LAST_CONVERSATION_ID,
[[[real_conv]]],
)
httpx_mock.add_response(
url=re.compile(r".*batchexecute.*"),
content=hptbtc_response.encode(),
method="POST",
)
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
"nb_123",
"What is this?",
source_ids=["src_001"],
)
# Wire shape check on the first request (chat-ask)
chat_request = next(
r for r in httpx_mock.get_requests() if "GenerateFreeFormStreamed" in str(r.url)
)
params = self._decode_params(chat_request)
assert params[4] is None, (
"New conversations must send null in params[4]; got "
f"{params[4]!r}. Sending a client-generated UUID orphans the "
"conversation from the web UI conversation list (issue #659)."
)
# AskResult adopts the hPTbtc id, not the stream id.
assert result.conversation_id == real_conv
assert result.conversation_id != stream_id, (
"AskResult.conversation_id must be the hPTbtc-fetched id, not "
"first[2][0] (which is a stream id; live API tests proved "
"follow-ups using it produce ghost turns — issue #659)."
)
assert result.is_follow_up is False
# And the SDK must have made the hPTbtc call.
assert any("batchexecute" in str(r.url) for r in httpx_mock.get_requests()), (
"SDK must call hPTbtc after a new-conversation ask to obtain the real conversation_id."
)
@pytest.mark.asyncio
async def test_new_conversation_raises_when_hptbtc_returns_no_conversation(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""If ``hPTbtc`` returns no conversation_id after the ask, raise
``ChatError`` — the turn was not recorded server-side. Silent
fallback to a client UUID is the original bug class (issue #659).
"""
import json
import re
from notebooklm.exceptions import ChatError
inner_data = [
[
"Answer text.",
None,
["some-stream-id", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
chat_response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=chat_response_body.encode(),
method="POST",
)
# hPTbtc returns an empty result -> get_conversation_id() -> None.
# A null ask now does TWO hPTbtc lookups (issue #1875): a pre-POST
# resolve of the notebook's current conversation and the existing
# post-POST id recovery. Both hit this empty response, so mark it
# reusable — otherwise the second lookup is unmocked and the test
# would fail there instead of on the intended ChatError.
empty_hptbtc = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [])
httpx_mock.add_response(
url=re.compile(r".*batchexecute.*"),
content=empty_hptbtc.encode(),
method="POST",
is_reusable=True,
)
async with NotebookLMClient(auth_tokens) as client:
with pytest.raises(ChatError, match="hPTbtc"):
await client.chat.ask(
"nb_123",
"Q?",
source_ids=["src_001"],
)
@pytest.mark.asyncio
async def test_follow_up_sends_caller_conversation_id_in_request(
self,
auth_tokens,
httpx_mock: HTTPXMock,
):
"""Follow-up asks forward the caller-supplied conversation_id
verbatim and do NOT call hPTbtc — the caller already has the real id.
"""
import json
import re
inner_data = [
[
"Follow-up answer.",
None,
["some-stream-id", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.ask(
"nb_123",
"Q?",
source_ids=["src_001"],
conversation_id="existing-conv-id",
)
chat_request = next(
r for r in httpx_mock.get_requests() if "GenerateFreeFormStreamed" in str(r.url)
)
params = self._decode_params(chat_request)
assert params[4] == "existing-conv-id"
# AskResult preserves the caller-supplied id; we no longer rebind
# to first[2][0] because that field is a stream id, not a conv_id.
assert result.conversation_id == "existing-conv-id"
# And no hPTbtc round-trip: the caller already supplied a real id.
assert not any("batchexecute" in str(r.url) for r in httpx_mock.get_requests()), (
"Follow-ups must not call hPTbtc — caller already has the id."
)
class TestGetConversationIdEdgeCases:
"""Tests for get_conversation_id edge cases ."""
@pytest.mark.asyncio
async def test_get_conversation_id_returns_none_on_empty_response(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_id returns None when RPC response is empty list."""
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_id("nb_123")
assert result is None
@pytest.mark.asyncio
async def test_get_conversation_id_returns_none_on_empty_nested(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_id returns None when response has no valid ID."""
# Non-empty response but no valid conv_id in it
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[[]]])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_id("nb_123")
assert result is None
@pytest.mark.asyncio
async def test_get_conversation_id_returns_none_on_non_string_conv(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_id returns None when conv entry is not a string."""
# Response with non-string first element in conv list
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[[42]]])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_id("nb_123")
assert result is None
class TestGetHistoryErrorHandling:
"""Tests for get_history error handling ."""
@pytest.mark.asyncio
async def test_get_history_returns_empty_on_chat_error(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history returns [] when get_conversation_turns raises ChatError."""
from notebooklm.exceptions import ChatError
id_response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[["conv_001"]]])
httpx_mock.add_response(content=id_response.encode())
async with NotebookLMClient(auth_tokens) as client:
with patch.object(
client.chat,
"get_conversation_turns",
new_callable=AsyncMock,
side_effect=ChatError("API error"),
):
result = await client.chat.get_history("nb_123")
assert result == []
@pytest.mark.asyncio
async def test_get_history_returns_empty_on_network_error(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history returns [] when get_conversation_turns raises NetworkError."""
from notebooklm.exceptions import NetworkError
id_response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[["conv_001"]]])
httpx_mock.add_response(content=id_response.encode())
async with NotebookLMClient(auth_tokens) as client:
with patch.object(
client.chat,
"get_conversation_turns",
new_callable=AsyncMock,
side_effect=NetworkError("connection error"),
):
result = await client.chat.get_history("nb_123")
assert result == []
@pytest.mark.asyncio
async def test_get_history_returns_empty_when_no_conversation(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history returns [] when get_conversation_id returns None."""
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_history("nb_123")
assert result == []
@pytest.mark.asyncio
async def test_get_history_raises_on_malformed_turns_container(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""A truthy non-list where the turn list belongs raises, not [] (#1485).
Historically this shape silently parsed to an empty history —
indistinguishable from a genuinely-empty conversation. The container
unwrap now raises ``UnknownRPCMethodError`` so wire drift is loud.
"""
from notebooklm.exceptions import UnknownRPCMethodError
id_response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[["conv_001"]]])
httpx_mock.add_response(content=id_response.encode())
async with NotebookLMClient(auth_tokens) as client:
with patch.object(
client.chat,
"get_conversation_turns",
new_callable=AsyncMock,
return_value=["not-the-turn-list"],
):
with pytest.raises(UnknownRPCMethodError):
await client.chat.get_history("nb_123")
@pytest.mark.asyncio
async def test_get_history_reverses_turns(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history reverses turns_data when turns_data[0][0] is a list ."""
id_response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[["conv_001"]]])
# Newest-first: [A1, Q1]
turns_response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[
[
[None, None, 2, None, [["The answer."]]],
[None, None, 1, "The question?"],
]
],
)
httpx_mock.add_response(content=id_response.encode())
httpx_mock.add_response(content=turns_response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_history("nb_123")
assert len(result) == 1
assert result[0] == ("The question?", "The answer.")
@pytest.mark.asyncio
async def test_get_history_with_provided_conversation_id(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history uses provided conversation_id without fetching it."""
turns_response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[
[
[None, None, 2, None, [["Direct answer."]]],
[None, None, 1, "Direct question?"],
]
],
)
httpx_mock.add_response(content=turns_response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_history("nb_123", conversation_id="conv_direct")
assert len(result) == 1
assert result[0] == ("Direct question?", "Direct answer.")
class TestBuildConversationHistory:
"""Tests for _build_conversation_history ."""
def test_build_conversation_history_returns_none_when_no_cached_turns(self, auth_tokens):
"""Test _build_conversation_history returns None for unknown conversation_id."""
client = NotebookLMClient(auth_tokens)
result = client.chat._build_conversation_history("nonexistent-conv-id")
assert result is None
def test_build_conversation_history_returns_list_when_turns_cached(self, auth_tokens):
"""Test _build_conversation_history returns history list when turns exist."""
client = NotebookLMClient(auth_tokens)
# Manually cache a turn on the chat sub-client (cache moved off Session).
client.chat._cache.cache_conversation_turn(
"test-conv", "What is AI?", "AI is artificial intelligence.", 1
)
result = client.chat._build_conversation_history("test-conv")
assert result is not None
assert len(result) == 2
# History format: [[answer, None, 2], [query, None, 1]]
assert result[0] == ["AI is artificial intelligence.", None, 2]
assert result[1] == ["What is AI?", None, 1]
class TestParseAskResponseEdgeCases:
"""Tests for _parse_ask_response_with_references edge cases ."""
def test_parse_response_with_stripped_prefix(self, auth_tokens):
"""Test that response starting with )]}' has the prefix stripped."""
import json
client = NotebookLMClient(auth_tokens)
inner_data = [["Answer text.", None, [12345], None, [[], None, None, [], 1]]]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert answer == "Answer text."
assert conv_id is None
def test_parse_response_without_prefix(self, auth_tokens):
"""Test response not starting with )]}' is parsed directly."""
import json
client = NotebookLMClient(auth_tokens)
inner_data = [["Answer without prefix.", None, [12345], None, [[], None, None, [], 1]]]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f"{len(chunk_json)}\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert answer == "Answer without prefix."
assert conv_id is None
def test_parse_empty_response_raises_chat_response_parse_error(self, auth_tokens):
"""A response with no parseable ``wrb.fr`` chunk raises.
The XSSI-prefix-only body used to return an empty
``StreamingChatParseResult``. Zero parseable chunks now means
wire-protocol drift / empty body, which is no longer the same
thing as a legitimate empty answer. The parser raises
``ChatResponseParseError`` so callers can surface the failure
instead of silently producing an empty ``AskResult``.
"""
from notebooklm.exceptions import ChatResponseParseError
client = NotebookLMClient(auth_tokens)
with pytest.raises(ChatResponseParseError):
client.chat._parse_ask_response_with_references(")]}'\n")
def test_parse_response_no_marked_answer_falls_back_to_unmarked(self, auth_tokens):
"""Test fallback to unmarked text when no marked answer exists ."""
import json
client = NotebookLMClient(auth_tokens)
# Response with no trailing 1 marker in type_info
inner_data = [
[
"This is unmarked text content.",
None,
[12345],
None,
[[], None, None], # no trailing 1 = unmarked
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert answer == "This is unmarked text content."
assert conv_id is None
def test_parse_response_assigns_citation_numbers(self, auth_tokens):
"""Test that citation_number is assigned based on order of appearance."""
import json
client = NotebookLMClient(auth_tokens)
inner_data = [
[
"Answer with citation.",
None,
[12345],
None,
[
[],
None,
None,
[
[
["chunk-001"],
[
None,
None,
0.9,
[[None]],
[[[100, 200, [[[50, 150, "cited text"]]]]]],
[[[["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"]]]],
["chunk-001"],
],
],
],
1,
],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert len(refs) == 1
assert refs[0].citation_number == 1
assert conv_id is None
class TestExtractAnswerAndRefsFromChunk:
"""Tests for _extract_answer_and_refs_from_chunk edge cases ."""
def test_invalid_json_returns_none(self, auth_tokens):
"""Test that invalid JSON input returns (None, False, [])."""
client = NotebookLMClient(auth_tokens)
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
"not-valid-json"
)
assert text is None
assert is_answer is False
assert refs == []
assert conv_id is None
def test_non_list_data_returns_none(self, auth_tokens):
"""Test that non-list JSON data returns (None, False, []) ."""
import json
client = NotebookLMClient(auth_tokens)
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps("a string value")
)
assert text is None
assert is_answer is False
assert refs == []
assert conv_id is None
def test_item_not_wrb_fr_is_skipped(self, auth_tokens):
"""Test that items where item[0] != 'wrb.fr' are skipped ."""
import json
client = NotebookLMClient(auth_tokens)
data = [["other.fr", "method_id", json.dumps([[["answer"]]])]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert conv_id is None
def test_inner_json_not_string_is_skipped(self, auth_tokens):
"""Test that non-string inner_json is skipped ."""
import json
client = NotebookLMClient(auth_tokens)
# item[2] is an integer, not a string
data = [["wrb.fr", "method_id", 42, None, None]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert conv_id is None
def test_inner_data_first_not_list_raises(self, auth_tokens):
"""A populated record whose answer row is not a list is drift.
Previously this silently returned ``(None, ...)`` (the answer was
dropped). Since the strict-decode migration of ``_chat.wire``
(ADR-0011) a non-list answer row in a *populated* ``wrb.fr`` record is
treated as Google-side wire drift and raises ``UnknownRPCMethodError``.
Strict decoding is the only mode (the ``NOTEBOOKLM_STRICT_DECODE=0``
soft-mode opt-out was retired in v0.7.0).
"""
import json
from notebooklm.exceptions import UnknownRPCMethodError
client = NotebookLMClient(auth_tokens)
# inner_data[0] is a string, not a list
inner_data = ["not a list"]
data = [["wrb.fr", "method_id", json.dumps(inner_data)]]
with pytest.raises(UnknownRPCMethodError):
client.chat._extract_answer_and_refs_from_chunk(json.dumps(data))
def test_inner_data_first_text_not_string_is_skipped(self, auth_tokens):
"""Test that non-string first[0] text is skipped ."""
import json
client = NotebookLMClient(auth_tokens)
# first[0] is None, not a string
inner_data = [[[None, None, [12345]]]]
data = [["wrb.fr", "method_id", json.dumps(inner_data)]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert conv_id is None
def test_inner_json_invalid_json_continues(self, auth_tokens):
"""Test that invalid inner JSON is caught and processing continues."""
import json
client = NotebookLMClient(auth_tokens)
# item[2] is a string but not valid JSON
data = [["wrb.fr", "method_id", "{not valid json}"]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert refs == []
assert conv_id is None
def test_returns_none_when_no_wrb_fr_items(self, auth_tokens):
"""Test that empty list or no matching items returns (None, False, [])."""
import json
client = NotebookLMClient(auth_tokens)
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps([])
)
assert text is None
assert is_answer is False
assert refs == []
assert conv_id is None
def test_item_with_too_few_elements_is_skipped(self, auth_tokens):
"""Test items with len < 3 are skipped."""
import json
client = NotebookLMClient(auth_tokens)
data = [["wrb.fr", "only_two"]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert conv_id is None
class TestParseCitationsEdgeCases:
"""Tests for _parse_citations edge cases (absence soft, malformed loud)."""
def test_parse_citations_raises_on_non_list_answer_row(self, auth_tokens):
"""A non-list ``first`` is structural drift and raises (was a silent-DEBUG ``[]``).
Flipped under the #1505 absence-vs-malformed policy: the stream parser
already raises ``UnknownRPCMethodError`` for a non-list answer row, so
the direct-call surface now matches instead of swallowing a TypeError.
"""
from notebooklm.exceptions import UnknownRPCMethodError
client = NotebookLMClient(auth_tokens)
with pytest.raises(UnknownRPCMethodError):
client.chat._parse_citations(None) # type: ignore[arg-type]
def test_parse_citations_returns_empty_when_first_too_short(self, auth_tokens):
"""Test _parse_citations returns [] when first has <= 4 elements."""
client = NotebookLMClient(auth_tokens)
first = ["text", None, None, None] # len == 4, no index 4
refs = client.chat._parse_citations(first)
assert refs == []
def test_parse_citations_returns_empty_when_type_info_too_short(self, auth_tokens):
"""Test _parse_citations returns [] when type_info has <= 3 elements."""
client = NotebookLMClient(auth_tokens)
first = ["text", None, None, None, [1, 2, 3]] # type_info has no index 3
refs = client.chat._parse_citations(first)
assert refs == []
def test_parse_citations_raises_when_type_info_3_truthy_non_list(self, auth_tokens):
"""A truthy non-list citation container is structural drift and raises.
Flipped under the #1505 absence-vs-malformed policy (was a silent
``[]``): real traffic sends ``None`` (absence, still soft) or a list
here — a truthy non-list means the container itself was reshaped.
"""
from notebooklm.exceptions import UnknownRPCMethodError
client = NotebookLMClient(auth_tokens)
first = ["text", None, None, None, [1, 2, 3, "not_a_list"]]
with pytest.raises(UnknownRPCMethodError):
client.chat._parse_citations(first)
class TestParseSingleCitationEdgeCases:
"""Tests for _parse_single_citation edge cases ."""
def test_parse_single_citation_returns_none_when_not_list(self, auth_tokens):
"""Test _parse_single_citation returns None when cite is not a list ."""
client = NotebookLMClient(auth_tokens)
result = client.chat._parse_single_citation("not a list")
assert result is None
def test_parse_single_citation_returns_none_when_too_short(self, auth_tokens):
"""Test _parse_single_citation returns None when cite has len < 2 ."""
client = NotebookLMClient(auth_tokens)
result = client.chat._parse_single_citation(["only_one"])
assert result is None
def test_parse_single_citation_returns_none_when_cite_inner_not_list(self, auth_tokens):
"""Test _parse_single_citation returns None when cite[1] is not a list ."""
client = NotebookLMClient(auth_tokens)
result = client.chat._parse_single_citation([["chunk-id"], "not_a_list"])
assert result is None
def test_parse_single_citation_returns_none_when_no_source_id(self, auth_tokens):
"""Test _parse_single_citation returns None when no valid UUID found."""
client = NotebookLMClient(auth_tokens)
# cite_inner with 6 elements but source_id_data has no valid UUID
cite = [
["chunk-001"],
[None, None, 0.9, [[None]], [], "not-a-uuid"],
]
result = client.chat._parse_single_citation(cite)
assert result is None
def test_parse_single_citation_with_non_string_chunk_id(self, auth_tokens):
"""Test _parse_single_citation with non-string first item in cite[0] ."""
client = NotebookLMClient(auth_tokens)
# cite[0][0] is not a string, so chunk_id stays None
valid_uuid = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
cite = [
[42], # non-string first item
[
None,
None,
0.9,
[[None]],
[],
[[[[valid_uuid]]]],
],
]
result = client.chat._parse_single_citation(cite)
assert result is not None
assert result.source_id == valid_uuid
assert result.chunk_id is None
def test_parse_single_citation_with_empty_cite_0(self, auth_tokens):
"""Test _parse_single_citation when cite[0] is empty list ."""
client = NotebookLMClient(auth_tokens)
valid_uuid = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
cite = [
[], # empty list - cite[0] is list but empty
[
None,
None,
0.9,
[[None]],
[],
[[[[valid_uuid]]]],
],
]
result = client.chat._parse_single_citation(cite)
assert result is not None
assert result.chunk_id is None
class TestExtractTextPassagesEdgeCases:
"""Tests for _extract_text_passages edge cases ."""
def test_extract_text_passages_returns_none_when_too_short(self, auth_tokens):
"""Test _extract_text_passages returns (None, None, None) when cite_inner too short."""
client = NotebookLMClient(auth_tokens)
cite_inner = [None, None, None, None] # len == 4, no index 4
result = client.chat._extract_text_passages(cite_inner)
assert result == (None, None, None)
def test_extract_text_passages_returns_none_when_index4_not_list(self, auth_tokens):
"""Test _extract_text_passages returns (None, None, None) when cite_inner[4] is not a list."""
client = NotebookLMClient(auth_tokens)
cite_inner = [None, None, None, None, "not_a_list"]
result = client.chat._extract_text_passages(cite_inner)
assert result == (None, None, None)
def test_extract_text_passages_skips_non_list_passage_wrapper(self, auth_tokens):
"""Test _extract_text_passages skips passage_wrapper that is not a list."""
client = NotebookLMClient(auth_tokens)
# cite_inner[4] contains a non-list item
cite_inner = [None, None, None, None, ["not_a_list_wrapper"]]
result = client.chat._extract_text_passages(cite_inner)
assert result == (None, None, None)
def test_extract_text_passages_skips_short_passage_data(self, auth_tokens):
"""Test _extract_text_passages skips passage_data with len < 3."""
client = NotebookLMClient(auth_tokens)
cite_inner = [None, None, None, None, [[[100, 200]]]] # passage_data has len 2
result = client.chat._extract_text_passages(cite_inner)
assert result == (None, None, None)
class TestCollectTextsFromNested:
"""Tests for _collect_texts_from_nested ."""
def test_non_list_input_returns_immediately(self, auth_tokens):
"""Test _collect_texts_from_nested returns immediately for non-list input ."""
client = NotebookLMClient(auth_tokens)
texts = []
client.chat._collect_texts_from_nested("not a list", texts)
assert texts == []
def test_non_list_input_none_returns_immediately(self, auth_tokens):
"""Test _collect_texts_from_nested returns immediately for None input."""
client = NotebookLMClient(auth_tokens)
texts = []
client.chat._collect_texts_from_nested(None, texts)
assert texts == []
def test_nested_group_not_list_is_skipped(self, auth_tokens):
"""Test _collect_texts_from_nested skips nested_group that is not a list."""
client = NotebookLMClient(auth_tokens)
texts = []
# nested contains a non-list item
client.chat._collect_texts_from_nested(["not_a_list_group"], texts)
assert texts == []
def test_text_val_is_list_extracts_strings(self, auth_tokens):
"""Test _collect_texts_from_nested extracts strings when text_val is a list ."""
client = NotebookLMClient(auth_tokens)
texts = []
# Structure: nested=[nested_group], nested_group=[inner], inner=[start, end, text_val]
nested = [[[0, 100, ["part one", " ", "part two"]]]]
client.chat._collect_texts_from_nested(nested, texts)
assert "part one" in texts
assert "part two" in texts
def test_text_val_list_skips_non_strings(self, auth_tokens):
"""Test _collect_texts_from_nested skips non-string items in text_val list."""
client = NotebookLMClient(auth_tokens)
texts = []
# text_val list contains a mix of string and non-string
nested = [[[0, 100, [42, "valid text", None]]]]
client.chat._collect_texts_from_nested(nested, texts)
assert texts == ["valid text"]
def test_inner_with_len_less_than_3_is_skipped(self, auth_tokens):
"""Test _collect_texts_from_nested skips inner items with len < 3."""
client = NotebookLMClient(auth_tokens)
texts = []
# inner has only 2 elements
nested = [[[0, 100]]]
client.chat._collect_texts_from_nested(nested, texts)
assert texts == []
class TestExtractUuidFromNested:
"""Tests for _extract_uuid_from_nested ."""
def test_max_depth_zero_returns_none_with_warning(self, auth_tokens):
"""Test _extract_uuid_from_nested returns None when max_depth=0 ."""
client = NotebookLMClient(auth_tokens)
result = client.chat._extract_uuid_from_nested("some-data", max_depth=0)
assert result is None
def test_none_data_returns_none(self, auth_tokens):
"""Test _extract_uuid_from_nested returns None for None input ."""
client = NotebookLMClient(auth_tokens)
result = client.chat._extract_uuid_from_nested(None)
assert result is None
def test_valid_uuid_string_is_returned(self, auth_tokens):
"""Test _extract_uuid_from_nested returns UUID when given directly as string."""
client = NotebookLMClient(auth_tokens)
valid_uuid = "12345678-1234-1234-1234-123456789abc"
result = client.chat._extract_uuid_from_nested(valid_uuid)
assert result == valid_uuid
def test_non_uuid_string_returns_none(self, auth_tokens):
"""Test _extract_uuid_from_nested returns None for non-UUID string."""
client = NotebookLMClient(auth_tokens)
result = client.chat._extract_uuid_from_nested("not-a-uuid")
assert result is None
def test_list_with_no_uuid_returns_none(self, auth_tokens):
"""Test _extract_uuid_from_nested returns None when list contains no UUID."""
client = NotebookLMClient(auth_tokens)
result = client.chat._extract_uuid_from_nested(["no", "uuid", "here"])
assert result is None
def test_nested_list_finds_uuid(self, auth_tokens):
"""Test _extract_uuid_from_nested finds UUID nested in lists."""
client = NotebookLMClient(auth_tokens)
valid_uuid = "abcdef12-abcd-abcd-abcd-abcdef123456"
result = client.chat._extract_uuid_from_nested([[[[valid_uuid]]]])
assert result == valid_uuid
def test_integer_data_returns_none(self, auth_tokens):
"""Test _extract_uuid_from_nested returns None for integer input."""
client = NotebookLMClient(auth_tokens)
result = client.chat._extract_uuid_from_nested(42)
assert result is None
class TestGetConversationIdNullRaw:
"""Tests for get_conversation_id when rpc_call returns None/falsy."""
@pytest.mark.asyncio
async def test_get_conversation_id_returns_none_when_raw_is_null(
self,
auth_tokens,
):
"""Test get_conversation_id returns None when rpc_call returns None (arc 231->230)."""
async with NotebookLMClient(auth_tokens) as client:
# Patch the direct ``rpc`` collaborator on ChatAPI to return
# None (bypasses decode error). Wave 8 of session-decoupling
# (ADR-0014 Rule 2 Corollary) replaced the old facade with
# direct constructor injection of the underlying collaborators,
# so we reach the chat dispatch surface via ``client.chat._rpc``.
with patch.object(
client.chat._rpc,
"rpc_call",
new_callable=AsyncMock,
return_value=None,
):
result = await client.chat.get_conversation_id("nb_123")
assert result is None
@pytest.mark.asyncio
async def test_get_conversation_id_returns_none_with_non_list_groups(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_conversation_id returns None when groups in raw are not lists (arc 231->230)."""
# Response has non-list items in the outer array (valid decode but no conv_id)
response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, ["not_a_list"])
httpx_mock.add_response(content=response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_conversation_id("nb_123")
assert result is None
class TestGetHistoryTurnsDataNotReversed:
"""Test get_history when turns_data doesn't qualify for reversal (arc 272->279)."""
@pytest.mark.asyncio
async def test_get_history_skips_reversal_when_turns_data_empty(
self,
auth_tokens,
httpx_mock: HTTPXMock,
build_rpc_response,
):
"""Test get_history returns [] when turns_data is empty (skips reversal, arc 272->279)."""
id_response = build_rpc_response(RPCMethod.GET_LAST_CONVERSATION_ID, [[["conv_001"]]])
# turns_data is an empty outer list, so turns_data[0] is falsy
turns_response = build_rpc_response(
RPCMethod.GET_CONVERSATION_TURNS,
[[]],
)
httpx_mock.add_response(content=id_response.encode())
httpx_mock.add_response(content=turns_response.encode())
async with NotebookLMClient(auth_tokens) as client:
result = await client.chat.get_history("nb_123")
assert result == []
class TestParseAskResponseNumericLengthPrefix:
"""Tests for the numeric length-prefixed format in _parse_ask_response ."""
def test_parse_response_with_length_prefix_at_end_of_lines(self, auth_tokens):
"""Test response with numeric line at end has no following line (arc 468->470)."""
import json
client = NotebookLMClient(auth_tokens)
# Response ends with a numeric line (no content follows it)
inner_data = [["Valid answer.", None, [12345], None, [[], None, None, [], 1]]]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
# Append a dangling numeric line at the end with no content following
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n99\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
# The valid chunk should still be parsed
assert answer == "Valid answer."
assert conv_id is None
def test_parse_response_with_direct_json_line_no_prefix(self, auth_tokens):
"""Test response where JSON lines are direct (not length-prefixed) ."""
import json
client = NotebookLMClient(auth_tokens)
# Response with direct JSON chunk (no length prefix)
inner_data = [["Direct JSON answer.", None, [12345], None, [[], None, None, [], 1]]]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
# No length prefix - just direct JSON
response_body = f")]}}'\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert answer == "Direct JSON answer."
assert conv_id is None
class TestExtractAnswerEmptyInnerData:
"""Tests for _extract_answer_and_refs_from_chunk when inner_data is empty (arc 544->532)."""
def test_empty_inner_data_returns_none(self, auth_tokens):
"""Test that inner_data == [] (empty list) causes loop to find nothing (arc 544->532)."""
import json
client = NotebookLMClient(auth_tokens)
# inner_data is an empty list -> len(inner_data) == 0, skips the processing
inner_data: list = []
data = [["wrb.fr", "method_id", json.dumps(inner_data)]]
text, is_answer, refs, conv_id = client.chat._extract_answer_and_refs_from_chunk(
json.dumps(data)
)
assert text is None
assert is_answer is False
assert conv_id is None
class TestExtractTextPassagesMultiplePassages:
"""Tests for _extract_text_passages with multiple passages ."""
def test_extract_text_passages_multiple_passages_updates_end_char(self, auth_tokens):
"""Test that end_char is updated for each valid passage ."""
client = NotebookLMClient(auth_tokens)
# Two passage_wrappers: first sets start_char and end_char, second updates end_char
cite_inner = [
None,
None,
None,
None,
[
# First passage_wrapper
[[100, 200, [[[50, 150, "first text"]]]]],
# Second passage_wrapper - end_char should be updated to 400
[[300, 400, [[[250, 380, "second text"]]]]],
],
]
cited_text, start_char, end_char = client.chat._extract_text_passages(cite_inner)
assert start_char == 100 # from first passage
assert end_char == 400 # updated by second passage
def test_extract_text_passages_start_char_not_reset_on_second_passage(self, auth_tokens):
"""Test start_char is only set once from the first valid passage (arc 676->678)."""
client = NotebookLMClient(auth_tokens)
cite_inner = [
None,
None,
None,
None,
[
# First passage: sets start_char=10
[[10, 100, [[[5, 50, "text one"]]]]],
# Second passage: start_char should NOT be reset to 200
[[200, 300, [[[150, 250, "text two"]]]]],
],
]
_, start_char, _ = client.chat._extract_text_passages(cite_inner)
assert start_char == 10 # only set from first passage
class TestCollectTextsFromNestedEmptyTextVal:
"""Test _collect_texts_from_nested when text_val list has no extractable strings (arc 709->703)."""
def test_text_val_list_all_empty_strings_adds_nothing(self, auth_tokens):
"""Test that text_val list with only whitespace strings adds nothing (arc 709->703)."""
client = NotebookLMClient(auth_tokens)
texts = []
# text_val is a list but all items are whitespace or empty
nested = [[[0, 100, [" ", "", " "]]]]
client.chat._collect_texts_from_nested(nested, texts)
assert texts == []
def test_text_val_list_with_non_string_items_adds_nothing(self, auth_tokens):
"""Test that text_val list with only non-string items adds nothing."""
client = NotebookLMClient(auth_tokens)
texts = []
# text_val is a list but all items are non-strings
nested = [[[0, 100, [None, 42, True]]]]
client.chat._collect_texts_from_nested(nested, texts)
assert texts == []
def test_text_val_neither_string_nor_list_does_nothing(self, auth_tokens):
"""Test that text_val that is neither string nor list is skipped (arc 709->703)."""
client = NotebookLMClient(auth_tokens)
texts = []
# text_val is an integer - neither str nor list - skips both branches
nested = [[[0, 100, 42]]]
client.chat._collect_texts_from_nested(nested, texts)
assert texts == []
class TestParseAskResponseBranchCoverage:
"""Tests for remaining branch coverage in _parse_ask_response and citation assignment."""
def test_existing_marked_answer_not_overwritten_by_shorter(self, auth_tokens):
"""Test process_chunk doesn't update best_marked_answer when text is shorter (arc 454->456)."""
import json
client = NotebookLMClient(auth_tokens)
# First chunk: longer marked answer becomes best_marked_answer
# Second chunk: shorter marked answer - should NOT overwrite (arc 454->456)
longer_inner = [
["This is the longer marked answer text.", None, [12345], None, [[], None, None, [], 1]]
]
shorter_inner = [["Short.", None, [12346], None, [[], None, None, [], 1]]]
longer_json = json.dumps(longer_inner)
shorter_json = json.dumps(shorter_inner)
def make_chunk(inner_json):
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
return f"{len(chunk_json)}\n{chunk_json}"
# Both chunks with marked answers
response_body = f")]}}'\n{make_chunk(longer_json)}\n{make_chunk(shorter_json)}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
# Longer marked answer wins
assert answer == "This is the longer marked answer text."
assert conv_id is None
def test_citation_number_not_reassigned_when_already_set(self, auth_tokens):
"""Test that citation_number already set is not overwritten (arc 496->495)."""
import json
client = NotebookLMClient(auth_tokens)
# Build a response where the reference has citation_number pre-assigned via chunk processing
# We need two chunks each returning the same reference so on second pass citation_number
# is already set. Actually simpler: provide a chunk where citation is parsed and assigned,
# then verify it's 1 (first assignment).
# The arc 496->495 fires when citation_number is not None - we test the final state
inner_data = [
[
"Answer with citation.",
None,
[12345],
None,
[
[],
None,
None,
[
[
["chunk-001"],
[
None,
None,
0.9,
[[None]],
[[[100, 200, [[[50, 150, "cited text"]]]]]],
[[[["eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee"]]]],
["chunk-001"],
],
],
],
1,
],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
# Since the citation-hardening pass, _parse_citations stamps each
# surviving reference with its RAW wire ordinal (so a skipped
# malformed row leaves a hole instead of shifting [N] markers onto
# the wrong citation). The final assignment's preserve-non-None arc
# is therefore the NORMAL path now: it must keep the raw ordinal
# untouched. With nothing skipped, raw ordinal == dense ordinal,
# which is what this asserts (citation_number == 1 for the sole ref).
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
answer, refs, conv_id = client.chat._parse_ask_response_with_references(response_body)
assert len(refs) == 1
assert refs[0].citation_number == 1
assert conv_id is None
class TestExtractTextPassagesNonIntEndChar:
"""Test _extract_text_passages when passage_data[1] is not an int (arc 678->682)."""
def test_non_int_end_char_drops_paired_range(self, auth_tokens):
"""A half-populated start/end pair is dropped to (None, None) so the
ChatReference paired-offset invariant accepts the result. The cited
text is still returned regardless.
"""
client = NotebookLMClient(auth_tokens)
# passage_data[1] is a string, not int — start was set, end never was.
cite_inner = [
None,
None,
None,
None,
[
[[100, "not_an_int", [[[50, 150, "some text"]]]]],
],
]
cited_text, start_char, end_char = client.chat._extract_text_passages(cite_inner)
assert start_char is None # paired-drop: end was never int, start is cleared too
assert end_char is None
assert cited_text is not None # text extraction still succeeded
class TestChatHL:
"""ask() must include the NOTEBOOKLM_HL interface language in the URL."""
@pytest.mark.asyncio
async def test_ask_url_contains_hl_from_env(
self,
auth_tokens,
httpx_mock: HTTPXMock,
monkeypatch,
mock_get_conversation_id,
):
"""When NOTEBOOKLM_HL=ja, the chat POST URL carries hl=ja."""
import json
import re
monkeypatch.setenv("NOTEBOOKLM_HL", "ja")
inner_data = [
[
"answer",
None,
["server-hl-env-conv", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
# Configure hPTbtc with hl=ja too so the post-ask request also
# honors the env var via the same code path.
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
await client.chat.ask(
notebook_id="test_nb",
question="Q",
source_ids=["src_001"],
)
chat_request = next(
r for r in httpx_mock.get_requests() if "GenerateFreeFormStreamed" in str(r.url)
)
assert "hl=ja" in str(chat_request.url)
@pytest.mark.asyncio
async def test_ask_url_defaults_hl_to_en(
self,
auth_tokens,
httpx_mock: HTTPXMock,
monkeypatch,
mock_get_conversation_id,
):
"""When NOTEBOOKLM_HL is unset, the chat URL carries hl=en."""
import json
import re
monkeypatch.delenv("NOTEBOOKLM_HL", raising=False)
inner_data = [
[
"answer",
None,
["server-hl-default-conv", 12345],
None,
[[], None, None, [], 1],
]
]
inner_json = json.dumps(inner_data)
chunk_json = json.dumps([["wrb.fr", None, inner_json]])
response_body = f")]}}'\n{len(chunk_json)}\n{chunk_json}\n"
httpx_mock.add_response(
url=re.compile(r".*GenerateFreeFormStreamed.*"),
content=response_body.encode(),
method="POST",
)
mock_get_conversation_id()
async with NotebookLMClient(auth_tokens) as client:
await client.chat.ask(
notebook_id="test_nb",
question="Q",
source_ids=["src_001"],
)
chat_request = next(
r for r in httpx_mock.get_requests() if "GenerateFreeFormStreamed" in str(r.url)
)
assert "hl=en" in str(chat_request.url)
Reference in New Issue View Git Blame Copy Permalink