kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 20:27:15 +01:00
Code Issues Actions 16 Packages Projects Releases Wiki Activity

crdt-sync: pylint --fail-under=10, scoped to source, on pylint 4.0.8

Browse Source 
The 8.0 bar tolerated two points of debt and let the score regress
unnoticed. Adopting the fleet's source-only pylint scope needed
ignore-paths, not ignore = ["tests"]: ignore matches only the basename
of what pylint is handed, and pre-commit hands it explicit file paths,
so a directory name there never fired under the hook.

What the scope change left in source: runtime imports moved above the
TYPE_CHECKING blocks, the consent-flow CLI plumbing shared between
google_id_token and seed_session instead of duplicated, the token
exchange split out of google_id_token to stay under the 250-line cap,
and http.server's do_<METHOD> names admitted by method-rgx since the
stdlib dispatches on that exact spelling.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HHCbkJWXHjtTUaqvnHRAqA
This commit is contained in:
Krzysztof Rudnicki 2026-09-12 19:06:30 +02:00
parent ee276cd8e5
commit c295db7a15
9 changed files with 162 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
crdt-sync/.pre-commit-config.yaml
View File

@ -103,12 +103,12 @@ repos:
- --disable-error-code=list-item
- repo: https://github.com/pylint-dev/pylint
rev: v3.3.2
rev: v4.0.8
hooks:
- id: pylint
args:
- --rcfile=crdt-sync/pyproject.toml
- --fail-under=8.0
- --fail-under=10
- --jobs=4
additional_dependencies:
- pytest

4
crdt-sync/crdt_sync/_pull.py
View File

@ -13,6 +13,8 @@ from dataclasses import dataclass
import logging
from typing import TYPE_CHECKING
from crdt_sync._revisions import revision_of
if TYPE_CHECKING:
from collections.abc import Callable
@ -20,8 +22,6 @@ if TYPE_CHECKING:
from crdt_sync._remote import RemoteStore
from crdt_sync._syncstate import SyncState
from crdt_sync._revisions import revision_of
_logger = logging.getLogger(__name__)

7
crdt-sync/crdt_sync/_sync.py
View File

@ -17,10 +17,6 @@ import logging
from typing import TYPE_CHECKING
from crdt_sync._log import merge_logs
if TYPE_CHECKING:
from crdt_sync._log import Log
from crdt_sync._pull import _pull_remote_logs, _PullContext, _remote_revs
from crdt_sync._revisions import revision_of
from crdt_sync._syncargs import LogCodec, RevisionTracking, SyncTarget
@ -31,6 +27,9 @@ from crdt_sync._syncstate import (
SyncStateStore,
)
if TYPE_CHECKING:
from crdt_sync._log import Log
# Named explicitly so the autofixer cannot prune an import that exists for its
# re-export: crdt_sync/__init__.py imports all of these from here.
__all__ = [

9
crdt-sync/crdt_sync/tests/test_tool_google_id_token.py
View File

@ -17,6 +17,7 @@ from unittest.mock import MagicMock, patch
import pytest
import requests
from tool import _token_exchange as tx
from tool import google_id_token as git
_CLIENT_ID = "client-id-for-the-web-application"
@ -68,7 +69,7 @@ def _run(result: dict[str, str], post: MagicMock | Exception, **kwargs: object)
"""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")
target = patch.object(tx.requests, "post")
with target as posted:
if isinstance(post, Exception):
posted.side_effect = post
@ -91,7 +92,7 @@ def test_the_exchange_posts_the_code_and_the_matching_redirect_uri() -> None:
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:
with starter, free_port, handler, patch.object(tx.requests, "post") as posted:
posted.return_value = _response(payload={"id_token": _ID_VALUE})
git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=8765)
@ -110,7 +111,7 @@ def test_a_fixed_port_is_used_verbatim_rather_than_a_free_one() -> None:
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:
with starter, free_port as picked, handler, patch.object(tx.requests, "post") as p:
p.return_value = _response(payload={"id_token": _ID_VALUE})
git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=8765)
@ -120,7 +121,7 @@ def test_a_fixed_port_is_used_verbatim_rather_than_a_free_one() -> None:
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:
with starter, free_port as picked, handler, patch.object(tx.requests, "post") as p:
p.return_value = _response(payload={"id_token": _ID_VALUE})
git.fetch_id_token(_CLIENT_ID, _CLIENT_CREDENTIAL, port=0)

23
crdt-sync/crdt_sync/tests/test_tool_seed_session_main.py
View File

@ -20,6 +20,7 @@ import pytest
from crdt_sync._config import ConfigError
from crdt_sync._firebase_auth import FirebaseAuthError
from crdt_sync.tests.test_tool_seed_session import _CONFIG, _ID_VALUE
from tool import google_id_token as git
from tool import seed_session as ss
from tool.google_id_token import TokenError
@ -32,7 +33,7 @@ def test_main_seeds_every_default_app(caplog: pytest.LogCaptureFixture) -> None:
"""With no --app, all of DEFAULT_APPS are seeded."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", return_value=list(ss.DEFAULT_APPS)) as seeded,
caplog.at_level(logging.INFO, logger="seed_session"),
):
@ -62,7 +63,7 @@ def test_a_single_app_can_be_selected() -> None:
"""--app narrows the run, which is what a rerun after a failure uses."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", return_value=["diet_guard"]) as seeded,
):
code = ss.main([*_ARGV, "--app", "diet_guard"])
@ -75,7 +76,7 @@ def test_the_app_flag_is_repeatable() -> None:
"""Rerunning two failed apps should not need two invocations."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", return_value=["diet_guard", "todo"]) as seeded,
):
ss.main([*_ARGV, "--app", "diet_guard", "--app", "todo"])
@ -98,7 +99,7 @@ def test_the_consent_flow_receives_the_client_and_port() -> None:
"""A Web client needs the registered port verbatim, not a free one."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE) as fetch,
patch.object(git, "fetch_id_token", return_value=_ID_VALUE) as fetch,
patch.object(ss, "seed_apps", return_value=[]),
):
ss.main([*_ARGV, "--port", "9999", "--no-browser"])
@ -111,7 +112,7 @@ def test_the_default_port_is_the_registered_one() -> None:
"""8765 is what is registered on the Web client as http://localhost:8765."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE) as fetch,
patch.object(git, "fetch_id_token", return_value=_ID_VALUE) as fetch,
patch.object(ss, "seed_apps", return_value=[]),
):
ss.main(_ARGV)
@ -125,7 +126,7 @@ def test_the_expected_uid_is_announced_before_the_flow(
"""So the wrong Google account can be spotted at the picker, not after."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", return_value=[]),
caplog.at_level(logging.INFO, logger="seed_session"),
):
@ -150,7 +151,7 @@ def test_the_named_failures_return_1(
"""Each is reported, not raised as a traceback."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", side_effect=failure),
patch.object(git, "fetch_id_token", side_effect=failure),
caplog.at_level(logging.INFO, logger="seed_session"),
):
code = ss.main(_ARGV)
@ -163,7 +164,7 @@ def test_an_unexpected_error_is_not_swallowed() -> None:
"""The reason the handler names its four types instead of Exception."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", side_effect=RuntimeError("a real bug")),
patch.object(git, "fetch_id_token", side_effect=RuntimeError("a real bug")),
pytest.raises(RuntimeError, match="a real bug"),
):
ss.main(_ARGV)
@ -175,7 +176,7 @@ def test_a_total_failure_suggests_rerunning_every_app(
"""Nothing verified, so nothing may be reported as already working."""
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", side_effect=ss.SeedError("all dead")),
caplog.at_level(logging.INFO, logger="seed_session"),
):
@ -194,7 +195,7 @@ def test_a_partial_failure_reports_what_already_works(
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", side_effect=partial),
caplog.at_level(logging.INFO, logger="seed_session"),
):
@ -212,7 +213,7 @@ def test_the_rerun_line_excludes_the_apps_that_already_work(
with (
patch.object(ss.FirebaseConfig, "load", return_value=_CONFIG),
patch.object(ss, "fetch_id_token", return_value=_ID_VALUE),
patch.object(git, "fetch_id_token", return_value=_ID_VALUE),
patch.object(ss, "seed_apps", side_effect=partial),
caplog.at_level(logging.INFO, logger="seed_session"),
):

17
crdt-sync/pyproject.toml
View File

@ -83,13 +83,30 @@ color_output = true
exclude = [".venv/"]
[tool.pylint.main]
# The repo root on the path, so pylint resolves the package's own imports the
# same way pytest does instead of reporting import-error for them.
init-hook = "import sys; sys.path.insert(0, 'crdt-sync')"
analyse-fallback-blocks = true
persistent = true
jobs = 0
py-version = "3.14"
ignore = [".venv", "__pycache__"]
# Test suites intentionally use patterns (protected-access, fixture-arg
# shadowing) that don't apply to source code; ruff's per-file-ignores carry
# the equivalent exemptions, so tests stay linted -- by ruff. This is the same
# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has
# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the
# basename of what pylint is *given*, and pre-commit hands it explicit file
# paths, so a directory name in `ignore` never fires under the hook.
ignore-paths = ["^(.*/)?tests/.*$"]
ignore-patterns = [".*\\.pyi$"]
[tool.pylint.basic]
# snake_case, plus the ``do_<METHOD>`` names http.server dispatches on: the
# stdlib looks handlers up as ``"do_" + command``, so ``do_GET`` is the only
# spelling that works and cannot be renamed to satisfy the style.
method-rgx = "([^\\W\\dA-Z][^\\WA-Z]{2,}|_[^\\WA-Z]*|__[^\\W\\dA-Z][^\\WA-Z]+__|do_[A-Z]+)$"
[tool.pylint.messages_control]
enable = "all"
disable = []

74
crdt-sync/tool/_token_exchange.py Normal file
View File

@ -0,0 +1,74 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Turning the captured OAuth callback into a Google ID token.
Split from :mod:`google_id_token`, which keeps the consent flow and the CLI:
this half validates what the loopback server caught and trades the
authorization code for tokens at Google's endpoint.
"""
from __future__ import annotations
import requests
from tool._oauth_callback import _CONSENT_TIMEOUT_SECONDS
# Assembled from host and path rather than written as one literal: a string
# ending in "/token" trips ruff's hardcoded-password check (S105), and the
# honest fix is to not have the literal rather than to silence the rule.
_OAUTH_HOST = "https://oauth2.googleapis.com"
_TOKEN_EXCHANGE_URL = f"{_OAUTH_HOST}/token"
_TIMEOUT_SECONDS = 30
class TokenError(Exception):
"""The authorization flow did not produce an ID token."""
def _authorization_code(params: dict[str, str], redirect_uri: str, state: str) -> str:
"""Validate the captured callback and return its authorization code."""
if not params:
msg = f"no response on {redirect_uri} within {_CONSENT_TIMEOUT_SECONDS}s"
raise TokenError(msg)
if "error" in params:
msg = f"authorization was refused: {params['error']}"
raise TokenError(msg)
if params.get("state") != state:
msg = "the callback state did not match; ignoring the response"
raise TokenError(msg)
code = params.get("code")
if not code:
msg = f"the callback carried no authorization code: {params}"
raise TokenError(msg)
return code
def _exchange_code(
code: str, client_id: str, client_secret: str, redirect_uri: str
) -> str:
"""Trade the authorization code for an ID token at Google's endpoint."""
try:
response = requests.post(
_TOKEN_EXCHANGE_URL,
data={
"code": code,
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
"grant_type": "authorization_code",
},
timeout=_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
msg = f"network error exchanging the code: {exc}"
raise TokenError(msg) from exc
if not response.ok:
msg = f"token exchange failed: HTTP {response.status_code} {response.text}"
raise TokenError(msg)
id_token = response.json().get("id_token")
if not id_token:
msg = "the token response carried no id_token; was 'openid' in the scopes?"
raise TokenError(msg)
return str(id_token)

108
crdt-sync/tool/google_id_token.py
View File

@ -43,36 +43,23 @@ import sys
import urllib.parse
import webbrowser
import requests
from tool._oauth_callback import (
_CONSENT_TIMEOUT_SECONDS,
_CallbackHandler,
_free_port,
_start_callback_server,
)
from tool._token_exchange import TokenError, _authorization_code, _exchange_code
_AUTH_ENDPOINT = "https://accounts.google.com/o/oauth2/v2/auth"
# Assembled from host and path rather than written as one literal: a string
# ending in "/token" trips ruff's hardcoded-password check (S105), and the
# honest fix is to not have the literal rather than to silence the rule.
_OAUTH_HOST = "https://oauth2.googleapis.com"
_TOKEN_EXCHANGE_URL = f"{_OAUTH_HOST}/token"
# openid gets an ID token at all; email makes the token carry the address, so
# link_google.py can show which account is about to be linked.
_SCOPES = "openid email profile"
_TIMEOUT_SECONDS = 30
_logger = logging.getLogger("google_id_token")
class TokenError(Exception):
"""The authorization flow did not produce an ID token."""
def fetch_id_token(
client_id: str,
client_secret: str,
@ -129,87 +116,64 @@ def fetch_id_token(
thread.join(timeout=_CONSENT_TIMEOUT_SECONDS)
server.server_close()
params = _CallbackHandler.result
if not params:
msg = f"no response on {redirect_uri} within {_CONSENT_TIMEOUT_SECONDS}s"
raise TokenError(msg)
if "error" in params:
msg = f"authorization was refused: {params['error']}"
raise TokenError(msg)
if params.get("state") != state:
msg = "the callback state did not match; ignoring the response"
raise TokenError(msg)
code = params.get("code")
if not code:
msg = f"the callback carried no authorization code: {params}"
raise TokenError(msg)
try:
response = requests.post(
_TOKEN_EXCHANGE_URL,
data={
"code": code,
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
"grant_type": "authorization_code",
},
timeout=_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
msg = f"network error exchanging the code: {exc}"
raise TokenError(msg) from exc
if not response.ok:
msg = f"token exchange failed: HTTP {response.status_code} {response.text}"
raise TokenError(msg)
id_token = response.json().get("id_token")
if not id_token:
msg = "the token response carried no id_token; was 'openid' in the scopes?"
raise TokenError(msg)
return str(id_token)
code = _authorization_code(_CallbackHandler.result, redirect_uri, state)
return _exchange_code(code, client_id, client_secret, redirect_uri)
def main(argv: list[str] | None = None) -> int:
"""Fetch an ID token and print or save it. Returns an exit code."""
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
def add_consent_args(
parser: argparse.ArgumentParser, *, port_default: int, port_help: str
) -> None:
"""Register the client-credential and consent options every flow CLI takes.
Shared with :mod:`tool.seed_session`, which runs the same consent flow
before seeding; only the port default and its explanation differ.
"""
parser.add_argument("--client-id", required=True, help="Web OAuth client id.")
parser.add_argument(
"--client-secret",
required=True,
help="Web OAuth client secret.",
)
parser.add_argument(
"--output",
help="Write the token here (mode 0600) instead of printing it.",
)
parser.add_argument(
"--no-browser",
action="store_true",
help="Print the consent URL without launching a browser.",
)
parser.add_argument(
"--port",
type=int,
default=0,
help=(
parser.add_argument("--port", type=int, default=port_default, help=port_help)
def fetch_id_token_for(args: argparse.Namespace) -> str:
"""Run :func:`fetch_id_token` from options added by :func:`add_consent_args`."""
return fetch_id_token(
args.client_id,
args.client_secret,
open_browser=not args.no_browser,
port=args.port,
)
def main(argv: list[str] | None = None) -> int:
"""Fetch an ID token and print or save it. Returns an exit code."""
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
add_consent_args(
parser,
port_default=0,
port_help=(
"Fixed loopback port for the redirect URI. Required for a Web "
"application client, which must have http://localhost:<port> "
"registered verbatim; 0 (the default) picks a free port and "
"suits a Desktop app client."
),
)
parser.add_argument(
"--output",
help="Write the token here (mode 0600) instead of printing it.",
)
args = parser.parse_args(argv)
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
try:
token = fetch_id_token(
args.client_id,
args.client_secret,
open_browser=not args.no_browser,
port=args.port,
)
token = fetch_id_token_for(args)
except TokenError:
_logger.exception(" FAIL")
return 1

39
crdt-sync/tool/seed_session.py
View File

@ -44,7 +44,7 @@ from crdt_sync._remote import RemoteSyncError
# Importable as a package module (tool/__init__.py exists) so the loopback
# OAuth flow is shared, not reimplemented.
from tool._seeded_apps import DEFAULT_APPS
from tool.google_id_token import TokenError, fetch_id_token
from tool.google_id_token import TokenError, add_consent_args, fetch_id_token_for
# A fixed default so the redirect URI is stable enough to register once on the
# Web client. Arbitrary but out of the ephemeral range, and free on this box.
@ -142,11 +142,15 @@ def seed_apps(
def main(argv: list[str] | None = None) -> int:
"""Run the consent flow and seed every app. Returns an exit code."""
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--client-id", required=True, help="Web OAuth client id.")
parser.add_argument(
"--client-secret",
required=True,
help="Web OAuth client secret.",
add_consent_args(
parser,
port_default=_DEFAULT_REDIRECT_PORT,
port_help=(
"Loopback port for the OAuth redirect. Must be registered on the "
"Web client as http://localhost:<port>, exactly -- a Web client "
"does not accept arbitrary loopback ports the way a Desktop "
"client does."
),
)
parser.add_argument(
"--app",
@ -162,22 +166,6 @@ def main(argv: list[str] | None = None) -> int:
+ ", ".join(DEFAULT_APPS)
),
)
parser.add_argument(
"--no-browser",
action="store_true",
help="Print the consent URL without launching a browser.",
)
parser.add_argument(
"--port",
type=int,
default=_DEFAULT_REDIRECT_PORT,
help=(
"Loopback port for the OAuth redirect. Must be registered on the "
"Web client as http://localhost:<port>, exactly -- a Web client "
"does not accept arbitrary loopback ports the way a Desktop "
"client does."
),
)
args = parser.parse_args(argv)
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
@ -190,12 +178,7 @@ def main(argv: list[str] | None = None) -> int:
config = FirebaseConfig.load()
_logger.info(" .. seeding %s", ", ".join(app_names))
_logger.info(" .. expecting uid %s (%s)", config.uid, config.email)
token = fetch_id_token(
args.client_id,
args.client_secret,
open_browser=not args.no_browser,
port=args.port,
)
token = fetch_id_token_for(args)
seeded = seed_apps(config, token, app_names)
except (ConfigError, FirebaseAuthError, TokenError, SeedError) as exc:
# logging.exception already renders the exception and its traceback.