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
713 lines
28 KiB
Python
713 lines
28 KiB
Python
"""Unit tests for E2E conftest CLI options.
|
|
|
|
Covers the --profile flag added in issue #339 without spinning up the full
|
|
E2E suite (which requires real auth).
|
|
|
|
The E2E conftest is loaded by file path so these unit tests can execute a fresh
|
|
copy of the hook module without invoking pytest's conftest discovery or the
|
|
authenticated E2E suite.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
from types import ModuleType, SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from notebooklm.exceptions import RateLimitError
|
|
|
|
CONFTEST_PATH = Path(__file__).resolve().parents[1] / "e2e" / "conftest.py"
|
|
pytest_plugins = ["pytester"]
|
|
|
|
|
|
def _load_e2e_conftest() -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location("e2e_conftest", CONFTEST_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _make_config(profile: str | None) -> SimpleNamespace:
|
|
return SimpleNamespace(getoption=lambda name: profile if name == "--profile" else None)
|
|
|
|
|
|
class _FakeItem:
|
|
def __init__(self, path: Path) -> None:
|
|
self.path = path
|
|
self.markers: list[object] = []
|
|
|
|
def add_marker(self, marker: object) -> None:
|
|
self.markers.append(marker)
|
|
|
|
|
|
class TestE2EMarkerContract:
|
|
"""E2E files are marked before pytest applies -m deselection."""
|
|
|
|
def test_item_under_e2e_directory_gets_e2e_marker(self):
|
|
conftest = _load_e2e_conftest()
|
|
item = _FakeItem(conftest.E2E_TEST_DIR / "test_chat.py")
|
|
|
|
conftest.pytest_itemcollected(item)
|
|
|
|
assert [marker.name for marker in item.markers] == ["e2e"]
|
|
|
|
def test_item_outside_e2e_directory_is_not_marked(self):
|
|
conftest = _load_e2e_conftest()
|
|
item = _FakeItem(Path(__file__))
|
|
|
|
conftest.pytest_itemcollected(item)
|
|
|
|
assert item.markers == []
|
|
|
|
def test_path_helper_uses_resolved_containment(self):
|
|
conftest = _load_e2e_conftest()
|
|
|
|
assert conftest._is_path_under(
|
|
conftest.E2E_TEST_DIR / "test_chat.py", conftest.E2E_TEST_DIR
|
|
)
|
|
assert not conftest._is_path_under(Path(__file__), conftest.E2E_TEST_DIR)
|
|
|
|
|
|
class TestProfileOptionLifecycle:
|
|
"""pytest_configure + pytest_unconfigure round-trip."""
|
|
|
|
def test_round_trip_no_prior_env(self, monkeypatch):
|
|
monkeypatch.delenv("NOTEBOOKLM_PROFILE", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
config = _make_config("work")
|
|
|
|
conftest.pytest_configure(config)
|
|
assert os.environ.get("NOTEBOOKLM_PROFILE") == "work"
|
|
|
|
conftest.pytest_unconfigure(config)
|
|
assert "NOTEBOOKLM_PROFILE" not in os.environ
|
|
|
|
def test_round_trip_restores_prior_env(self, monkeypatch):
|
|
monkeypatch.setenv("NOTEBOOKLM_PROFILE", "preset")
|
|
conftest = _load_e2e_conftest()
|
|
config = _make_config("work")
|
|
|
|
conftest.pytest_configure(config)
|
|
assert os.environ.get("NOTEBOOKLM_PROFILE") == "work"
|
|
|
|
conftest.pytest_unconfigure(config)
|
|
assert os.environ.get("NOTEBOOKLM_PROFILE") == "preset"
|
|
|
|
def test_no_flag_with_prior_env_is_noop(self, monkeypatch):
|
|
monkeypatch.setenv("NOTEBOOKLM_PROFILE", "preset")
|
|
conftest = _load_e2e_conftest()
|
|
config = _make_config(None)
|
|
|
|
conftest.pytest_configure(config)
|
|
conftest.pytest_unconfigure(config)
|
|
assert os.environ.get("NOTEBOOKLM_PROFILE") == "preset"
|
|
|
|
def test_no_flag_without_prior_env_is_noop(self, monkeypatch):
|
|
monkeypatch.delenv("NOTEBOOKLM_PROFILE", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
config = _make_config(None)
|
|
|
|
conftest.pytest_configure(config)
|
|
conftest.pytest_unconfigure(config)
|
|
assert "NOTEBOOKLM_PROFILE" not in os.environ
|
|
|
|
|
|
class TestArgvProfile:
|
|
"""Parsing of --profile out of argv (used at import time)."""
|
|
|
|
def test_long_form(self):
|
|
argv = ["pytest", "--profile", "work", "tests/e2e"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) == "work"
|
|
|
|
def test_equals_form(self):
|
|
argv = ["pytest", "--profile=work", "tests/e2e"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) == "work"
|
|
|
|
def test_absent(self):
|
|
argv = ["pytest", "tests/e2e", "-m", "e2e"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) is None
|
|
|
|
def test_long_form_missing_value_returns_none(self):
|
|
argv = ["pytest", "--profile"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) is None
|
|
|
|
def test_last_occurrence_wins(self):
|
|
argv = ["pytest", "--profile", "foo", "--profile", "bar"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) == "bar"
|
|
|
|
def test_long_form_rejects_dash_prefixed_value(self):
|
|
argv = ["pytest", "--profile", "--verbose"]
|
|
assert _load_e2e_conftest()._argv_profile(argv) is None
|
|
|
|
|
|
class TestRateLimitSkipSummary:
|
|
"""pytest_terminal_summary surfaces chat rate-limit skips so green CI doesn't hide drift."""
|
|
|
|
@staticmethod
|
|
def _make_reporter(
|
|
reports,
|
|
*,
|
|
passed=None,
|
|
failed=None,
|
|
):
|
|
write_calls: list[tuple] = []
|
|
return SimpleNamespace(
|
|
stats={"skipped": reports, "passed": passed or [], "failed": failed or []},
|
|
write_sep=lambda *a, **kw: write_calls.append(("sep", a, kw)),
|
|
write_line=lambda *a, **kw: write_calls.append(("line", a, kw)),
|
|
_writes=write_calls,
|
|
)
|
|
|
|
@staticmethod
|
|
def _make_session(reporter, *, exitstatus=pytest.ExitCode.OK):
|
|
pluginmanager = SimpleNamespace(
|
|
get_plugin=lambda name: reporter if name == "terminalreporter" else None
|
|
)
|
|
return SimpleNamespace(
|
|
config=SimpleNamespace(pluginmanager=pluginmanager),
|
|
exitstatus=exitstatus,
|
|
)
|
|
|
|
@classmethod
|
|
def _finish(
|
|
cls,
|
|
conftest,
|
|
reporter,
|
|
*,
|
|
exitstatus=pytest.ExitCode.OK,
|
|
) -> SimpleNamespace:
|
|
session = cls._make_session(reporter, exitstatus=exitstatus)
|
|
conftest.pytest_sessionfinish(session, exitstatus)
|
|
return session
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _enforce_floor(self, monkeypatch):
|
|
# Default these tests to the enforced path (nightly semantics) with inline
|
|
# exit-status delivery (no sentinel), matching the pre-existing assertions.
|
|
# Individual tests override (delenv the flag, or set the sentinel path).
|
|
monkeypatch.setenv("E2E_ENFORCE_COVERAGE_FLOOR", "1")
|
|
monkeypatch.delenv("E2E_COVERAGE_FLOOR_SENTINEL", raising=False)
|
|
|
|
@staticmethod
|
|
def _report(
|
|
nodeid: str,
|
|
*,
|
|
live_chat_ask: bool = False,
|
|
live_generation: bool = False,
|
|
when: str = "call",
|
|
):
|
|
keywords: dict[str, int] = {}
|
|
if live_chat_ask:
|
|
keywords["live_chat_ask"] = 1
|
|
if live_generation:
|
|
keywords["live_generation"] = 1
|
|
return SimpleNamespace(nodeid=nodeid, keywords=keywords, when=when)
|
|
|
|
@classmethod
|
|
def _skipped(
|
|
cls,
|
|
nodeid: str,
|
|
reason: str,
|
|
*,
|
|
live_chat_ask: bool = False,
|
|
live_generation: bool = False,
|
|
when: str = "call",
|
|
) -> SimpleNamespace:
|
|
report = cls._report(
|
|
nodeid, live_chat_ask=live_chat_ask, live_generation=live_generation, when=when
|
|
)
|
|
report.longrepr = ("file.py", 1, f"Skipped: {reason}")
|
|
return report
|
|
|
|
@classmethod
|
|
def _passed(
|
|
cls,
|
|
nodeid: str,
|
|
*,
|
|
live_chat_ask: bool = False,
|
|
live_generation: bool = False,
|
|
when: str = "call",
|
|
) -> SimpleNamespace:
|
|
return cls._report(
|
|
nodeid, live_chat_ask=live_chat_ask, live_generation=live_generation, when=when
|
|
)
|
|
|
|
def test_counts_only_rate_limit_skips(self, monkeypatch, tmp_path, capsys):
|
|
monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary.md"))
|
|
monkeypatch.setenv("GITHUB_ACTIONS", "true")
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[
|
|
self._skipped("t::a", "Chat request was rate limited"),
|
|
self._skipped("t::b", "no auth configured"),
|
|
self._skipped("t::c", "rejected by the API"),
|
|
self._skipped("t::d", "Chat request failed with HTTP 429: ..."),
|
|
self._skipped("t::e", "Too Many Requests"),
|
|
self._skipped("t::f", "chat rate-limited by upstream"),
|
|
]
|
|
)
|
|
conftest.pytest_terminal_summary(tr, 0, None)
|
|
|
|
summary = (tmp_path / "summary.md").read_text()
|
|
assert "Rate-limit skips: 5" in summary
|
|
assert all(nid in summary for nid in ("t::a", "t::c", "t::d", "t::e", "t::f"))
|
|
assert "t::b" not in summary
|
|
assert "::warning::5 test(s) skipped" in capsys.readouterr().out
|
|
|
|
def test_no_skips_emits_nothing(self, monkeypatch, tmp_path, capsys):
|
|
monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(tmp_path / "summary.md"))
|
|
monkeypatch.setenv("GITHUB_ACTIONS", "true")
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter([self._skipped("t::a", "no auth configured")])
|
|
conftest.pytest_terminal_summary(tr, 0, None)
|
|
|
|
assert not (tmp_path / "summary.md").exists()
|
|
assert capsys.readouterr().out == ""
|
|
assert tr._writes == []
|
|
|
|
def test_no_github_env_skips_annotations(self, monkeypatch, tmp_path, capsys):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter([self._skipped("t::a", "rate limited")])
|
|
conftest.pytest_terminal_summary(tr, 0, None)
|
|
|
|
# Still emits the pytest section locally — just no GH-specific bits.
|
|
assert any(call[0] == "sep" for call in tr._writes)
|
|
assert capsys.readouterr().out == ""
|
|
|
|
def test_live_chat_floor_fails_when_marked_asks_all_rate_limited(self, monkeypatch, capsys):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[
|
|
self._skipped(
|
|
"test_server_live.py::TestRestServerLiveChat::test_chat_ask_returns_answer",
|
|
"chat rate-limited (surfaced through the REST route)",
|
|
live_chat_ask=True,
|
|
)
|
|
]
|
|
)
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.OK, None)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
|
|
assert any(
|
|
call[0] == "sep" and call[1][1] == "live chat coverage floor failed"
|
|
for call in tr._writes
|
|
)
|
|
assert capsys.readouterr().out == ""
|
|
|
|
def test_live_chat_floor_allows_one_marked_pass(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_chat.py::test_rate_limited", "rate limited", live_chat_ask=True)],
|
|
passed=[self._passed("test_chat.py::test_ok", live_chat_ask=True)],
|
|
)
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.OK, None)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
assert not any(
|
|
call[0] == "sep" and call[1][1] == "live chat coverage floor failed"
|
|
for call in tr._writes
|
|
)
|
|
|
|
def test_live_chat_floor_ignores_unmarked_passes(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_chat.py::test_rate_limited", "rate limited", live_chat_ask=True)],
|
|
passed=[self._passed("test_unrelated.py::test_ok")],
|
|
)
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.OK, None)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
|
|
|
|
def test_live_chat_floor_ignores_unmarked_rate_limit_skips(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter([self._skipped("test_generation.py::test_audio", "rate limited")])
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.OK, None)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
|
|
def test_live_chat_floor_counts_setup_rate_limit_skips(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[
|
|
self._skipped(
|
|
"test_chat.py::test_auth_skip",
|
|
"rate limited",
|
|
live_chat_ask=True,
|
|
when="setup",
|
|
)
|
|
]
|
|
)
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.OK, None)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
|
|
|
|
def test_live_chat_floor_does_not_rewrite_non_green_exitstatus(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
|
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_chat.py::test_rate_limited", "rate limited", live_chat_ask=True)],
|
|
)
|
|
conftest.pytest_terminal_summary(tr, pytest.ExitCode.USAGE_ERROR, None)
|
|
session = self._finish(conftest, tr, exitstatus=pytest.ExitCode.USAGE_ERROR)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.USAGE_ERROR
|
|
assert not any(
|
|
call[0] == "sep" and call[1][1] == "live chat coverage floor failed"
|
|
for call in tr._writes
|
|
)
|
|
|
|
def test_floor_not_enforced_without_flag_stays_green(self, monkeypatch):
|
|
# Release-path semantics (#1819): with E2E_ENFORCE_COVERAGE_FLOOR unset, a
|
|
# fully-throttled surface skips freely and never reds the job.
|
|
monkeypatch.delenv("E2E_ENFORCE_COVERAGE_FLOOR", raising=False)
|
|
monkeypatch.delenv("E2E_COVERAGE_FLOOR_SENTINEL", raising=False)
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter([self._skipped("t::a", "rate limited", live_chat_ask=True)])
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
|
|
def test_generation_floor_fails_when_all_marked_rate_limited(self, monkeypatch):
|
|
conftest = _load_e2e_conftest()
|
|
tr = self._make_reporter(
|
|
[
|
|
self._skipped(
|
|
"test_interactive_mind_map.py::t", "Rate limit: quota", live_generation=True
|
|
)
|
|
]
|
|
)
|
|
session = self._finish(conftest, tr)
|
|
assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
|
|
|
|
def test_generation_floor_allows_one_marked_pass(self, monkeypatch):
|
|
conftest = _load_e2e_conftest()
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::t_audio", "Rate limit: q", live_generation=True)],
|
|
passed=[self._passed("test_generation.py::t_report", live_generation=True)],
|
|
)
|
|
session = self._finish(conftest, tr)
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
|
|
def test_sentinel_records_skip_event_without_failing_exit(self, monkeypatch, tmp_path):
|
|
# Nightly delivery: a hollow surface (marked skip, no marked pass) appends a
|
|
# SKIP event and leaves exit status alone, so a pure-skip run stays exit 0 and
|
|
# doesn't trip the continue-on-error retry. The enforce step reconciles later.
|
|
sentinel = tmp_path / "floor"
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(sentinel))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::t", "Rate limit: q", live_generation=True)]
|
|
)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
assert "SKIP\tlive generation" in sentinel.read_text()
|
|
|
|
def test_sentinel_records_skip_despite_unrelated_failure(self, monkeypatch, tmp_path):
|
|
# Masking guard (#1819): the SKIP event is still recorded when an UNRELATED
|
|
# test failed on the main run, so a transient failure that recovers on retry
|
|
# can't mask a hollow surface.
|
|
sentinel = tmp_path / "floor"
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(sentinel))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::t", "Rate limit: q", live_generation=True)],
|
|
failed=[self._report("test_notebooks.py::unrelated", when="call")],
|
|
)
|
|
self._finish(conftest, tr, exitstatus=pytest.ExitCode.TESTS_FAILED)
|
|
|
|
assert "SKIP\tlive generation" in sentinel.read_text()
|
|
|
|
def test_sentinel_records_pass_event_when_marked_test_passed(self, monkeypatch, tmp_path):
|
|
# A marked pass records PASS (real coverage) even when another marked test
|
|
# rate-limit-skipped in the same run — the enforce step reconciles SKIP vs PASS
|
|
# so this surface does NOT breach.
|
|
sentinel = tmp_path / "floor"
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(sentinel))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::a", "Rate limit: q", live_generation=True)],
|
|
passed=[self._passed("test_generation.py::b", live_generation=True)],
|
|
)
|
|
self._finish(conftest, tr)
|
|
|
|
text = sentinel.read_text()
|
|
assert "PASS\tlive generation" in text
|
|
assert "SKIP\tlive generation" not in text
|
|
|
|
def test_sentinel_records_skip_when_marked_test_failed(self, monkeypatch, tmp_path):
|
|
# A marked FAILURE is not a pass, so the surface still records SKIP (a breach
|
|
# candidate). The retry step appends its own PASS/SKIP, and the enforce step
|
|
# reconciles — so a marked test that fails on main then passes on retry clears
|
|
# the surface, while one that fails then skips keeps it breached (codex/coderabbit).
|
|
sentinel = tmp_path / "floor"
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(sentinel))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::a", "Rate limit: q", live_generation=True)],
|
|
failed=[self._report("test_generation.py::b", live_generation=True, when="setup")],
|
|
)
|
|
self._finish(conftest, tr, exitstatus=pytest.ExitCode.TESTS_FAILED)
|
|
|
|
assert "SKIP\tlive generation" in sentinel.read_text()
|
|
|
|
def test_floor_falls_back_to_exitstatus_when_sentinel_unwritable(self, monkeypatch, tmp_path):
|
|
# If the sentinel write fails, don't silently hollow-green: fall back to the
|
|
# inline exit-code path (best-effort) (codex). Point the sentinel at a path
|
|
# whose parent is a file, so makedirs + open both fail.
|
|
blocker = tmp_path / "blocker"
|
|
blocker.write_text("not a dir")
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(blocker / "floor"))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::a", "Rate limit: q", live_generation=True)]
|
|
)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.TESTS_FAILED
|
|
|
|
def test_floor_sentinel_creates_missing_parent_dir(self, monkeypatch, tmp_path):
|
|
# A missing parent directory must not turn the enforcement signal into a
|
|
# silent no-op — the writer creates it (gemini).
|
|
sentinel = tmp_path / "nested" / "sub" / "floor"
|
|
monkeypatch.setenv("E2E_COVERAGE_FLOOR_SENTINEL", str(sentinel))
|
|
conftest = _load_e2e_conftest()
|
|
|
|
tr = self._make_reporter(
|
|
[self._skipped("test_generation.py::a", "Rate limit: q", live_generation=True)]
|
|
)
|
|
session = self._finish(conftest, tr)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.OK
|
|
assert "SKIP\tlive generation" in sentinel.read_text()
|
|
|
|
def test_floor_ignores_broken_run_exit_codes(self, monkeypatch):
|
|
# A usage/internal error is its own signal — the floor must not evaluate or
|
|
# rewrite that exit code (claude review).
|
|
conftest = _load_e2e_conftest()
|
|
tr = self._make_reporter(
|
|
[self._skipped("t::a", "rate limited", live_chat_ask=True)],
|
|
)
|
|
session = self._finish(conftest, tr, exitstatus=pytest.ExitCode.USAGE_ERROR)
|
|
|
|
assert session.exitstatus == pytest.ExitCode.USAGE_ERROR
|
|
|
|
def test_live_chat_floor_changes_real_pytest_exit_code(self, pytester: pytest.Pytester):
|
|
pytester.makepyprojecttoml(
|
|
"""
|
|
[tool.pytest.ini_options]
|
|
markers = [
|
|
"live_chat_ask: chat ask floor marker",
|
|
]
|
|
"""
|
|
)
|
|
pytester.makeconftest(
|
|
"""
|
|
import pytest
|
|
|
|
def pytest_sessionfinish(session, exitstatus):
|
|
terminalreporter = session.config.pluginmanager.get_plugin("terminalreporter")
|
|
if exitstatus == pytest.ExitCode.OK and terminalreporter.stats.get("skipped"):
|
|
session.exitstatus = pytest.ExitCode.TESTS_FAILED
|
|
|
|
def pytest_terminal_summary(terminalreporter, exitstatus, config):
|
|
if exitstatus == pytest.ExitCode.OK and terminalreporter.stats.get("skipped"):
|
|
terminalreporter.write_sep("=", "live chat coverage floor failed")
|
|
"""
|
|
)
|
|
pytester.makepyfile(
|
|
test_floor="""
|
|
import pytest
|
|
|
|
@pytest.mark.live_chat_ask
|
|
def test_all_chat_asks_rate_limited():
|
|
pytest.skip("chat rate-limited by upstream")
|
|
"""
|
|
)
|
|
|
|
result = pytester.runpytest_subprocess("-q")
|
|
|
|
assert result.ret == pytest.ExitCode.TESTS_FAILED
|
|
result.stdout.fnmatch_lines(["*live chat coverage floor failed*"])
|
|
|
|
|
|
class TestGenerationRateLimitSkip:
|
|
"""_install_generation_rate_limit_skip turns typed RateLimitError into skips.
|
|
|
|
The RPC layer raises RateLimitError from generate_* before any
|
|
GenerationStatus exists, so assert_generation_started's is_rate_limited
|
|
path never runs. Only the typed RateLimitError may skip; every other
|
|
exception must propagate (no-xfail-live-service-errors policy).
|
|
"""
|
|
|
|
@staticmethod
|
|
def _make_client():
|
|
class FakeArtifacts:
|
|
async def generate_audio(self, notebook_id):
|
|
raise RateLimitError(
|
|
"API rate limit or quota exceeded. Please wait before retrying."
|
|
)
|
|
|
|
async def generate_video(self, notebook_id):
|
|
return f"video:{notebook_id}"
|
|
|
|
async def revise_slide(self, notebook_id):
|
|
raise ValueError("not a rate limit")
|
|
|
|
async def retry_failed(self, notebook_id, artifact_id):
|
|
raise RateLimitError("Resource exhausted (status 8): quota")
|
|
|
|
async def delete(self, notebook_id, artifact_id):
|
|
raise RateLimitError("should never be wrapped")
|
|
|
|
class FakeMindMaps:
|
|
async def generate(self, notebook_id):
|
|
raise RateLimitError("Resource exhausted (status 8): quota")
|
|
|
|
async def list(self, notebook_id):
|
|
raise RateLimitError("read path — should never be wrapped")
|
|
|
|
return SimpleNamespace(artifacts=FakeArtifacts(), mind_maps=FakeMindMaps())
|
|
|
|
async def test_rate_limit_error_becomes_skip(self):
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(pytest.skip.Exception) as excinfo:
|
|
await client.artifacts.generate_audio("nb-1")
|
|
|
|
# Reason must match _RATE_LIMIT_PHRASES so pytest_terminal_summary
|
|
# surfaces the skip in the rate-limit section + GH annotations.
|
|
reason = str(excinfo.value).lower()
|
|
assert any(phrase in reason for phrase in conftest._RATE_LIMIT_PHRASES)
|
|
|
|
async def test_other_exceptions_propagate(self):
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(ValueError, match="not a rate limit"):
|
|
await client.artifacts.revise_slide("nb-1")
|
|
|
|
async def test_successful_calls_pass_through_per_method(self):
|
|
# Closure safety: each wrapped name must bind its own original.
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
assert await client.artifacts.generate_video("nb-1") == "video:nb-1"
|
|
|
|
async def test_non_generation_methods_are_not_wrapped(self):
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(RateLimitError):
|
|
await client.artifacts.delete("nb-1", "art-1")
|
|
|
|
async def test_mind_maps_generate_becomes_skip(self):
|
|
# The #1819 gap: the interactive mind map creates via client.mind_maps.generate,
|
|
# a different namespace than client.artifacts.generate_* — it must skip too.
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(pytest.skip.Exception) as excinfo:
|
|
await client.mind_maps.generate("nb-1")
|
|
assert any(p in str(excinfo.value).lower() for p in conftest._RATE_LIMIT_PHRASES)
|
|
|
|
async def test_mind_maps_read_methods_are_not_wrapped(self):
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(RateLimitError):
|
|
await client.mind_maps.list("nb-1")
|
|
|
|
async def test_retry_failed_becomes_skip(self):
|
|
# retry_failed re-runs generation and raises RateLimitError on quota, so it
|
|
# must skip like generate_* (claude review — same #1819 regression class).
|
|
conftest = _load_e2e_conftest()
|
|
client = self._make_client()
|
|
conftest._install_generation_rate_limit_skip(client)
|
|
|
|
with pytest.raises(pytest.skip.Exception):
|
|
await client.artifacts.retry_failed("nb-1", "art-1")
|
|
|
|
|
|
class TestGenerationSkipRegistryCoverage:
|
|
"""The skip registry must cover every live generate/revise entrypoint.
|
|
|
|
Guards against a new CREATE_ARTIFACT method (or namespace) landing without
|
|
being wrapped for the RateLimitError→skip fixture — the exact regression class
|
|
that caused #1819 (mind_maps.generate was created after the artifacts-only
|
|
wrapper and slipped through).
|
|
"""
|
|
|
|
def test_registry_covers_all_generate_and_revise_methods(self):
|
|
from notebooklm._artifacts import ArtifactsAPI
|
|
from notebooklm._mind_maps_api import MindMapsAPI
|
|
|
|
conftest = _load_e2e_conftest()
|
|
classes = {"artifacts": ArtifactsAPI, "mind_maps": MindMapsAPI}
|
|
|
|
# Registry namespaces and the known generation-capable classes must stay in
|
|
# lockstep — adding one without the other trips this guard.
|
|
assert set(conftest._GENERATION_SKIP_TARGETS) == set(classes)
|
|
|
|
for ns_name, cls in classes.items():
|
|
matches = conftest._GENERATION_SKIP_TARGETS[ns_name]
|
|
# generate*/revise*/retry* are the CREATE_ARTIFACT-quota surfaces that
|
|
# raise RateLimitError before a GenerationStatus exists (retry_failed
|
|
# re-runs generation, so it raises on quota too).
|
|
gen_methods = [
|
|
name
|
|
for name in dir(cls)
|
|
if not name.startswith("_")
|
|
and name.startswith(("generate", "revise", "retry"))
|
|
and callable(getattr(cls, name))
|
|
]
|
|
assert gen_methods, f"{ns_name}: expected at least one generate/revise/retry method"
|
|
uncovered = [name for name in gen_methods if not matches(name)]
|
|
assert not uncovered, (
|
|
f"{ns_name}: generate/revise/retry methods not in skip registry: {uncovered}"
|
|
)
|