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
411 lines
12 KiB
Python
411 lines
12 KiB
Python
"""Unit tests for the private source listing service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from notebooklm._source.listing import SourceLister
|
|
from notebooklm.exceptions import RPCError
|
|
from notebooklm.rpc import RPCMethod
|
|
from notebooklm.rpc.types import SourceStatus
|
|
from notebooklm.types import Source
|
|
|
|
|
|
class RecordingRpc:
|
|
def __init__(self, response: Any) -> None:
|
|
self.response = response
|
|
self.calls: list[tuple[RPCMethod, list[Any], str | None]] = []
|
|
|
|
async def rpc_call(
|
|
self,
|
|
method: RPCMethod,
|
|
params: list[Any],
|
|
source_path: str = "/",
|
|
allow_null: bool = False,
|
|
_is_retry: bool = False,
|
|
*,
|
|
disable_internal_retries: bool = False,
|
|
operation_variant: str | None = None,
|
|
) -> Any:
|
|
self.calls.append((method, params, source_path))
|
|
return self.response
|
|
|
|
|
|
def source_entry(
|
|
source_id: str,
|
|
*,
|
|
title: str = "Source",
|
|
metadata: list[Any] | None = None,
|
|
status: list[Any] | None = None,
|
|
) -> list[Any]:
|
|
return [
|
|
[source_id],
|
|
title,
|
|
metadata or [None, 11, [1704067200, 0], None, 5],
|
|
status or [None, 2],
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_uses_exact_get_notebook_rpc_shape() -> None:
|
|
rpc = RecordingRpc([["Notebook", []]])
|
|
lister = SourceLister(rpc)
|
|
|
|
assert await lister.list("nb_123") == []
|
|
|
|
assert rpc.calls == [
|
|
(
|
|
RPCMethod.GET_NOTEBOOK,
|
|
# #1549: GET_NOTEBOOK tail migrated to the nested template block.
|
|
[
|
|
"nb_123",
|
|
None,
|
|
[2, None, None, [1, None, None, None, None, None, None, None, None, None, [1]]],
|
|
None,
|
|
0,
|
|
],
|
|
"/notebook/nb_123",
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("payload", "message"),
|
|
[
|
|
(None, "Empty or invalid notebook response"),
|
|
([], "Empty or invalid notebook response"),
|
|
(["notebook"], "Unexpected notebook structure"),
|
|
([["Notebook"]], "Unexpected notebook structure"),
|
|
([["Notebook", "not-a-list"]], "Sources data for nb_123 is not a list"),
|
|
],
|
|
)
|
|
async def test_malformed_payloads_log_and_raise(
|
|
payload: Any,
|
|
message: str,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
# Strict decoding is the only mode (the ``NOTEBOOKLM_STRICT_DECODE=0``
|
|
# soft-mode opt-out was retired in v0.7.0): a malformed payload always logs
|
|
# the diagnostic warning AND raises, rather than silently returning ``[]``
|
|
# (issue #1159).
|
|
lister = SourceLister(RecordingRpc(payload))
|
|
caplog.set_level("WARNING", logger="notebooklm._sources")
|
|
|
|
with pytest.raises(RPCError):
|
|
await lister.list("nb_123")
|
|
|
|
assert message in caplog.text
|
|
assert caplog.records[0].name == "notebooklm._sources"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("payload", "message"),
|
|
[
|
|
(None, "API response structure changed"),
|
|
([], "API response structure changed"),
|
|
(["notebook"], "API response structure changed"),
|
|
([["Notebook"]], "API response structure changed"),
|
|
([["Notebook", "not-a-list"]], "sources data is str, not list"),
|
|
],
|
|
)
|
|
async def test_strict_mode_raises_rpc_error_for_malformed_payloads(
|
|
payload: Any,
|
|
message: str,
|
|
) -> None:
|
|
lister = SourceLister(RecordingRpc(payload))
|
|
|
|
with pytest.raises(RPCError, match=message):
|
|
await lister.list("nb_123", strict=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("payload", "message"),
|
|
[
|
|
(None, "API response structure changed"),
|
|
([], "API response structure changed"),
|
|
(["notebook"], "API response structure changed"),
|
|
([["Notebook"]], "API response structure changed"),
|
|
([["Notebook", "not-a-list"]], "sources data is str, not list"),
|
|
],
|
|
)
|
|
async def test_malformed_payloads_raise_by_default_under_strict_decode(
|
|
payload: Any,
|
|
message: str,
|
|
caplog: pytest.LogCaptureFixture,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# Regression for issue #1159: with strict-decode at its default (ON), a
|
|
# malformed/error-shaped response must raise rather than silently return
|
|
# an empty list, even without an explicit ``strict=True``.
|
|
monkeypatch.delenv("NOTEBOOKLM_STRICT_DECODE", raising=False)
|
|
lister = SourceLister(RecordingRpc(payload))
|
|
caplog.set_level("WARNING", logger="notebooklm._sources")
|
|
|
|
with pytest.raises(RPCError, match=message):
|
|
await lister.list("nb_123")
|
|
|
|
# The drift WARNING on the historical "SourcesAPI.list:" prefix must still
|
|
# fire in strict mode so log-based monitoring keeps its breadcrumb.
|
|
assert "SourcesAPI.list:" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("strict", [False, True])
|
|
async def test_null_sources_slot_is_empty_notebook_not_malformed(
|
|
strict: bool,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# A genuinely empty notebook elides the sources slot as ``None`` (see
|
|
# tests/cassettes/notebook_zero_sources.yaml). Per issue #1159 this is a
|
|
# valid empty state and must return ``[]`` regardless of strict mode,
|
|
# never raising or warning.
|
|
monkeypatch.delenv("NOTEBOOKLM_STRICT_DECODE", raising=False)
|
|
lister = SourceLister(
|
|
RecordingRpc([["Notebook Title", None, "nb_123", "", None, [1, False], None, None, None]])
|
|
)
|
|
|
|
assert await lister.list("nb_123", strict=strict) == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_url_precedence_uses_index_7_then_5_and_ignores_bare_http() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry(
|
|
"src_both",
|
|
metadata=[
|
|
"https://bare.example/ignored",
|
|
11,
|
|
[1704067200, 0],
|
|
None,
|
|
5,
|
|
["https://index-5.example/source"],
|
|
None,
|
|
["https://index-7.example/source"],
|
|
],
|
|
),
|
|
source_entry(
|
|
"src_index_5",
|
|
metadata=[
|
|
"https://bare.example/ignored",
|
|
11,
|
|
[1704067200, 0],
|
|
None,
|
|
9,
|
|
["https://youtube.example/watch?v=abc"],
|
|
None,
|
|
None,
|
|
],
|
|
),
|
|
source_entry(
|
|
"src_bare",
|
|
metadata=[
|
|
"https://bare.example/ignored",
|
|
11,
|
|
[1704067200, 0],
|
|
None,
|
|
5,
|
|
None,
|
|
None,
|
|
None,
|
|
],
|
|
),
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
sources = await lister.list("nb_123")
|
|
|
|
assert [source.url for source in sources] == [
|
|
"https://index-7.example/source",
|
|
"https://youtube.example/watch?v=abc",
|
|
None,
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_and_type_code_parsing() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry("src_processing", status=[None, SourceStatus.PROCESSING]),
|
|
source_entry("src_unknown_status", status=[None, 999]),
|
|
source_entry(
|
|
"src_non_int_type",
|
|
metadata=[None, 11, [1704067200, 0], None, "5"],
|
|
),
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
sources = await lister.list("nb_123")
|
|
|
|
assert sources[0].status == SourceStatus.PROCESSING
|
|
assert sources[0]._type_code == 5
|
|
assert sources[1].status == SourceStatus.READY
|
|
assert sources[2]._type_code is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nested_drive_source_id_is_extracted() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
[
|
|
[None, True, ["drive_src"]],
|
|
"Drive Source",
|
|
[None, 11, [1704067200, 0], None, 2],
|
|
[None, 2],
|
|
],
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
sources = await lister.list("nb_123")
|
|
|
|
assert len(sources) == 1
|
|
assert sources[0].id == "drive_src"
|
|
assert sources[0].title == "Drive Source"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_malformed_source_id_shape_logs_and_skips(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry("src_valid", title="Valid"),
|
|
[
|
|
[None, True, []],
|
|
"Broken",
|
|
[None, 11, [1704067200, 0], None, 2],
|
|
[None, 2],
|
|
],
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
caplog.set_level("WARNING", logger="notebooklm._sources")
|
|
|
|
sources = await lister.list("nb_123")
|
|
|
|
assert [source.id for source in sources] == ["src_valid"]
|
|
assert "Skipping source with unexpected id shape" in caplog.text
|
|
assert "[None, True, []]" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_created_at_uses_shared_timestamp_parser() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry(
|
|
"src_timestamp",
|
|
metadata=[None, 11, [1704067200, 123], None, 5],
|
|
),
|
|
source_entry(
|
|
"src_bad_timestamp",
|
|
metadata=[None, 11, [None], None, 5],
|
|
),
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
sources = await lister.list("nb_123")
|
|
|
|
assert sources[0].created_at is not None
|
|
assert int(sources[0].created_at.timestamp()) == 1704067200
|
|
assert sources[1].created_at is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_filters_list_results() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry("src_1", title="One"),
|
|
source_entry("src_2", title="Two"),
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
source = await lister.get("nb_123", "src_2")
|
|
|
|
assert source is not None
|
|
assert source.id == "src_2"
|
|
assert source.title == "Two"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_returns_none_when_source_not_found() -> None:
|
|
lister = SourceLister(
|
|
RecordingRpc(
|
|
[
|
|
[
|
|
"Notebook",
|
|
[
|
|
source_entry("src_1", title="One"),
|
|
source_entry("src_2", title="Two"),
|
|
],
|
|
]
|
|
]
|
|
)
|
|
)
|
|
|
|
assert await lister.get("nb_123", "missing") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_can_use_late_bound_list_hook() -> None:
|
|
lister = SourceLister(RecordingRpc([["Notebook", []]]))
|
|
calls: list[str] = []
|
|
|
|
async def list_sources(notebook_id: str):
|
|
calls.append(notebook_id)
|
|
return [
|
|
Source(id="src_1", title="One"),
|
|
Source(id="src_2", title="Two"),
|
|
]
|
|
|
|
source = await lister.get("nb_123", "src_2", list_sources=list_sources)
|
|
|
|
assert calls == ["nb_123"]
|
|
assert source is not None
|
|
assert source.id == "src_2"
|