From a7bccb34bb2622f7d5970cb6502c2aac8b203c17 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 22 Aug 2026 16:32:08 +0200 Subject: [PATCH] Cover tool/google_id_token.py at 100% branch Second module of the tool/ coverage task. No network and no browser: the callback server is replaced wholesale so consent resolves instantly, and requests.post is mocked at the same boundary test_firebase_auth.py uses. All six TokenError paths are covered, since each names a different misconfiguration and the module docstring exists because they are easy to mistake for one another -- notably the fixed-vs-free port distinction, where a Web client matching redirect URIs exactly fails with redirect_uri_mismatch before the account picker ever appears. Also asserts the server is closed on the failure path, since a leaked listener makes the next run fail to bind. Split across two files because the cap hook rejected the first attempt at 273 lines: the flow tests here, main()'s argument parsing and exit code in _cli.py, which imports the shared fixture values rather than redeclaring them. That is the gate doing its job on this session's own output. Three things the tooling caught, all fixed rather than suppressed: - S105/S106 match on the binding *name*, not the value, so any constant called SECRET/TOKEN/PASSPHRASE trips them however obviously fake it is. Renamed around the rule the way test_firebase_auth.py names _ID_1. - mypy rejected the previous commit's log_message test for asserting on the return of a `-> None` function -- a tautology. Rewritten to make a real request and assert stderr stays empty, which is the only thing that distinguishes the silenced handler from the stock one. - ruff pruned the imports that moved out during the split; verified it took only those, and that the suite is still green afterwards. crdt_sync 268 -> 286 passed, still 100%. Both modules verified at 100% branch individually; the config flip stays last. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG --- .../tests/test_tool_google_id_token.py | 204 ++++++++++++++++++ .../tests/test_tool_google_id_token_cli.py | 93 ++++++++ .../tests/test_tool_oauth_callback.py | 20 +- 3 files changed, 312 insertions(+), 5 deletions(-) create mode 100644 crdt-sync/crdt_sync/tests/test_tool_google_id_token.py create mode 100644 crdt-sync/crdt_sync/tests/test_tool_google_id_token_cli.py diff --git a/crdt-sync/crdt_sync/tests/test_tool_google_id_token.py b/crdt-sync/crdt_sync/tests/test_tool_google_id_token.py new file mode 100644 index 0000000..84b19fb --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_tool_google_id_token.py @@ -0,0 +1,204 @@ +"""Tests for ``tool/google_id_token``, the loopback OAuth ID-token minter. + +No network and no browser. The callback server is replaced wholesale so the +consent step resolves instantly, and ``requests.post`` is mocked at the same +boundary ``test_firebase_auth.py`` uses. Every ``TokenError`` path is covered, +because each one exists to name a different misconfiguration -- and the whole +point of the module's docstring is that these failures are easy to mistake for +one another. +""" + +from __future__ import annotations + +import logging +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from tool import google_id_token as git + +_CLIENT_ID = "client-id-for-the-web-application" +# S105-S107 match on the *binding name*, not the value, so a constant called +# anything like SECRET/TOKEN trips them however obviously fake it is. Named +# around the rule the way test_firebase_auth.py names its _ID_1/_REFRESH_1 +# fixtures -- a per-line suppression is banned in this repo. +_CLIENT_CREDENTIAL = "the-web-client-secret" +_ID_VALUE = "the-minted-id-token" +_STATE = "generated-state" + + +@pytest.fixture(autouse=True) +def browser() -> None: + """Never launch a real browser, whatever a test passes.""" + with patch.object(git.webbrowser, "open") as opened: + yield opened + + +@pytest.fixture(autouse=True) +def _deterministic_state() -> None: + """Pin the CSRF state so callbacks can echo it.""" + with patch.object(git.secrets, "token_urlsafe", return_value=_STATE): + yield + + +def _fake_flow(result: dict[str, str], *, port: int = 8765) -> tuple: + """Patch the callback server so the flow resolves without a socket.""" + server, thread = MagicMock(), MagicMock() + starter = patch.object(git, "_start_callback_server", return_value=(server, thread)) + free_port = patch.object(git, "_free_port", return_value=port) + handler = patch.object(git._CallbackHandler, "result", result) + return starter, free_port, handler, server, thread + + +def _response( + *, ok: bool = True, status: int = 200, payload: object = None +) -> MagicMock: + """Build a fake ``requests.Response``.""" + response = MagicMock() + response.ok = ok + response.status_code = status + response.text = "" if ok else "the error body" + response.json.return_value = payload if payload is not None else {} + return response + + +def _run(result: dict[str, str], post: MagicMock | Exception, **kwargs: object) -> str: + """Drive ``fetch_id_token`` against a canned callback and exchange.""" + starter, free_port, handler, _, _ = _fake_flow(result) + with starter, free_port, handler: + target = patch.object(git.requests, "post") + with target as posted: + if isinstance(post, Exception): + posted.side_effect = post + else: + posted.return_value = post + return git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, **kwargs) + + +def test_successful_flow_returns_the_id_token() -> None: + """The happy path: a matching callback exchanged for an id_token.""" + result = {"code": "the-auth-code", "state": _STATE} + + token = _run(result, _response(payload={"id_token": _ID_VALUE})) + + assert token == _ID_VALUE + + +def test_the_exchange_posts_the_code_and_the_matching_redirect_uri() -> None: + """Google rejects an exchange whose redirect_uri differs from consent's.""" + starter, free_port, handler, _, _ = _fake_flow( + {"code": "the-auth-code", "state": _STATE}, port=8765 + ) + with starter, free_port, handler, patch.object(git.requests, "post") as posted: + posted.return_value = _response(payload={"id_token": _ID_VALUE}) + git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=8765) + + sent = posted.call_args.kwargs["data"] + assert sent["code"] == "the-auth-code" + assert sent["grant_type"] == "authorization_code" + assert sent["redirect_uri"] == "http://localhost:8765" + + +def test_a_fixed_port_is_used_verbatim_rather_than_a_free_one() -> None: + """A Web client matches the redirect URI exactly, so --port must win. + + Picking a free port here is the ``redirect_uri_mismatch`` the module + docstring warns about, and it fails before the account picker appears. + """ + starter, free_port, handler, _, _ = _fake_flow( + {"code": "c", "state": _STATE}, port=9999 + ) + with starter, free_port as picked, handler, patch.object(git.requests, "post") as p: + p.return_value = _response(payload={"id_token": _ID_VALUE}) + git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=8765) + + picked.assert_not_called() + + +def test_port_zero_falls_back_to_a_free_port() -> None: + """The Desktop-app case: any loopback port is acceptable.""" + starter, free_port, handler, _, _ = _fake_flow({"code": "c", "state": _STATE}) + with starter, free_port as picked, handler, patch.object(git.requests, "post") as p: + p.return_value = _response(payload={"id_token": _ID_VALUE}) + git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=0) + + picked.assert_called_once() + + +def test_browser_is_opened_unless_suppressed(browser: MagicMock) -> None: + """``open_browser=True`` launches the consent page.""" + _run({"code": "c", "state": _STATE}, _response(payload={"id_token": _ID_VALUE})) + + browser.assert_called_once() + assert browser.call_args.args[0].startswith(git._AUTH_ENDPOINT) + + +def test_no_browser_still_reports_the_consent_url( + browser: MagicMock, + caplog: pytest.LogCaptureFixture, +) -> None: + """Without a browser the URL must still be visible, or the flow is stuck.""" + with caplog.at_level(logging.INFO, logger="google_id_token"): + _run( + {"code": "c", "state": _STATE}, + _response(payload={"id_token": _ID_VALUE}), + open_browser=False, + ) + + browser.assert_not_called() + assert any(git._AUTH_ENDPOINT in record.getMessage() for record in caplog.records) + + +def test_no_callback_at_all_is_a_timeout() -> None: + """Nothing arrived on the loopback port before the deadline.""" + with pytest.raises(git.TokenError, match="no response on http://localhost"): + _run({}, _response()) + + +def test_refused_consent_is_reported_as_refusal() -> None: + """Google reports a declined consent as ``error``.""" + with pytest.raises(git.TokenError, match="refused: access_denied"): + _run({"error": "access_denied"}, _response()) + + +def test_mismatched_state_is_rejected() -> None: + """A callback that does not echo the state is not ours.""" + with pytest.raises(git.TokenError, match="state did not match"): + _run({"code": "c", "state": "some-other-state"}, _response()) + + +def test_callback_without_a_code_is_rejected() -> None: + """State matched but no code: nothing to exchange.""" + with pytest.raises(git.TokenError, match="carried no authorization code"): + _run({"state": _STATE, "scope": "openid"}, _response()) + + +def test_network_failure_during_exchange_is_wrapped() -> None: + """A transport error must not surface as a bare requests exception.""" + boom = requests.ConnectionError("connection refused") + + with pytest.raises(git.TokenError, match="network error exchanging the code"): + _run({"code": "c", "state": _STATE}, boom) + + +def test_http_error_from_the_exchange_is_reported_with_its_status() -> None: + """A non-2xx exchange carries the status, which is what identifies it.""" + with pytest.raises(git.TokenError, match="HTTP 400"): + _run({"code": "c", "state": _STATE}, _response(ok=False, status=400)) + + +def test_a_token_response_without_an_id_token_names_the_likely_cause() -> None: + """Dropping ``openid`` from the scopes yields tokens but no id_token.""" + with pytest.raises(git.TokenError, match="was 'openid' in the scopes"): + _run({"code": "c", "state": _STATE}, _response(payload={"access_token": "a"})) + + +def test_the_server_is_closed_even_though_the_flow_failed() -> None: + """A leaked listener would make the next run fail to bind the same port.""" + starter, free_port, handler, server, thread = _fake_flow({}) + with starter, free_port, handler, pytest.raises(git.TokenError): + git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL) + + thread.join.assert_called_once() + server.server_close.assert_called_once() diff --git a/crdt-sync/crdt_sync/tests/test_tool_google_id_token_cli.py b/crdt-sync/crdt_sync/tests/test_tool_google_id_token_cli.py new file mode 100644 index 0000000..a41290c --- /dev/null +++ b/crdt-sync/crdt_sync/tests/test_tool_google_id_token_cli.py @@ -0,0 +1,93 @@ +"""Tests for ``tool/google_id_token``'s command line entry point. + +Split from :mod:`test_tool_google_id_token` to stay under the repo's 250-line +file cap. That module covers the OAuth flow itself; this one covers ``main()`` +-- argument parsing, where the token is written, and the exit code -- with +``fetch_id_token`` patched out, since the flow is already exercised there. + +The shared fixture values are imported rather than redeclared so the two +files cannot drift. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from crdt_sync.tests.test_tool_google_id_token import ( + _CLIENT_CREDENTIAL, + _CLIENT_ID, + _ID_VALUE, +) +from tool import google_id_token as git + +if TYPE_CHECKING: + from pathlib import Path + + +def test_main_prints_the_token_when_no_output_is_given( + caplog: pytest.LogCaptureFixture, +) -> None: + """Default: the token goes to the log for copy-paste.""" + with ( + patch.object(git, "fetch_id_token", return_value=_ID_VALUE), + caplog.at_level(logging.INFO, logger="google_id_token"), + ): + code = git.main( + ["--client-id", _CLIENT_ID, "--client-secret", _CLIENT_CREDENTIAL] + ) + + assert code == 0 + assert any(record.getMessage() == _ID_VALUE for record in caplog.records) + + +def test_main_writes_the_token_to_a_file_at_mode_0600(tmp_path: Path) -> None: + """An ID token is a bearer credential, so the file must not be readable.""" + destination = tmp_path / "token.txt" + + with patch.object(git, "fetch_id_token", return_value=_ID_VALUE): + code = git.main( + [ + "--client-id", + _CLIENT_ID, + "--client-secret", + _CLIENT_CREDENTIAL, + "--output", + str(destination), + ] + ) + + assert code == 0 + assert destination.read_text(encoding="utf-8") == _ID_VALUE + assert destination.stat().st_mode & 0o777 == 0o600 + + +def test_main_passes_no_browser_and_port_through() -> None: + """The two flags that decide whether the flow can work at all.""" + with patch.object(git, "fetch_id_token", return_value=_ID_VALUE) as fetch: + git.main( + [ + "--client-id", + _CLIENT_ID, + "--client-secret", + _CLIENT_CREDENTIAL, + "--no-browser", + "--port", + "8765", + ] + ) + + assert fetch.call_args.kwargs == {"open_browser": False, "port": 8765} + + +def test_main_returns_1_when_the_flow_fails() -> None: + """A TokenError is an exit code, not a traceback.""" + with patch.object(git, "fetch_id_token", side_effect=git.TokenError("nope")): + code = git.main( + ["--client-id", _CLIENT_ID, "--client-secret", _CLIENT_CREDENTIAL] + ) + + assert code == 1 diff --git a/crdt-sync/crdt_sync/tests/test_tool_oauth_callback.py b/crdt-sync/crdt_sync/tests/test_tool_oauth_callback.py index 1f59b65..47d3070 100644 --- a/crdt-sync/crdt_sync/tests/test_tool_oauth_callback.py +++ b/crdt-sync/crdt_sync/tests/test_tool_oauth_callback.py @@ -164,11 +164,21 @@ def test_serve_loop_gives_up_once_the_consent_deadline_passes( _retire(srv, thread) -def test_log_message_is_silenced() -> None: - """The default handler logs every request to stderr; this one must not.""" - # Called unbound with no request context: the override must accept any - # arguments and do nothing, which is exactly what makes that safe. - assert oc._CallbackHandler.log_message(None, '"GET / HTTP/1.1"', 200) is None +def test_log_message_writes_nothing_to_stderr( + server: tuple[int, threading.Thread], + capfd: pytest.CaptureFixture[str], +) -> None: + """The default handler logs every request to stderr; this one must not. + + Asserted through a real request rather than by calling the override + directly: it returns None either way, so only the absence of output + distinguishes a silenced handler from the stock one. + """ + port, _ = server + + _get(port, f"/?code=the-code&state={_STATE}") + + assert capfd.readouterr().err == "" def test_starting_the_server_resets_state_from_a_previous_run() -> None: