* [PATCH v2 01/14] Add ci script and mypy suppressions
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 02/14] Type make_msg and drop test suppressions Tamir Duberstein
` (13 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Add a simple shell script that runs mypy and pytest.
Introduce targeted mypy suppressions for current test-only issues so the
baseline is green before tightening types further.
Also add pytest-asyncio so pytest recognizes
asyncio_default_fixture_loop_scope and stops warning about it.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
ci.sh | 6 ++++++
pyproject.toml | 14 ++++++++++++++
2 files changed, 20 insertions(+)
diff --git a/ci.sh b/ci.sh
new file mode 100755
index 0000000..1e8ccb3
--- /dev/null
+++ b/ci.sh
@@ -0,0 +1,6 @@
+#!/usr/bin/env sh
+
+set -eu
+
+uv run mypy .
+uv run pytest --durations=0
diff --git a/pyproject.toml b/pyproject.toml
index 2353a18..e31a05c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -40,6 +40,7 @@ dev = [
"build",
"mypy",
"pytest",
+ "pytest-asyncio",
"ruff",
"types-requests",
]
@@ -56,5 +57,18 @@ strict = true
module = "authheaders"
ignore_missing_imports = true
+[[tool.mypy.overrides]]
+module = [
+ "test_email_utils",
+ "test_formatting",
+ "test_message",
+ "test_thread",
+]
+disable_error_code = ["attr-defined"]
+
+[[tool.mypy.overrides]]
+module = "test_auth_headers"
+disable_error_code = ["unused-ignore"]
+
[tool.ruff]
target-version = "py39"
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 02/14] Type make_msg and drop test suppressions
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 01/14] Add ci script and mypy suppressions Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 03/14] Add ruff import checks to b4 CI Tamir Duberstein
` (12 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Replace the make_msg class helper with a typed callable fixture and
update tests to call it directly. This keeps argument checking in the
tests while simplifying the fixture implementation.
Remove the temporary mypy suppressions now that the underlying typing
issues are fixed in the test code.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
pyproject.toml | 13 --------
tests/conftest.py | 79 ++++++++++++++++++++++++++--------------------
| 4 +--
tests/test_email_utils.py | 60 ++++++++++++++++++-----------------
tests/test_formatting.py | 74 ++++++++++++++++++++++---------------------
tests/test_message.py | 10 +++---
tests/test_thread.py | 34 ++++++++++----------
7 files changed, 140 insertions(+), 134 deletions(-)
diff --git a/pyproject.toml b/pyproject.toml
index e31a05c..7c9e1da 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -57,18 +57,5 @@ strict = true
module = "authheaders"
ignore_missing_imports = true
-[[tool.mypy.overrides]]
-module = [
- "test_email_utils",
- "test_formatting",
- "test_message",
- "test_thread",
-]
-disable_error_code = ["attr-defined"]
-
-[[tool.mypy.overrides]]
-module = "test_auth_headers"
-disable_error_code = ["unused-ignore"]
-
[tool.ruff]
target-version = "py39"
diff --git a/tests/conftest.py b/tests/conftest.py
index cf1a9c7..b84fc35 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -4,6 +4,7 @@ from __future__ import annotations
import email.utils
import textwrap
+from typing import Protocol
import pytest
@@ -12,41 +13,51 @@ from email.message import EmailMessage
from liblore import emlpolicy
+class MsgFactory(Protocol):
+ def __call__(
+ self,
+ subject: str = ...,
+ from_addr: tuple[str, str] = ...,
+ msgid: str | None = ...,
+ in_reply_to: str | None = ...,
+ references: list[str] | None = ...,
+ body: str = ...,
+ date: str | None = ...,
+ ) -> EmailMessage: ...
+
+
@pytest.fixture()
-def make_msg() -> type:
- """Factory fixture that returns a helper class for building test messages."""
-
- class MsgFactory:
- _counter = 0
-
- @classmethod
- def create(
- cls,
- subject: str = 'Test message',
- from_addr: tuple[str, str] = ('Test User', 'test@example.com'),
- msgid: str | None = None,
- in_reply_to: str | None = None,
- references: list[str] | None = None,
- body: str = 'Hello, world!\n',
- date: str | None = None,
- ) -> EmailMessage:
- cls._counter += 1
- if msgid is None:
- msgid = f'msg{cls._counter}@example.com'
- msg = EmailMessage(policy=emlpolicy)
- msg['Subject'] = subject
- msg['From'] = email.utils.formataddr(from_addr)
- msg['Message-Id'] = f'<{msgid}>'
- if in_reply_to:
- msg['In-Reply-To'] = f'<{in_reply_to}>'
- if references:
- msg['References'] = ' '.join(f'<{r}>' for r in references)
- if date:
- msg['Date'] = date
- msg.set_content(body)
- return msg
-
- return MsgFactory
+def make_msg() -> MsgFactory:
+ """Factory fixture that returns a helper callable for test messages."""
+ counter = 0
+
+ def create(
+ subject: str = 'Test message',
+ from_addr: tuple[str, str] = ('Test User', 'test@example.com'),
+ msgid: str | None = None,
+ in_reply_to: str | None = None,
+ references: list[str] | None = None,
+ body: str = 'Hello, world!\n',
+ date: str | None = None,
+ ) -> EmailMessage:
+ nonlocal counter
+ counter += 1
+ if msgid is None:
+ msgid = f'msg{counter}@example.com'
+ msg = EmailMessage(policy=emlpolicy)
+ msg['Subject'] = subject
+ msg['From'] = email.utils.formataddr(from_addr)
+ msg['Message-Id'] = f'<{msgid}>'
+ if in_reply_to:
+ msg['In-Reply-To'] = f'<{in_reply_to}>'
+ if references:
+ msg['References'] = ' '.join(f'<{r}>' for r in references)
+ if date:
+ msg['Date'] = date
+ msg.set_content(body)
+ return msg
+
+ return create
@pytest.fixture()
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index 6c524b5..9edf4cd 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -67,8 +67,8 @@ class TestAuthenticateMsgs:
assert msg['Authentication-Results'] == (
'liblore; dkim=pass header.d=example.com'
)
- fake.authenticate_message.assert_called_once() # type: ignore[attr-defined]
- call_kwargs = fake.authenticate_message.call_args # type: ignore[attr-defined]
+ fake.authenticate_message.assert_called_once()
+ call_kwargs = fake.authenticate_message.call_args
assert call_kwargs[0][1] == 'liblore'
assert call_kwargs[1]['dkim'] is True
assert call_kwargs[1]['dmarc'] is True
diff --git a/tests/test_email_utils.py b/tests/test_email_utils.py
index 97c4ff3..ad8775f 100644
--- a/tests/test_email_utils.py
+++ b/tests/test_email_utils.py
@@ -4,6 +4,8 @@ from __future__ import annotations
from email.message import EmailMessage
+from conftest import MsgFactory
+
from liblore import emlpolicy
from liblore.utils import (
msg_get_author,
@@ -16,43 +18,43 @@ from liblore.utils import (
class TestMsgGetSubject:
- def test_plain_subject(self, make_msg: type) -> None:
- msg = make_msg.create(subject='Just a plain subject')
+ def test_plain_subject(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='Just a plain subject')
assert msg_get_subject(msg) == 'Just a plain subject'
- def test_no_strip(self, make_msg: type) -> None:
- msg = make_msg.create(subject='[PATCH v3 2/5] subsys: fix thing')
+ def test_no_strip(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='[PATCH v3 2/5] subsys: fix thing')
assert msg_get_subject(msg) == '[PATCH v3 2/5] subsys: fix thing'
- def test_strip_patch_prefix(self, make_msg: type) -> None:
- msg = make_msg.create(subject='[PATCH v3 2/5] subsys: fix thing')
+ def test_strip_patch_prefix(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='[PATCH v3 2/5] subsys: fix thing')
assert msg_get_subject(msg, strip_prefixes=True) == 'subsys: fix thing'
- def test_strip_re_and_prefix(self, make_msg: type) -> None:
- msg = make_msg.create(subject='Re: [PATCH] Something cool')
+ def test_strip_re_and_prefix(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='Re: [PATCH] Something cool')
assert msg_get_subject(msg, strip_prefixes=True) == 'Something cool'
- def test_strip_multiple_prefixes(self, make_msg: type) -> None:
- msg = make_msg.create(subject='[RFC PATCH v2 0/3] [net-next] New feature')
+ def test_strip_multiple_prefixes(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='[RFC PATCH v2 0/3] [net-next] New feature')
result = msg_get_subject(msg, strip_prefixes=True)
assert result == 'New feature'
- def test_strip_aw_prefix(self, make_msg: type) -> None:
- msg = make_msg.create(subject='Aw: [PATCH] German reply')
+ def test_strip_aw_prefix(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='Aw: [PATCH] German reply')
assert msg_get_subject(msg, strip_prefixes=True) == 'German reply'
def test_no_subject(self) -> None:
msg = EmailMessage(policy=emlpolicy)
assert msg_get_subject(msg) == ''
- def test_strip_no_brackets(self, make_msg: type) -> None:
- msg = make_msg.create(subject='No brackets here')
+ def test_strip_no_brackets(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(subject='No brackets here')
assert msg_get_subject(msg, strip_prefixes=True) == 'No brackets here'
class TestMsgGetAuthor:
- def test_normal_from(self, make_msg: type) -> None:
- msg = make_msg.create(from_addr=('Jane Doe', 'jane@example.com'))
+ def test_normal_from(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(from_addr=('Jane Doe', 'jane@example.com'))
name, addr = msg_get_author(msg)
assert name == 'Jane Doe'
assert addr == 'jane@example.com'
@@ -72,23 +74,23 @@ class TestMsgGetAuthor:
class TestMsgGetPayload:
- def test_plain_body(self, make_msg: type) -> None:
- msg = make_msg.create(body='This is the body.\n')
+ def test_plain_body(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(body='This is the body.\n')
assert 'This is the body.' in msg_get_payload(msg)
- def test_strip_signature(self, make_msg: type) -> None:
- msg = make_msg.create(body='Body text.\n-- \nMy Sig\n')
+ def test_strip_signature(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(body='Body text.\n-- \nMy Sig\n')
result = msg_get_payload(msg, strip_signature=True)
assert 'Body text.' in result
assert 'My Sig' not in result
- def test_keep_signature(self, make_msg: type) -> None:
- msg = make_msg.create(body='Body text.\n-- \nMy Sig\n')
+ def test_keep_signature(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(body='Body text.\n-- \nMy Sig\n')
result = msg_get_payload(msg, strip_signature=False)
assert 'My Sig' in result
- def test_strip_quoted(self, make_msg: type) -> None:
- msg = make_msg.create(body='My reply.\n> Quoted line.\nMore text.\n')
+ def test_strip_quoted(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(body='My reply.\n> Quoted line.\nMore text.\n')
result = msg_get_payload(msg, strip_quoted=True)
assert 'My reply.' in result
assert 'Quoted line.' not in result
@@ -118,12 +120,12 @@ class TestMsgGetRecipients:
class TestSortMsgsByReceived:
- def test_sorts_by_date(self, make_msg: type) -> None:
- msg1 = make_msg.create(
+ def test_sorts_by_date(self, make_msg: MsgFactory) -> None:
+ msg1 = make_msg(
msgid='older@x.com',
date='Mon, 01 Jan 2024 00:00:00 +0000',
)
- msg2 = make_msg.create(
+ msg2 = make_msg(
msgid='newer@x.com',
date='Tue, 02 Jan 2024 00:00:00 +0000',
)
@@ -132,8 +134,8 @@ class TestSortMsgsByReceived:
assert result[0]['Message-Id'] == '<older@x.com>'
assert result[1]['Message-Id'] == '<newer@x.com>'
- def test_skips_dateless(self, make_msg: type) -> None:
- msg = make_msg.create()
+ def test_skips_dateless(self, make_msg: MsgFactory) -> None:
+ msg = make_msg()
# No Date header set at all — del it if make_msg added one
if 'Date' in msg:
del msg['Date']
diff --git a/tests/test_formatting.py b/tests/test_formatting.py
index 45afa64..a13593e 100644
--- a/tests/test_formatting.py
+++ b/tests/test_formatting.py
@@ -6,6 +6,8 @@ import pytest
from email.message import EmailMessage
+from conftest import MsgFactory
+
from liblore.utils import (
clean_header,
format_addrs,
@@ -165,9 +167,9 @@ class TestGetMsgAsBytes:
class TestMinimizeThread:
- def test_headers_filtered(self, make_msg: type) -> None:
+ def test_headers_filtered(self, make_msg: MsgFactory) -> None:
"""Only headers in MINIMIZE_KEEP_HEADERS are kept."""
- msg = make_msg.create(subject='Test', body='Hello.\n')
+ msg = make_msg(subject='Test', body='Hello.\n')
msg['X-Custom'] = 'should be dropped'
msg['List-Id'] = '<test.example.com>'
result = minimize_thread([msg])
@@ -176,9 +178,9 @@ class TestMinimizeThread:
assert result[0]['X-Custom'] is None
assert result[0]['List-Id'] is None
- def test_keep_headers_default(self, make_msg: type) -> None:
+ def test_keep_headers_default(self, make_msg: MsgFactory) -> None:
"""All default MINIMIZE_KEEP_HEADERS are preserved when present."""
- msg = make_msg.create(
+ msg = make_msg(
subject='Test',
from_addr=('Alice', 'alice@example.com'),
body='Hello.\n',
@@ -200,9 +202,9 @@ class TestMinimizeThread:
assert mmsg['Reply-To'] is not None
assert mmsg['In-Reply-To'] is not None
- def test_custom_keep_headers(self, make_msg: type) -> None:
+ def test_custom_keep_headers(self, make_msg: MsgFactory) -> None:
"""Callers can override which headers to keep."""
- msg = make_msg.create(subject='Test', body='Hello.\n',
+ msg = make_msg(subject='Test', body='Hello.\n',
date='Mon, 01 Jan 2024 00:00:00 +0000')
result = minimize_thread([msg], keep_headers=('Subject',))
assert len(result) == 1
@@ -210,10 +212,10 @@ class TestMinimizeThread:
assert result[0]['From'] is None
assert result[0]['Date'] is None
- def test_multi_level_quotes_stripped(self, make_msg: type) -> None:
+ def test_multi_level_quotes_stripped(self, make_msg: MsgFactory) -> None:
"""Lines with >> (multi-level quoting) are removed."""
body = 'My reply.\n> Single quote.\n>> Double quote.\nMore text.\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
assert len(result) == 1
text = result[0].get_payload()
@@ -222,26 +224,26 @@ class TestMinimizeThread:
assert 'Double quote.' not in text
assert 'More text.' in text
- def test_empty_quote_lines_stripped(self, make_msg: type) -> None:
+ def test_empty_quote_lines_stripped(self, make_msg: MsgFactory) -> None:
"""Bare '>' lines are cleaned up."""
body = 'Reply.\n>\n> Real quote.\nMore text.\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'Reply.' in text
assert 'Real quote.' in text
assert 'More text.' in text
- def test_bottom_quotes_dropped(self, make_msg: type) -> None:
+ def test_bottom_quotes_dropped(self, make_msg: MsgFactory) -> None:
"""Trailing quoted blocks at the end of a message are dropped."""
body = 'My reply.\n> Original message.\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'My reply.' in text
assert 'Original message.' not in text
- def test_diff_preserved(self, make_msg: type) -> None:
+ def test_diff_preserved(self, make_msg: MsgFactory) -> None:
"""Messages with diff content are not minimized."""
body = (
'> Quoted context.\n'
@@ -252,42 +254,42 @@ class TestMinimizeThread:
'-old\n'
'+new\n'
)
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'diff --git' in text
assert 'Quoted context.' in text
- def test_diffstat_preserved(self, make_msg: type) -> None:
+ def test_diffstat_preserved(self, make_msg: MsgFactory) -> None:
"""Messages with diffstat content are not minimized."""
body = '> Quoted.\n 3 files changed, 10 insertions(+), 5 deletions(-)\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'Quoted.' in text
assert '3 files changed' in text
- def test_empty_after_minimize_dropped(self, make_msg: type) -> None:
+ def test_empty_after_minimize_dropped(self, make_msg: MsgFactory) -> None:
"""Messages that become empty after minimization are dropped."""
body = '> Only quoted text.\n>> And deeper quotes.\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
assert len(result) == 0
- def test_signature_preserved(self, make_msg: type) -> None:
+ def test_signature_preserved(self, make_msg: MsgFactory) -> None:
"""Compliant signatures are kept after quote processing."""
body = 'My reply.\n-- \nJane Doe\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'My reply.' in text
assert '-- \n' in text
assert 'Jane Doe' in text
- def test_trailing_quote_before_sig_dropped(self, make_msg: type) -> None:
+ def test_trailing_quote_before_sig_dropped(self, make_msg: MsgFactory) -> None:
"""A trailing quoted block before a signature is dropped."""
body = 'My reply.\n> Huge quoted original.\n> More quoting.\n-- \nJane Doe\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'My reply.' in text
@@ -295,14 +297,14 @@ class TestMinimizeThread:
assert 'More quoting.' not in text
assert 'Jane Doe' in text
- def test_only_quotes_before_sig_dropped(self, make_msg: type) -> None:
+ def test_only_quotes_before_sig_dropped(self, make_msg: MsgFactory) -> None:
"""Message with only quotes before a signature is dropped entirely."""
body = '> Only quoted text.\n-- \nJane Doe\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
assert len(result) == 0
- def test_reduce_quote_context(self, make_msg: type) -> None:
+ def test_reduce_quote_context(self, make_msg: MsgFactory) -> None:
"""Long quoted blocks are reduced to the last paragraph."""
body = (
'On Monday, Julius Caesar wrote:\n'
@@ -318,7 +320,7 @@ class TestMinimizeThread:
'> Third paragraph line two.\n'
'My reply here.\n'
)
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg], reduce_quote_context=True)
text = result[0].get_payload()
assert '... skip 7 lines ...' in text
@@ -328,7 +330,7 @@ class TestMinimizeThread:
assert 'Second paragraph' not in text
assert 'My reply here.' in text
- def test_reduce_quote_context_short_quote_untouched(self, make_msg: type) -> None:
+ def test_reduce_quote_context_short_quote_untouched(self, make_msg: MsgFactory) -> None:
"""Quotes with 5 or fewer skippable lines are left alone."""
body = (
'> Line one.\n'
@@ -337,29 +339,29 @@ class TestMinimizeThread:
'> Last para.\n'
'Reply.\n'
)
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg], reduce_quote_context=True)
text = result[0].get_payload()
assert 'skip' not in text
assert 'Line one.' in text
assert 'Last para.' in text
- def test_reduce_quote_context_off_by_default(self, make_msg: type) -> None:
+ def test_reduce_quote_context_off_by_default(self, make_msg: MsgFactory) -> None:
"""Long quotes are untouched when reduce_quote_context is False."""
lines = ''.join(f'> Line {i}.\n' for i in range(20))
body = f'{lines}Reply.\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg])
text = result[0].get_payload()
assert 'skip' not in text
assert 'Line 0.' in text
assert 'Line 19.' in text
- def test_reduce_quote_context_preserves_sig(self, make_msg: type) -> None:
+ def test_reduce_quote_context_preserves_sig(self, make_msg: MsgFactory) -> None:
"""Signature is preserved when reducing quote context."""
lines = ''.join(f'> Line {i}.\n' for i in range(10))
body = f'On Monday, someone wrote:\n{lines}>\n> Last para.\nReply.\n-- \nKR\n'
- msg = make_msg.create(body=body)
+ msg = make_msg(body=body)
result = minimize_thread([msg], reduce_quote_context=True)
text = result[0].get_payload()
assert 'skip' in text
@@ -368,11 +370,11 @@ class TestMinimizeThread:
assert '-- \n' in text
assert 'KR' in text
- def test_multiple_messages(self, make_msg: type) -> None:
+ def test_multiple_messages(self, make_msg: MsgFactory) -> None:
"""Multiple messages in a thread are all processed."""
- msg1 = make_msg.create(body='First message.\n')
- msg2 = make_msg.create(body='Reply.\n> First message.\n')
- msg3 = make_msg.create(body='> Only quotes.\n')
+ msg1 = make_msg(body='First message.\n')
+ msg2 = make_msg(body='Reply.\n> First message.\n')
+ msg3 = make_msg(body='> Only quotes.\n')
result = minimize_thread([msg1, msg2, msg3])
# msg3 should be dropped (all quotes)
assert len(result) == 2
diff --git a/tests/test_message.py b/tests/test_message.py
index 1a20afb..947e112 100644
--- a/tests/test_message.py
+++ b/tests/test_message.py
@@ -4,6 +4,8 @@ from __future__ import annotations
from email.message import EmailMessage
+from conftest import MsgFactory
+
from liblore.utils import clean_header, get_clean_msgid, parse_message
@@ -35,16 +37,16 @@ class TestCleanHeader:
class TestGetCleanMsgid:
- def test_extracts_msgid(self, make_msg: type) -> None:
- msg = make_msg.create(msgid='test123@example.com')
+ def test_extracts_msgid(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(msgid='test123@example.com')
assert get_clean_msgid(msg) == 'test123@example.com'
def test_missing_header(self) -> None:
msg = EmailMessage()
assert get_clean_msgid(msg) is None
- def test_custom_header(self, make_msg: type) -> None:
- msg = make_msg.create(in_reply_to='parent@example.com')
+ def test_custom_header(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(in_reply_to='parent@example.com')
assert get_clean_msgid(msg, 'In-Reply-To') == 'parent@example.com'
diff --git a/tests/test_thread.py b/tests/test_thread.py
index b404a2a..62a9a1e 100644
--- a/tests/test_thread.py
+++ b/tests/test_thread.py
@@ -2,35 +2,37 @@
"""Tests for liblore.thread."""
from __future__ import annotations
+from conftest import MsgFactory
+
from liblore.utils import get_clean_msgid, get_strict_thread
class TestGetStrictThread:
- def test_simple_thread(self, make_msg: type) -> None:
- root = make_msg.create(msgid='root@x.com', subject='Root')
- reply = make_msg.create(
+ def test_simple_thread(self, make_msg: MsgFactory) -> None:
+ root = make_msg(msgid='root@x.com', subject='Root')
+ reply = make_msg(
msgid='reply@x.com',
subject='Re: Root',
in_reply_to='root@x.com',
)
- unrelated = make_msg.create(msgid='other@x.com', subject='Other')
+ unrelated = make_msg(msgid='other@x.com', subject='Other')
result = get_strict_thread([root, reply, unrelated], 'root@x.com')
assert result is not None
ids = {get_clean_msgid(m) for m in result}
assert ids == {'root@x.com', 'reply@x.com'}
- def test_returns_none_for_missing_msgid(self, make_msg: type) -> None:
- msg = make_msg.create(msgid='exists@x.com')
+ def test_returns_none_for_missing_msgid(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(msgid='exists@x.com')
result = get_strict_thread([msg], 'nonexistent@x.com')
assert result is None
- def test_noparent(self, make_msg: type) -> None:
- parent = make_msg.create(msgid='parent@x.com')
- child = make_msg.create(
+ def test_noparent(self, make_msg: MsgFactory) -> None:
+ parent = make_msg(msgid='parent@x.com')
+ child = make_msg(
msgid='child@x.com',
in_reply_to='parent@x.com',
)
- grandchild = make_msg.create(
+ grandchild = make_msg(
msgid='grandchild@x.com',
in_reply_to='child@x.com',
)
@@ -44,10 +46,10 @@ class TestGetStrictThread:
assert 'child@x.com' in ids
assert 'grandchild@x.com' in ids
- def test_references_chain(self, make_msg: type) -> None:
- msg1 = make_msg.create(msgid='a@x.com')
- msg2 = make_msg.create(msgid='b@x.com', references=['a@x.com'])
- msg3 = make_msg.create(
+ def test_references_chain(self, make_msg: MsgFactory) -> None:
+ msg1 = make_msg(msgid='a@x.com')
+ msg2 = make_msg(msgid='b@x.com', references=['a@x.com'])
+ msg3 = make_msg(
msgid='c@x.com',
references=['a@x.com', 'b@x.com'],
)
@@ -55,8 +57,8 @@ class TestGetStrictThread:
assert result is not None
assert len(result) == 3
- def test_single_message(self, make_msg: type) -> None:
- msg = make_msg.create(msgid='solo@x.com')
+ def test_single_message(self, make_msg: MsgFactory) -> None:
+ msg = make_msg(msgid='solo@x.com')
result = get_strict_thread([msg], 'solo@x.com')
assert result is not None
assert len(result) == 1
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 03/14] Add ruff import checks to b4 CI
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 01/14] Add ci script and mypy suppressions Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 02/14] Type make_msg and drop test suppressions Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 04/14] Add ruff format check to CI Tamir Duberstein
` (11 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Run ruff as part of the b4 CI check script and enable import sorting in
Ruff so import ordering regressions show up in the review UI.
Keep Ruff as a subprocess and document the related upstream API
limitation.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
ci.sh | 1 +
pyproject.toml | 4 ++--
src/liblore/__init__.py | 1 -
src/liblore/node.py | 3 +--
src/liblore/utils.py | 3 +--
tests/conftest.py | 3 +--
| 1 -
tests/test_email_utils.py | 1 -
tests/test_formatting.py | 3 +--
tests/test_mbox.py | 4 +---
tests/test_node.py | 1 -
11 files changed, 8 insertions(+), 17 deletions(-)
diff --git a/ci.sh b/ci.sh
index 1e8ccb3..3001db2 100755
--- a/ci.sh
+++ b/ci.sh
@@ -2,5 +2,6 @@
set -eu
+uv run ruff check
uv run mypy .
uv run pytest --durations=0
diff --git a/pyproject.toml b/pyproject.toml
index 7c9e1da..38be519 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -57,5 +57,5 @@ strict = true
module = "authheaders"
ignore_missing_imports = true
-[tool.ruff]
-target-version = "py39"
+[tool.ruff.lint]
+extend-select = ["I"]
diff --git a/src/liblore/__init__.py b/src/liblore/__init__.py
index ff4314a..1ffaaeb 100644
--- a/src/liblore/__init__.py
+++ b/src/liblore/__init__.py
@@ -3,7 +3,6 @@
"""liblore — shared library for public-inbox / lore.kernel.org access."""
import email.charset
import email.policy
-
from email.message import EmailMessage
__version__ = '0.7.1'
diff --git a/src/liblore/node.py b/src/liblore/node.py
index 5fb6343..66da1d7 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -15,11 +15,10 @@ import time
import types
import urllib.parse
from datetime import datetime, timezone
+from email.message import EmailMessage
import requests
-from email.message import EmailMessage
-
from liblore import LibloreError, RemoteError
from liblore.utils import (
get_strict_thread,
diff --git a/src/liblore/utils.py b/src/liblore/utils.py
index 8d485da..e8926c6 100644
--- a/src/liblore/utils.py
+++ b/src/liblore/utils.py
@@ -3,7 +3,6 @@
"""Message parsing, email utilities, threading, and mbox splitting."""
from __future__ import annotations
-from collections.abc import Sequence
import datetime
import email.header
import email.parser
@@ -14,7 +13,7 @@ import logging
import re
import textwrap
import urllib.parse
-
+from collections.abc import Sequence
from email.message import EmailMessage
from liblore import emlpolicy
diff --git a/tests/conftest.py b/tests/conftest.py
index b84fc35..ec8ec80 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -4,12 +4,11 @@ from __future__ import annotations
import email.utils
import textwrap
+from email.message import EmailMessage
from typing import Protocol
import pytest
-from email.message import EmailMessage
-
from liblore import emlpolicy
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index 9edf4cd..9097048 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -13,7 +13,6 @@ import pytest
from liblore import LibloreError
from liblore.node import LoreNode
-
# =====================================================================
# Import-time validation
# =====================================================================
diff --git a/tests/test_email_utils.py b/tests/test_email_utils.py
index ad8775f..74216d4 100644
--- a/tests/test_email_utils.py
+++ b/tests/test_email_utils.py
@@ -16,7 +16,6 @@ from liblore.utils import (
)
-
class TestMsgGetSubject:
def test_plain_subject(self, make_msg: MsgFactory) -> None:
msg = make_msg(subject='Just a plain subject')
diff --git a/tests/test_formatting.py b/tests/test_formatting.py
index a13593e..cb3cf80 100644
--- a/tests/test_formatting.py
+++ b/tests/test_formatting.py
@@ -2,10 +2,9 @@
"""Tests for email formatting and thread minimization."""
from __future__ import annotations
-import pytest
-
from email.message import EmailMessage
+import pytest
from conftest import MsgFactory
from liblore.utils import (
diff --git a/tests/test_mbox.py b/tests/test_mbox.py
index 3815e3f..9bc2271 100644
--- a/tests/test_mbox.py
+++ b/tests/test_mbox.py
@@ -3,9 +3,9 @@
from __future__ import annotations
import textwrap
-
from email.message import EmailMessage
+from liblore import emlpolicy
from liblore.utils import (
get_clean_msgid,
get_preferred_duplicate,
@@ -15,8 +15,6 @@ from liblore.utils import (
split_mbox_as_bytes,
)
-from liblore import emlpolicy
-
class TestSplitMbox:
def test_splits_two_messages(self, sample_mbox: bytes) -> None:
diff --git a/tests/test_node.py b/tests/test_node.py
index c2e7bf3..54d76d8 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -14,7 +14,6 @@ import requests
from liblore import RemoteError
from liblore.node import LoreNode
-
# =====================================================================
# Session management
# =====================================================================
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 04/14] Add ruff format check to CI
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (2 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 03/14] Add ruff import checks to b4 CI Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 05/14] Add pyright strict checks " Tamir Duberstein
` (10 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Enable ruff format checking in the b4 CI script and configure Ruff's
formatter in pyproject.toml.
Apply a one-time repo-wide format pass so the new check enforces the
current style without leaving the branch permanently red.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
ci.sh | 1 +
pyproject.toml | 3 +
src/liblore/__init__.py | 1 +
src/liblore/node.py | 61 ++++++-----
src/liblore/utils.py | 86 +++++++++++-----
tests/conftest.py | 1 +
| 4 +
tests/test_cache.py | 23 +++--
tests/test_email_utils.py | 2 +-
tests/test_formatting.py | 251 ++++++++++++++++++++++++++++-----------------
tests/test_mbox.py | 3 +
tests/test_message.py | 1 +
tests/test_node.py | 152 +++++++++++++++++++--------
tests/test_thread.py | 1 +
14 files changed, 395 insertions(+), 195 deletions(-)
diff --git a/ci.sh b/ci.sh
index 3001db2..ec0baf8 100755
--- a/ci.sh
+++ b/ci.sh
@@ -2,6 +2,7 @@
set -eu
+uv run ruff format --check
uv run ruff check
uv run mypy .
uv run pytest --durations=0
diff --git a/pyproject.toml b/pyproject.toml
index 38be519..9bfadbe 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -59,3 +59,6 @@ ignore_missing_imports = true
[tool.ruff.lint]
extend-select = ["I"]
+
+[tool.ruff.format]
+quote-style = "single"
diff --git a/src/liblore/__init__.py b/src/liblore/__init__.py
index 1ffaaeb..2ba94e7 100644
--- a/src/liblore/__init__.py
+++ b/src/liblore/__init__.py
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2025-2026 The Linux Foundation
"""liblore — shared library for public-inbox / lore.kernel.org access."""
+
import email.charset
import email.policy
from email.message import EmailMessage
diff --git a/src/liblore/node.py b/src/liblore/node.py
index 66da1d7..503428b 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2025-2026 The Linux Foundation
"""LoreNode — primary API for interacting with public-inbox servers."""
+
from __future__ import annotations
import concurrent.futures
@@ -49,7 +50,9 @@ def _get_config_from_git(
try:
result = subprocess.run(
['git', 'config', '-z', '--get-regexp', regexp],
- capture_output=True, text=True, timeout=5,
+ capture_output=True,
+ text=True,
+ timeout=5,
)
if result.returncode != 0:
return {}
@@ -99,7 +102,9 @@ def _get_subsection_config(
try:
result = subprocess.run(
['git', 'config', '-z', '--get-regexp', f'^{escaped}'],
- capture_output=True, text=True, timeout=5,
+ capture_output=True,
+ text=True,
+ timeout=5,
)
if result.returncode != 0:
return {}
@@ -114,7 +119,7 @@ def _get_subsection_config(
key, value = entry.split('\n', 1)
else:
key, value = entry, 'true'
- varname = key[len(prefix):].lower()
+ varname = key[len(prefix) :].lower()
if varname in multivals:
existing = config.get(varname)
if not isinstance(existing, list):
@@ -168,7 +173,7 @@ class LoreNode:
# Build the ordered list of origins to try
self._all_origins: list[str] = []
- for fb in (fallback_urls or []):
+ for fb in fallback_urls or []:
fb = fb.rstrip('/')
fb_parsed = urllib.parse.urlparse(fb)
if not fb_parsed.scheme or not fb_parsed.netloc:
@@ -191,6 +196,7 @@ class LoreNode:
if add_auth_headers:
try:
import authheaders
+
self._authheaders = authheaders
except ImportError:
raise LibloreError(
@@ -280,7 +286,9 @@ class LoreNode:
# Try [liblore "<origin>"] first, fall back to [lore] for
# lore.kernel.org.
gitcfg = _get_subsection_config(
- 'liblore', origin, multivals=['fallback'],
+ 'liblore',
+ origin,
+ multivals=['fallback'],
)
if not gitcfg and parsed.netloc == 'lore.kernel.org':
gitcfg = _get_config_from_git(r'^lore\.', multivals=['fallback'])
@@ -334,7 +342,9 @@ class LoreNode:
"""
return self._user_agent_plus
- def set_user_agent(self, app_name: str, version: str, plus: str | None = None) -> None:
+ def set_user_agent(
+ self, app_name: str, version: str, plus: str | None = None
+ ) -> None:
"""Set the User-Agent to ``app_name/version`` (optionally ``+plus``).
When *plus* is not provided, the value from ``lore.useragentplus``
@@ -415,7 +425,7 @@ class LoreNode:
def _rewrite_url(self, url: str, origin: str) -> str:
"""Replace the canonical origin in *url* with *origin*."""
- return origin + url[len(self._canonical_origin):]
+ return origin + url[len(self._canonical_origin) :]
def _probe_cache_key(self) -> str:
"""Cache key for probe results, stable regardless of current order."""
@@ -596,12 +606,14 @@ class LoreNode:
logger.debug('Trying %s %s', method, request_url)
try:
resp: requests.Response = getattr(
- session, method.lower(),
+ session,
+ method.lower(),
)(request_url, **kwargs)
except (requests.ConnectionError, requests.Timeout) as exc:
logger.warning(
'Request to %s failed (%s), trying next host',
- origin, exc,
+ origin,
+ exc,
)
last_exc = exc
continue
@@ -609,7 +621,8 @@ class LoreNode:
if resp.status_code >= 500:
logger.warning(
'Request to %s returned %d, trying next host',
- origin, resp.status_code,
+ origin,
+ resp.status_code,
)
last_resp = resp
continue
@@ -622,9 +635,7 @@ class LoreNode:
return last_resp
if last_exc is not None:
- raise RemoteError(
- f'All hosts failed for {url}: {last_exc}'
- ) from last_exc
+ raise RemoteError(f'All hosts failed for {url}: {last_exc}') from last_exc
# last_resp must be a 5xx from the final origin
assert last_resp is not None
@@ -650,7 +661,9 @@ class LoreNode:
return None
age = int(time.time() - st.st_mtime)
if age > self._cache_ttl:
- logger.debug('Cache expired (%ds > %ds): %s', age, self._cache_ttl, key[:12])
+ logger.debug(
+ 'Cache expired (%ds > %ds): %s', age, self._cache_ttl, key[:12]
+ )
try:
os.unlink(path)
except OSError:
@@ -699,8 +712,12 @@ class LoreNode:
for msg in msgs:
msg_bytes = msg.as_bytes()
auth_result = self._authheaders.authenticate_message(
- msg_bytes, 'liblore',
- dkim=True, dmarc=True, arc=True, spf=False,
+ msg_bytes,
+ 'liblore',
+ dkim=True,
+ dmarc=True,
+ arc=True,
+ spf=False,
)
if auth_result:
# authheaders returns the full header line, so strip
@@ -785,7 +802,9 @@ class LoreNode:
return cached
t_param = '&t=1' if full_threads else ''
- query_url = self._url + '/?x=m' + t_param + '&q=' + urllib.parse.quote_plus(query)
+ query_url = (
+ self._url + '/?x=m' + t_param + '&q=' + urllib.parse.quote_plus(query)
+ )
resp = self._request('POST', query_url, data='x=m')
if resp.status_code != 200:
raise RemoteError('Server returned an error: %s' % resp.status_code)
@@ -842,9 +861,7 @@ class LoreNode:
if strict:
strict_msgs = get_strict_thread(msgs, msgid)
if not isinstance(strict_msgs, list) or not len(strict_msgs):
- raise LookupError(
- 'No messages found for msgid=%s' % msgid
- )
+ raise LookupError('No messages found for msgid=%s' % msgid)
msgs = strict_msgs
if sort:
@@ -1013,9 +1030,7 @@ class LoreNode:
for i, query in enumerate(queries):
if i > 0:
time.sleep(0.1)
- results.append(
- self.get_thread_by_query(query, full_threads=full_threads)
- )
+ results.append(self.get_thread_by_query(query, full_threads=full_threads))
return results
def validate(self) -> None:
diff --git a/src/liblore/utils.py b/src/liblore/utils.py
index e8926c6..04823bf 100644
--- a/src/liblore/utils.py
+++ b/src/liblore/utils.py
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2025-2026 The Linux Foundation
"""Message parsing, email utilities, threading, and mbox splitting."""
+
from __future__ import annotations
import datetime
@@ -36,6 +37,7 @@ _GTFROM_RE = re.compile(rb'^>+From ')
# Header and message-ID utilities
# =====================================================================
+
def clean_header(hdrval: str | None) -> str:
"""Decode an RFC 2047 encoded header and normalise whitespace.
@@ -166,7 +168,6 @@ def msg_get_payload(
return '\n'.join(stripped)
-
# Adapted from email._parseaddr — RFC 5322 specials that require quoting
_QSPECIALS = re.compile(r'[()<>@,:;.\"\[\]]')
@@ -187,7 +188,11 @@ def format_addrs(pairs: list[tuple[str, str]], clean: bool = True) -> str:
if clean:
name = clean_header(name)
# Work around https://github.com/python/cpython/issues/100900
- if not name.startswith('=?') and not name.startswith('"') and _QSPECIALS.search(name):
+ if (
+ not name.startswith('=?')
+ and not name.startswith('"')
+ and _QSPECIALS.search(name)
+ ):
addrs.append(f'"{email.utils.quote(name)}" <{addr}>')
continue
if name.isascii():
@@ -216,7 +221,8 @@ def wrap_header(hdr: tuple[str, str], width: int = 75, nl: str = '\n') -> bytes:
for addr in email.utils.getaddresses([hval]):
if not addr[0].isascii():
enc_name = email.quoprimime.header_encode(
- addr[0].encode(), charset='utf-8')
+ addr[0].encode(), charset='utf-8'
+ )
qp = format_addrs([(enc_name, addr[1])], clean=False)
else:
qp = format_addrs([addr], clean=False)
@@ -236,14 +242,19 @@ def wrap_header(hdr: tuple[str, str], width: int = 75, nl: str = '\n') -> bytes:
return hdata.encode()
# Trick: replace ': ' with ':_' so textwrap doesn't break there
hdata = hdata.replace(': ', ':_', 1)
- wrapped = textwrap.wrap(hdata, break_long_words=False,
- break_on_hyphens=False,
- subsequent_indent=' ', width=width)
+ wrapped = textwrap.wrap(
+ hdata,
+ break_long_words=False,
+ break_on_hyphens=False,
+ subsequent_indent=' ',
+ width=width,
+ )
return nl.join(wrapped).replace(':_', ': ', 1).encode()
# Non-ASCII: encode as RFC 2047 quoted-printable
qp = f'{hname}: ' + email.quoprimime.header_encode(
- hval.encode(), charset='utf-8')
+ hval.encode(), charset='utf-8'
+ )
if len(qp) <= width:
return qp.encode()
@@ -252,7 +263,7 @@ def wrap_header(hdr: tuple[str, str], width: int = 75, nl: str = '\n') -> bytes:
if len(_parts):
wrapat -= 1
# Don't break in the middle of a =XX escape sequence
- while '=' in qp[wrapat - 2:wrapat]:
+ while '=' in qp[wrapat - 2 : wrapat]:
wrapat -= 1
_parts.append(qp[:wrapat] + '?=')
qp = '=?utf-8?q?' + qp[wrapat:]
@@ -434,8 +445,14 @@ DIFFSTAT_RE = re.compile(r'^\s*\d+ file.*changed', re.M)
# Default set of headers to keep when minimizing messages.
MINIMIZE_KEEP_HEADERS: tuple[str, ...] = (
- 'From', 'To', 'Cc', 'Subject', 'Date',
- 'Message-ID', 'Reply-To', 'In-Reply-To',
+ 'From',
+ 'To',
+ 'Cc',
+ 'Subject',
+ 'Date',
+ 'Message-ID',
+ 'Reply-To',
+ 'In-Reply-To',
)
@@ -481,7 +498,11 @@ def minimize_thread(
body = msg_get_payload(msg, strip_signature=False)
- if not re.search(r'^>', body, re.M) or DIFF_RE.search(body) or DIFFSTAT_RE.search(body):
+ if (
+ not re.search(r'^>', body, re.M)
+ or DIFF_RE.search(body)
+ or DIFFSTAT_RE.search(body)
+ ):
mmsg.set_payload(body, charset='utf-8')
mmsgs.append(mmsg)
continue
@@ -548,11 +569,15 @@ def minimize_thread(
skipped = last_break
if skipped <= 5:
continue
- last_para = qchunk[last_break + 1:]
- chunks[idx] = (True, [
- f'> [... skip {skipped} lines ...]',
- '>',
- ] + last_para)
+ last_para = qchunk[last_break + 1 :]
+ chunks[idx] = (
+ True,
+ [
+ f'> [... skip {skipped} lines ...]',
+ '>',
+ ]
+ + last_para,
+ )
parts: list[str] = []
for _quoted, chunk in chunks:
@@ -599,7 +624,7 @@ def _get_raw_header(raw: bytes, name: str) -> str | None:
if value is not None:
break
if line.lower().startswith(prefix):
- value = line[len(prefix):].strip()
+ value = line[len(prefix) :].strip()
if value is not None:
return value.decode('ascii', errors='replace')
@@ -680,7 +705,8 @@ DEFAULT_LISTID_PREFERENCE: list[str] = [
def _listid_preference_index(
- listid: str | None, listid_preference: list[str],
+ listid: str | None,
+ listid_preference: list[str],
) -> int:
"""Return the preference index for a List-Id value.
@@ -716,16 +742,17 @@ def get_preferred_duplicate(
if listid_preference is None:
listid_preference = DEFAULT_LISTID_PREFERENCE
- idx1 = _listid_preference_index(
- get_clean_msgid(msg1, 'List-Id'), listid_preference)
- idx2 = _listid_preference_index(
- get_clean_msgid(msg2, 'List-Id'), listid_preference)
+ idx1 = _listid_preference_index(get_clean_msgid(msg1, 'List-Id'), listid_preference)
+ idx2 = _listid_preference_index(get_clean_msgid(msg2, 'List-Id'), listid_preference)
if idx1 <= idx2:
- logger.debug('Picked duplicate from preferred source: %s',
- get_clean_msgid(msg1, 'List-Id'))
+ logger.debug(
+ 'Picked duplicate from preferred source: %s',
+ get_clean_msgid(msg1, 'List-Id'),
+ )
return msg1
- logger.debug('Picked duplicate from preferred source: %s',
- get_clean_msgid(msg2, 'List-Id'))
+ logger.debug(
+ 'Picked duplicate from preferred source: %s', get_clean_msgid(msg2, 'List-Id')
+ )
return msg2
@@ -799,9 +826,10 @@ def split_and_dedupe(
# =====================================================================
-# Pure helpers
+# Pure helpers
# =====================================================================
+
def get_msgid_from_url(url: str) -> str:
"""Extract a message ID from a lore.kernel.org URL.
@@ -809,7 +837,9 @@ def get_msgid_from_url(url: str) -> str:
returned with angle brackets stripped.
"""
if '://' in url:
- matches = re.search(r'^https?://[^@]+/([^/]+(?:@|%40)[^/]+)', url, re.IGNORECASE)
+ matches = re.search(
+ r'^https?://[^@]+/([^/]+(?:@|%40)[^/]+)', url, re.IGNORECASE
+ )
if matches:
return urllib.parse.unquote(matches.groups()[0])
return url.strip('<>')
diff --git a/tests/conftest.py b/tests/conftest.py
index ec8ec80..28fedaf 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Shared fixtures for liblore tests."""
+
from __future__ import annotations
import email.utils
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index 9097048..8cec624 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for optional authheaders integration in LoreNode."""
+
from __future__ import annotations
import gzip
@@ -17,6 +18,7 @@ from liblore.node import LoreNode
# Import-time validation
# =====================================================================
+
class TestAuthHeadersImport:
def test_raises_when_authheaders_missing(self) -> None:
with patch.dict(sys.modules, {'authheaders': None}):
@@ -41,6 +43,7 @@ class TestAuthHeadersImport:
# _authenticate_msgs
# =====================================================================
+
class TestAuthenticateMsgs:
def test_noop_when_disabled(self) -> None:
node = LoreNode()
@@ -116,6 +119,7 @@ class TestAuthenticateMsgs:
# Integration with fetch methods
# =====================================================================
+
class TestAuthInFetchMethods:
@pytest.fixture()
def auth_node(self, sample_mbox: bytes) -> LoreNode:
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 9e7933b..b822e7c 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for LoreNode filesystem caching."""
+
from __future__ import annotations
import gzip
@@ -82,7 +83,6 @@ class TestCacheReadWrite:
class TestClearCache:
-
def test_clears_cache_files(self, tmp_path: Path) -> None:
node = LoreNode(cache_dir=str(tmp_path))
node._cache_write(node._cache_key('a', '1'), b'data1')
@@ -104,7 +104,6 @@ class TestClearCache:
class TestProperties:
-
def test_url_property(self) -> None:
node = LoreNode('https://lore.kernel.org/all/')
# Trailing slash is stripped by constructor
@@ -122,8 +121,12 @@ class TestProperties:
class TestCachedMethods:
"""Integration tests: verify cache hit avoids network, cache miss hits network."""
- def _make_node(self, tmp_path: Path, sample_mbox: bytes) -> tuple[LoreNode, MagicMock]:
- node = LoreNode('https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60)
+ def _make_node(
+ self, tmp_path: Path, sample_mbox: bytes
+ ) -> tuple[LoreNode, MagicMock]:
+ node = LoreNode(
+ 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
+ )
mock_session = MagicMock()
mock_resp = MagicMock()
mock_resp.status_code = 200
@@ -158,7 +161,9 @@ class TestCachedMethods:
assert mock_session.post.call_count == 2
def test_get_message_by_msgid_caches(self, tmp_path: Path) -> None:
- node = LoreNode('https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60)
+ node = LoreNode(
+ 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
+ )
mock_session = MagicMock()
mock_resp = MagicMock()
mock_resp.status_code = 200
@@ -197,7 +202,9 @@ class TestCachedMethods:
# tmp_path should be empty since we didn't set cache_dir to it
assert list(tmp_path.iterdir()) == []
- def test_fetch_thread_since_not_cached(self, tmp_path: Path, sample_mbox: bytes) -> None:
+ def test_fetch_thread_since_not_cached(
+ self, tmp_path: Path, sample_mbox: bytes
+ ) -> None:
"""_fetch_thread_since should NOT be cached."""
node, mock_session = self._make_node(tmp_path, sample_mbox)
node._fetch_thread_since('test@example.com', 'dt:20240101..')
@@ -207,7 +214,9 @@ class TestCachedMethods:
def test_error_not_cached(self, tmp_path: Path) -> None:
"""Network errors should not be cached."""
- node = LoreNode('https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60)
+ node = LoreNode(
+ 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
+ )
mock_session = MagicMock()
mock_resp = MagicMock()
mock_resp.status_code = 404
diff --git a/tests/test_email_utils.py b/tests/test_email_utils.py
index 74216d4..9276010 100644
--- a/tests/test_email_utils.py
+++ b/tests/test_email_utils.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for liblore.email_utils."""
+
from __future__ import annotations
from email.message import EmailMessage
@@ -100,7 +101,6 @@ class TestMsgGetPayload:
assert msg_get_payload(msg) == ''
-
class TestMsgGetRecipients:
def test_to_cc_from(self) -> None:
msg = EmailMessage(policy=emlpolicy)
diff --git a/tests/test_formatting.py b/tests/test_formatting.py
index cb3cf80..f179a3a 100644
--- a/tests/test_formatting.py
+++ b/tests/test_formatting.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for email formatting and thread minimization."""
+
from __future__ import annotations
from email.message import EmailMessage
@@ -19,105 +20,168 @@ from liblore.utils import (
# Adapted from b4 test_format_addrs — validates cpython#100900 workaround
# and RFC 2047 handling in address formatting.
class TestFormatAddrs:
- @pytest.mark.parametrize('pairs,verify,clean', [
- ([('', 'foo@example.com'), ('Foo Bar', 'bar@example.com')],
- 'foo@example.com, Foo Bar <bar@example.com>', True),
- ([('', 'foo@example.com'), ('Foo, Bar', 'bar@example.com')],
- 'foo@example.com, "Foo, Bar" <bar@example.com>', True),
- ([('', 'foo@example.com'), ('F\u00f4o, Bar', 'bar@example.com')],
- 'foo@example.com, "F\u00f4o, Bar" <bar@example.com>', True),
- ([('', 'foo@example.com'), ('=?utf-8?q?Qu=C3=BBx_Foo?=', 'quux@example.com')],
- 'foo@example.com, Qu\u00fbx Foo <quux@example.com>', True),
- ([('', 'foo@example.com'), ('=?utf-8?q?Qu=C3=BBx=2C_Foo?=', 'quux@example.com')],
- 'foo@example.com, "Qu\u00fbx, Foo" <quux@example.com>', True),
- ([('', 'foo@example.com'), ('=?utf-8?q?Qu=C3=BBx=2C_Foo?=', 'quux@example.com')],
- 'foo@example.com, =?utf-8?q?Qu=C3=BBx=2C_Foo?= <quux@example.com>', False),
- ])
- def test_format_addrs(self, pairs: list[tuple[str, str]],
- verify: str, clean: bool) -> None:
+ @pytest.mark.parametrize(
+ 'pairs,verify,clean',
+ [
+ (
+ [('', 'foo@example.com'), ('Foo Bar', 'bar@example.com')],
+ 'foo@example.com, Foo Bar <bar@example.com>',
+ True,
+ ),
+ (
+ [('', 'foo@example.com'), ('Foo, Bar', 'bar@example.com')],
+ 'foo@example.com, "Foo, Bar" <bar@example.com>',
+ True,
+ ),
+ (
+ [('', 'foo@example.com'), ('F\u00f4o, Bar', 'bar@example.com')],
+ 'foo@example.com, "F\u00f4o, Bar" <bar@example.com>',
+ True,
+ ),
+ (
+ [
+ ('', 'foo@example.com'),
+ ('=?utf-8?q?Qu=C3=BBx_Foo?=', 'quux@example.com'),
+ ],
+ 'foo@example.com, Qu\u00fbx Foo <quux@example.com>',
+ True,
+ ),
+ (
+ [
+ ('', 'foo@example.com'),
+ ('=?utf-8?q?Qu=C3=BBx=2C_Foo?=', 'quux@example.com'),
+ ],
+ 'foo@example.com, "Qu\u00fbx, Foo" <quux@example.com>',
+ True,
+ ),
+ (
+ [
+ ('', 'foo@example.com'),
+ ('=?utf-8?q?Qu=C3=BBx=2C_Foo?=', 'quux@example.com'),
+ ],
+ 'foo@example.com, =?utf-8?q?Qu=C3=BBx=2C_Foo?= <quux@example.com>',
+ False,
+ ),
+ ],
+ )
+ def test_format_addrs(
+ self, pairs: list[tuple[str, str]], verify: str, clean: bool
+ ) -> None:
assert format_addrs(pairs, clean) == verify
# Adapted from b4 test_header_wrapping — validates RFC 2047 encoding,
# line wrapping, and proper handling of address vs non-address headers.
class TestWrapHeader:
- @pytest.mark.parametrize('hval,verify', [
- # Short ASCII — no wrapping needed
- ('short-ascii', 'short-ascii'),
- # Short Unicode — RFC 2047 QP encoded
- ('short-unic\u00f4de', '=?utf-8?q?short-unic=C3=B4de?='),
- # Long ASCII — wrapped at word boundary
- ('Lorem ipsum dolor sit amet consectetur adipiscing elit '
- 'sed do eiusmod tempor incididunt ut labore et dolore magna aliqua',
- 'Lorem ipsum dolor sit amet consectetur adipiscing elit sed do\n'
- ' eiusmod tempor incididunt ut labore et dolore magna aliqua'),
- # Long Unicode — split across multiple encoded lines
- ('Lorem \u00eepsum dolor sit amet consectetur adipiscing el\u00eet '
- 'sed do eiusmod temp\u00f4r incididunt ut labore et dol\u00f4re magna aliqua',
- '=?utf-8?q?Lorem_=C3=AEpsum_dolor_sit_amet_consectetur_adipiscin?=\n'
- ' =?utf-8?q?g_el=C3=AEt_sed_do_eiusmod_temp=C3=B4r_incididunt_ut_labore_et?=\n'
- ' =?utf-8?q?_dol=C3=B4re_magna_aliqua?='),
- # Exactly 75 chars — boundary condition
- ('Lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiu',
- 'Lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiu'),
- # Unicode on escape boundary
- ('Lorem ipsum dolor sit amet consectetur adipiscin el\u00eet',
- '=?utf-8?q?Lorem_ipsum_dolor_sit_amet_consectetur_adipiscin_el?=\n'
- ' =?utf-8?q?=C3=AEt?='),
- # Unicode 1 char too long
- ('Lorem ipsum dolor sit amet consectetur adipi el\u00eet',
- '=?utf-8?q?Lorem_ipsum_dolor_sit_amet_consectetur_adipi_el=C3=AE?=\n'
- ' =?utf-8?q?t?='),
- ])
+ @pytest.mark.parametrize(
+ 'hval,verify',
+ [
+ # Short ASCII — no wrapping needed
+ ('short-ascii', 'short-ascii'),
+ # Short Unicode — RFC 2047 QP encoded
+ ('short-unic\u00f4de', '=?utf-8?q?short-unic=C3=B4de?='),
+ # Long ASCII — wrapped at word boundary
+ (
+ 'Lorem ipsum dolor sit amet consectetur adipiscing elit '
+ 'sed do eiusmod tempor incididunt ut labore et dolore magna aliqua',
+ 'Lorem ipsum dolor sit amet consectetur adipiscing elit sed do\n'
+ ' eiusmod tempor incididunt ut labore et dolore magna aliqua',
+ ),
+ # Long Unicode — split across multiple encoded lines
+ (
+ 'Lorem \u00eepsum dolor sit amet consectetur adipiscing el\u00eet '
+ 'sed do eiusmod temp\u00f4r incididunt ut labore et dol\u00f4re magna aliqua',
+ '=?utf-8?q?Lorem_=C3=AEpsum_dolor_sit_amet_consectetur_adipiscin?=\n'
+ ' =?utf-8?q?g_el=C3=AEt_sed_do_eiusmod_temp=C3=B4r_incididunt_ut_labore_et?=\n'
+ ' =?utf-8?q?_dol=C3=B4re_magna_aliqua?=',
+ ),
+ # Exactly 75 chars — boundary condition
+ (
+ 'Lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiu',
+ 'Lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiu',
+ ),
+ # Unicode on escape boundary
+ (
+ 'Lorem ipsum dolor sit amet consectetur adipiscin el\u00eet',
+ '=?utf-8?q?Lorem_ipsum_dolor_sit_amet_consectetur_adipiscin_el?=\n'
+ ' =?utf-8?q?=C3=AEt?=',
+ ),
+ # Unicode 1 char too long
+ (
+ 'Lorem ipsum dolor sit amet consectetur adipi el\u00eet',
+ '=?utf-8?q?Lorem_ipsum_dolor_sit_amet_consectetur_adipi_el=C3=AE?=\n'
+ ' =?utf-8?q?t?=',
+ ),
+ ],
+ )
def test_non_address_header(self, hval: str, verify: str) -> None:
wrapped = wrap_header(('X-Header', hval))
assert wrapped.decode() == f'X-Header: {verify}'
- @pytest.mark.parametrize('hval,verify', [
- # Single address
- ('foo@example.com', 'foo@example.com'),
- # Two addresses
- ('foo@example.com, bar@example.com',
- 'foo@example.com, bar@example.com'),
- # Mixed plain + display name
- ('foo@example.com, Foo Bar <bar@example.com>',
- 'foo@example.com, Foo Bar <bar@example.com>'),
- # Mixed Unicode — non-ASCII name gets QP encoded
- ('foo@example.com, Foo Bar <bar@example.com>, F\u00f4o Baz <baz@example.com>',
- 'foo@example.com, Foo Bar <bar@example.com>, \n'
- ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>'),
- # Complex with quoted specials
- ('foo@example.com, Foo Bar <bar@example.com>, '
- 'F\u00f4o Baz <baz@example.com>, "Quux, Foo" <quux@example.com>',
- 'foo@example.com, Foo Bar <bar@example.com>, \n'
- ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>, '
- '"Quux, Foo" <quux@example.com>'),
- # Long local part forces line wrap
- ('01234567890123456789012345678901234567890123456789012345678901@example.org, '
- '\u00e4 <foo@example.org>',
- '01234567890123456789012345678901234567890123456789012345678901@example.org, \n'
- ' =?utf-8?q?=C3=A4?= <foo@example.org>'),
- # cpython#100900 — Unicode name with RFC 5322 specials
- ('foo@example.com, Foo Bar <bar@example.com>, '
- 'F\u00f4o Baz <baz@example.com>, "Qu\u00fbx, Foo" <quux@example.com>',
- 'foo@example.com, Foo Bar <bar@example.com>, \n'
- ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>, \n'
- ' =?utf-8?q?Qu=C3=BBx=2C_Foo?= <quux@example.com>'),
- ])
+ @pytest.mark.parametrize(
+ 'hval,verify',
+ [
+ # Single address
+ ('foo@example.com', 'foo@example.com'),
+ # Two addresses
+ ('foo@example.com, bar@example.com', 'foo@example.com, bar@example.com'),
+ # Mixed plain + display name
+ (
+ 'foo@example.com, Foo Bar <bar@example.com>',
+ 'foo@example.com, Foo Bar <bar@example.com>',
+ ),
+ # Mixed Unicode — non-ASCII name gets QP encoded
+ (
+ 'foo@example.com, Foo Bar <bar@example.com>, F\u00f4o Baz <baz@example.com>',
+ 'foo@example.com, Foo Bar <bar@example.com>, \n'
+ ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>',
+ ),
+ # Complex with quoted specials
+ (
+ 'foo@example.com, Foo Bar <bar@example.com>, '
+ 'F\u00f4o Baz <baz@example.com>, "Quux, Foo" <quux@example.com>',
+ 'foo@example.com, Foo Bar <bar@example.com>, \n'
+ ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>, '
+ '"Quux, Foo" <quux@example.com>',
+ ),
+ # Long local part forces line wrap
+ (
+ '01234567890123456789012345678901234567890123456789012345678901@example.org, '
+ '\u00e4 <foo@example.org>',
+ '01234567890123456789012345678901234567890123456789012345678901@example.org, \n'
+ ' =?utf-8?q?=C3=A4?= <foo@example.org>',
+ ),
+ # cpython#100900 — Unicode name with RFC 5322 specials
+ (
+ 'foo@example.com, Foo Bar <bar@example.com>, '
+ 'F\u00f4o Baz <baz@example.com>, "Qu\u00fbx, Foo" <quux@example.com>',
+ 'foo@example.com, Foo Bar <bar@example.com>, \n'
+ ' =?utf-8?q?F=C3=B4o_Baz?= <baz@example.com>, \n'
+ ' =?utf-8?q?Qu=C3=BBx=2C_Foo?= <quux@example.com>',
+ ),
+ ],
+ )
def test_address_header(self, hval: str, verify: str) -> None:
wrapped = wrap_header(('To', hval))
assert wrapped.decode() == f'To: {verify}'
- @pytest.mark.parametrize('hval,verify', [
- # Short message-id
- ('<20240319-short-message-id@example.com>',
- '<20240319-short-message-id@example.com>'),
- # Long message-id — unbreakable, stays on one line
- ('<20240319-very-long-message-id-that-spans-multiple-lines-for-sure'
- '-because-longer-than-75-characters-abcde123456@longdomain.example.com>',
- '<20240319-very-long-message-id-that-spans-multiple-lines-for-sure'
- '-because-longer-than-75-characters-abcde123456@longdomain.example.com>'),
- ])
+ @pytest.mark.parametrize(
+ 'hval,verify',
+ [
+ # Short message-id
+ (
+ '<20240319-short-message-id@example.com>',
+ '<20240319-short-message-id@example.com>',
+ ),
+ # Long message-id — unbreakable, stays on one line
+ (
+ '<20240319-very-long-message-id-that-spans-multiple-lines-for-sure'
+ '-because-longer-than-75-characters-abcde123456@longdomain.example.com>',
+ '<20240319-very-long-message-id-that-spans-multiple-lines-for-sure'
+ '-because-longer-than-75-characters-abcde123456@longdomain.example.com>',
+ ),
+ ],
+ )
def test_message_id(self, hval: str, verify: str) -> None:
wrapped = wrap_header(('Message-ID', hval))
assert wrapped.decode() == f'Message-ID: {verify}'
@@ -203,8 +267,9 @@ class TestMinimizeThread:
def test_custom_keep_headers(self, make_msg: MsgFactory) -> None:
"""Callers can override which headers to keep."""
- msg = make_msg(subject='Test', body='Hello.\n',
- date='Mon, 01 Jan 2024 00:00:00 +0000')
+ msg = make_msg(
+ subject='Test', body='Hello.\n', date='Mon, 01 Jan 2024 00:00:00 +0000'
+ )
result = minimize_thread([msg], keep_headers=('Subject',))
assert len(result) == 1
assert result[0]['Subject'] == 'Test'
@@ -329,15 +394,11 @@ class TestMinimizeThread:
assert 'Second paragraph' not in text
assert 'My reply here.' in text
- def test_reduce_quote_context_short_quote_untouched(self, make_msg: MsgFactory) -> None:
+ def test_reduce_quote_context_short_quote_untouched(
+ self, make_msg: MsgFactory
+ ) -> None:
"""Quotes with 5 or fewer skippable lines are left alone."""
- body = (
- '> Line one.\n'
- '> Line two.\n'
- '>\n'
- '> Last para.\n'
- 'Reply.\n'
- )
+ body = '> Line one.\n> Line two.\n>\n> Last para.\nReply.\n'
msg = make_msg(body=body)
result = minimize_thread([msg], reduce_quote_context=True)
text = result[0].get_payload()
diff --git a/tests/test_mbox.py b/tests/test_mbox.py
index 9bc2271..7269192 100644
--- a/tests/test_mbox.py
+++ b/tests/test_mbox.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for liblore.mbox."""
+
from __future__ import annotations
import textwrap
@@ -176,6 +177,7 @@ class TestSplitMboxAsBytes:
def test_roundtrip_with_split_mbox(self, sample_mbox: bytes) -> None:
"""split_mbox_as_bytes + parse_message should match split_mbox."""
from liblore.utils import parse_message
+
raw_chunks = split_mbox_as_bytes(sample_mbox)
parsed = split_mbox(sample_mbox)
assert len(raw_chunks) == len(parsed)
@@ -265,6 +267,7 @@ class TestSplitAndDedupeAsBytes:
def test_roundtrip_with_split_and_dedupe(self, sample_mbox: bytes) -> None:
"""Parsing the bytes output should match split_and_dedupe."""
from liblore.utils import parse_message
+
doubled = sample_mbox + sample_mbox
chunks = split_and_dedupe_as_bytes(doubled)
msgs = split_and_dedupe(doubled)
diff --git a/tests/test_message.py b/tests/test_message.py
index 947e112..77c4cd4 100644
--- a/tests/test_message.py
+++ b/tests/test_message.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for liblore.message."""
+
from __future__ import annotations
from email.message import EmailMessage
diff --git a/tests/test_node.py b/tests/test_node.py
index 54d76d8..8f4ccac 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for liblore.node (LoreNode)."""
+
from __future__ import annotations
import gzip
@@ -18,6 +19,7 @@ from liblore.node import LoreNode
# Session management
# =====================================================================
+
class TestSessionManagement:
def test_creates_session(self) -> None:
node = LoreNode()
@@ -115,6 +117,7 @@ class TestSessionManagement:
# get_mbox_by_msgid / get_mbox_by_query
# =====================================================================
+
class TestGetMboxByMsgid:
def test_returns_raw_bytes(self, sample_mbox: bytes) -> None:
node = LoreNode('https://lore.kernel.org/all')
@@ -243,6 +246,7 @@ class TestGetMboxByQuery:
# get_thread_by_msgid
# =====================================================================
+
class TestGetThreadByMsgid:
def test_full_thread(self, sample_mbox: bytes) -> None:
node = LoreNode('https://lore.kernel.org/all')
@@ -315,6 +319,7 @@ class TestGetThreadByMsgid:
# get_thread_updates_since
# =====================================================================
+
class TestGetThreadUpdatesSince:
def test_returns_messages(self, sample_mbox: bytes) -> None:
node = LoreNode('https://lore.kernel.org/all')
@@ -370,7 +375,9 @@ class TestGetThreadUpdatesSince:
since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
msgs = node.get_thread_updates_since(
- 'first@example.com', since, sort=True,
+ 'first@example.com',
+ since,
+ sort=True,
)
assert len(msgs) >= 1
@@ -391,6 +398,7 @@ class TestGetThreadUpdatesSince:
# get_thread_by_query
# =====================================================================
+
class TestGetThreadByQuery:
def test_posts_query(self, sample_mbox: bytes) -> None:
node = LoreNode('https://lore.kernel.org/all')
@@ -423,6 +431,7 @@ class TestGetThreadByQuery:
# get_message_by_msgid
# =====================================================================
+
class TestGetMessageByMsgid:
def test_fetches_raw(self) -> None:
node = LoreNode('https://lore.kernel.org/all')
@@ -451,6 +460,7 @@ class TestGetMessageByMsgid:
# batch_get_thread_by_msgid
# =====================================================================
+
class TestBatchGetThreadByMsgid:
def test_returns_ordered_results(self) -> None:
node = LoreNode()
@@ -482,11 +492,17 @@ class TestBatchGetThreadByMsgid:
with patch('liblore.node.time.sleep'):
node.batch_get_thread_by_msgid(
- ['a@x'], strict=False, sort=True, since='20240101',
+ ['a@x'],
+ strict=False,
+ sort=True,
+ since='20240101',
)
node.get_thread_by_msgid.assert_called_once_with(
- 'a@x', strict=False, sort=True, since='20240101',
+ 'a@x',
+ strict=False,
+ sort=True,
+ since='20240101',
)
def test_sleep_count_matches_gaps(self) -> None:
@@ -514,6 +530,7 @@ class TestBatchGetThreadByMsgid:
# batch_get_thread_by_query
# =====================================================================
+
class TestBatchGetThreadByQuery:
def test_returns_ordered_results(self) -> None:
node = LoreNode()
@@ -564,6 +581,7 @@ class TestBatchGetThreadByQuery:
# validate
# =====================================================================
+
class TestValidate:
def test_valid_url(self) -> None:
node = LoreNode('https://lore.kernel.org/lkml')
@@ -603,6 +621,7 @@ class TestValidate:
# URL fallback
# =====================================================================
+
class TestFallback:
"""Tests for the fallback_urls feature."""
@@ -807,6 +826,7 @@ class TestFallback:
def test_invalid_fallback_url_no_scheme(self) -> None:
"""Fallback URL without scheme raises LibloreError."""
from liblore import LibloreError
+
with pytest.raises(LibloreError, match='Invalid fallback URL'):
LoreNode(
'https://lore.kernel.org/all',
@@ -816,6 +836,7 @@ class TestFallback:
def test_invalid_fallback_url_with_path(self) -> None:
"""Fallback URL with a path component raises LibloreError."""
from liblore import LibloreError
+
with pytest.raises(LibloreError, match='must be a scheme://host origin'):
LoreNode(
'https://lore.kernel.org/all',
@@ -863,6 +884,7 @@ class TestFallback:
# Origin probing
# =====================================================================
+
class TestProbeOrigins:
"""Tests for the probe_origins() fastest-mirror feature."""
@@ -998,7 +1020,8 @@ class TestProbeOrigins:
assert h['User-Agent'] == 'myapp/1.0'
def test_auto_probe_triggers_on_first_request(
- self, sample_mbox: bytes,
+ self,
+ sample_mbox: bytes,
) -> None:
"""With auto_probe=True, first _request() triggers probe."""
node = LoreNode(
@@ -1038,6 +1061,7 @@ class TestProbeOrigins:
node.set_requests_session(mock_session)
probe_count = 0
+
def fake_head(url: str, **kwargs: object) -> MagicMock:
nonlocal probe_count
probe_count += 1
@@ -1105,10 +1129,12 @@ class TestProbeOrigins:
# Backdate cache file to force expiry
import glob as glob_mod
+
for f in glob_mod.glob(os.path.join(cache_dir, '*.lore.cache')):
os.utime(f, (0, 0))
probe_called = False
+
def fake_head_2(url: str, **kwargs: object) -> MagicMock:
nonlocal probe_called
probe_called = True
@@ -1123,7 +1149,8 @@ class TestProbeOrigins:
assert probe_called
def test_probe_cache_ignored_when_origins_change(
- self, tmp_path: object,
+ self,
+ tmp_path: object,
) -> None:
"""Cache is ignored when the set of origins differs."""
cache_dir = str(tmp_path)
@@ -1149,6 +1176,7 @@ class TestProbeOrigins:
)
probe_called = False
+
def fake_head_2(url: str, **kwargs: object) -> MagicMock:
nonlocal probe_called
probe_called = True
@@ -1207,6 +1235,7 @@ class TestProbeOrigins:
# Git config integration
# =====================================================================
+
class TestFromGitConfig:
"""Tests for LoreNode.from_git_config() with legacy [lore] section.
@@ -1226,8 +1255,10 @@ class TestFromGitConfig:
'probettl': '7200',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config()
assert node._all_origins == [
@@ -1246,8 +1277,10 @@ class TestFromGitConfig:
'autoprobe': 'true',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config(
fallback_urls=['https://explicit.example.com'],
auto_probe=False,
@@ -1261,16 +1294,20 @@ class TestFromGitConfig:
def test_git_not_installed(self) -> None:
"""Works fine when both config helpers return empty."""
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value={}):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value={}),
+ ):
node = LoreNode.from_git_config()
assert node._all_origins == ['https://lore.kernel.org']
def test_no_config_keys(self) -> None:
"""Works fine when no lore.* keys exist in git config."""
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value={}):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value={}),
+ ):
node = LoreNode.from_git_config()
assert node._all_origins == ['https://lore.kernel.org']
@@ -1280,16 +1317,20 @@ class TestFromGitConfig:
"""Non-numeric lore.probetimeout is silently ignored."""
gitcfg: dict[str, str | list[str]] = {'probetimeout': 'notanumber'}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config()
assert node._probe_timeout == 5.0 # default
def test_custom_url_passed_through(self) -> None:
"""The url argument is forwarded to __init__."""
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value={}):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value={}),
+ ):
node = LoreNode.from_git_config(
url='https://my-inbox.example.com/lists',
)
@@ -1302,8 +1343,10 @@ class TestFromGitConfig:
'useragentplus': '550e8400-e29b-41d4',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config()
assert node._user_agent_plus == '550e8400-e29b-41d4'
@@ -1317,8 +1360,10 @@ class TestFromGitConfig:
'useragentplus': 'from-git-config',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config()
node.set_user_agent('myapp', '1.0', plus='explicit')
@@ -1336,8 +1381,10 @@ class TestFromGitConfig:
'useragentplus': 'myuuid',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=gitcfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=gitcfg),
+ ):
node = LoreNode.from_git_config()
node.set_user_agent('bugspray', '0.3')
@@ -1426,7 +1473,9 @@ class TestGetSubsectionConfig:
)
with patch('liblore.node.subprocess.run', return_value=mock_result):
cfg = _get_subsection_config(
- 'liblore', 'https://lore.kernel.org', multivals=['fallback'],
+ 'liblore',
+ 'https://lore.kernel.org',
+ multivals=['fallback'],
)
assert cfg == {
@@ -1444,12 +1493,12 @@ class TestGetSubsectionConfig:
mock_result = MagicMock()
mock_result.returncode = 0
- mock_result.stdout = (
- 'liblore.https://subspace.kernel.org.fallback\nhttps://mirror.example.com\x00'
- )
+ mock_result.stdout = 'liblore.https://subspace.kernel.org.fallback\nhttps://mirror.example.com\x00'
with patch('liblore.node.subprocess.run', return_value=mock_result):
cfg = _get_subsection_config(
- 'liblore', 'https://subspace.kernel.org', multivals=['fallback'],
+ 'liblore',
+ 'https://subspace.kernel.org',
+ multivals=['fallback'],
)
assert cfg == {
@@ -1465,7 +1514,8 @@ class TestGetSubsectionConfig:
mock_result.stdout = ''
with patch('liblore.node.subprocess.run', return_value=mock_result):
cfg = _get_subsection_config(
- 'liblore', 'https://nonexistent.example.com',
+ 'liblore',
+ 'https://nonexistent.example.com',
)
assert cfg == {}
@@ -1479,7 +1529,8 @@ class TestGetSubsectionConfig:
side_effect=FileNotFoundError('git not found'),
):
cfg = _get_subsection_config(
- 'liblore', 'https://lore.kernel.org',
+ 'liblore',
+ 'https://lore.kernel.org',
)
assert cfg == {}
@@ -1496,8 +1547,10 @@ class TestFromGitConfigSubsections:
'useragentplus': 'subsection-uuid',
}
- with patch('liblore.node._get_subsection_config', return_value=subsection_cfg), \
- patch('liblore.node._get_config_from_git', return_value={}) as mock_legacy:
+ with (
+ patch('liblore.node._get_subsection_config', return_value=subsection_cfg),
+ patch('liblore.node._get_config_from_git', return_value={}) as mock_legacy,
+ ):
node = LoreNode.from_git_config()
assert node._all_origins == [
@@ -1516,8 +1569,10 @@ class TestFromGitConfigSubsections:
'autoprobe': 'true',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=legacy_cfg):
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch('liblore.node._get_config_from_git', return_value=legacy_cfg),
+ ):
node = LoreNode.from_git_config()
assert node._all_origins == [
@@ -1533,8 +1588,12 @@ class TestFromGitConfigSubsections:
'autoprobe': 'true',
}
- with patch('liblore.node._get_subsection_config', return_value={}), \
- patch('liblore.node._get_config_from_git', return_value=legacy_cfg) as mock_legacy:
+ with (
+ patch('liblore.node._get_subsection_config', return_value={}),
+ patch(
+ 'liblore.node._get_config_from_git', return_value=legacy_cfg
+ ) as mock_legacy,
+ ):
node = LoreNode.from_git_config(
url='https://subspace.kernel.org/_lists/helpdesk',
)
@@ -1552,8 +1611,10 @@ class TestFromGitConfigSubsections:
'useragentplus': 'subspace-token',
}
- with patch('liblore.node._get_subsection_config', return_value=subsection_cfg), \
- patch('liblore.node._get_config_from_git') as mock_legacy:
+ with (
+ patch('liblore.node._get_subsection_config', return_value=subsection_cfg),
+ patch('liblore.node._get_config_from_git') as mock_legacy,
+ ):
node = LoreNode.from_git_config(
url='https://subspace.kernel.org/_lists/helpdesk',
)
@@ -1576,8 +1637,12 @@ class TestFromGitConfigSubsections:
'useragentplus': 'old-uuid',
}
- with patch('liblore.node._get_subsection_config', return_value=subsection_cfg), \
- patch('liblore.node._get_config_from_git', return_value=legacy_cfg) as mock_legacy:
+ with (
+ patch('liblore.node._get_subsection_config', return_value=subsection_cfg),
+ patch(
+ 'liblore.node._get_config_from_git', return_value=legacy_cfg
+ ) as mock_legacy,
+ ):
node = LoreNode.from_git_config()
# Subsection wins
@@ -1613,6 +1678,7 @@ class TestFromGitConfigSubsections:
# Public API: request()
# =====================================================================
+
class TestRequest:
"""Tests for the public request() method."""
@@ -1670,7 +1736,8 @@ class TestRequest:
node.set_requests_session(mock_session)
node.request(
- 'GET', 'https://lore.kernel.org/manifest.js.gz',
+ 'GET',
+ 'https://lore.kernel.org/manifest.js.gz',
timeout=30,
)
_, kwargs = mock_session.get.call_args
@@ -1681,6 +1748,7 @@ class TestRequest:
# Public API: user_agent_plus property
# =====================================================================
+
class TestUserAgentPlusProperty:
"""Tests for the user_agent_plus read-only property."""
@@ -1710,6 +1778,7 @@ class TestUserAgentPlusProperty:
# Public API: origins property
# =====================================================================
+
class TestOriginsProperty:
"""Tests for the origins read-only property."""
@@ -1763,6 +1832,7 @@ class TestOriginsProperty:
# Public API: canonical_origin property
# =====================================================================
+
class TestCanonicalOriginProperty:
"""Tests for the canonical_origin read-only property."""
diff --git a/tests/test_thread.py b/tests/test_thread.py
index 62a9a1e..d835373 100644
--- a/tests/test_thread.py
+++ b/tests/test_thread.py
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: GPL-2.0-or-later
"""Tests for liblore.thread."""
+
from __future__ import annotations
from conftest import MsgFactory
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 05/14] Add pyright strict checks to CI
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (3 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 04/14] Add ruff format check to CI Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 06/14] Replace HTTP session mocks with responses Tamir Duberstein
` (9 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Configure pyright in strict mode and run it from the b4 CI checker.
Allow private-usage checks to be disabled only for tests via a pyright
execution environment override, and silence missing type stubs for the
optional authheaders dependency.
Add a small number of casts in tests to satisfy pyright. These casts
should be removed in a follow-up commit by tightening the mocked types
further.
---
ci.sh | 1 +
pyproject.toml | 11 ++++++++---
src/liblore/node.py | 2 +-
tests/test_node.py | 11 ++++++++---
4 files changed, 18 insertions(+), 7 deletions(-)
diff --git a/ci.sh b/ci.sh
index ec0baf8..4b07fa2 100755
--- a/ci.sh
+++ b/ci.sh
@@ -5,4 +5,5 @@ set -eu
uv run ruff format --check
uv run ruff check
uv run mypy .
+uv run pyright
uv run pytest --durations=0
diff --git a/pyproject.toml b/pyproject.toml
index 9bfadbe..7a96d23 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,6 +39,7 @@ Repository = "https://git.kernel.org/pub/scm/utils/liblore/liblore.git"
dev = [
"build",
"mypy",
+ "pyright",
"pytest",
"pytest-asyncio",
"ruff",
@@ -53,9 +54,13 @@ asyncio_default_fixture_loop_scope = "function"
[tool.mypy]
strict = true
-[[tool.mypy.overrides]]
-module = "authheaders"
-ignore_missing_imports = true
+[tool.pyright]
+typeCheckingMode = "strict"
+
+executionEnvironments = [
+ # We're testing private APIs quite a bit.
+ { root = "tests", reportPrivateUsage = false },
+]
[tool.ruff.lint]
extend-select = ["I"]
diff --git a/src/liblore/node.py b/src/liblore/node.py
index 503428b..095ba4f 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -195,7 +195,7 @@ class LoreNode:
self._authheaders: types.ModuleType | None = None
if add_auth_headers:
try:
- import authheaders
+ import authheaders # type: ignore[import-untyped]
self._authheaders = authheaders
except ImportError:
diff --git a/tests/test_node.py b/tests/test_node.py
index 8f4ccac..af140db 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -7,6 +7,7 @@ import gzip
import os
from datetime import datetime, timezone
from email.message import EmailMessage
+from typing import cast
from unittest.mock import MagicMock, call, patch
import pytest
@@ -25,7 +26,9 @@ class TestSessionManagement:
node = LoreNode()
s = node._get_session()
assert s is not None
- assert 'liblore/' in s.headers['User-Agent']
+ user_agent = s.headers['User-Agent']
+ assert isinstance(user_agent, str)
+ assert 'liblore/' in user_agent
node.close()
def test_returns_same_session(self) -> None:
@@ -74,7 +77,9 @@ class TestSessionManagement:
def test_default_no_plus(self) -> None:
node = LoreNode()
s = node._get_session()
- assert '+' not in s.headers['User-Agent']
+ user_agent = s.headers['User-Agent']
+ assert isinstance(user_agent, str)
+ assert '+' not in user_agent
node.close()
def test_set_requests_session(self) -> None:
@@ -1008,7 +1013,7 @@ class TestProbeOrigins:
def fake_head(url: str, **kwargs: object) -> MagicMock:
headers = kwargs.get('headers', {})
assert isinstance(headers, dict)
- captured_headers.append(headers)
+ captured_headers.append(cast(dict[str, str], headers))
resp = MagicMock()
resp.status_code = 200
return resp
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 06/14] Replace HTTP session mocks with responses
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (4 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 05/14] Add pyright strict checks " Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 07/14] Add ty checks to CI Tamir Duberstein
` (8 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Convert networking tests from patched requests sessions to responses
mocks. Each test now registers requests on a local RequestsMock so
expected HTTP traffic is declared inline.
Add an autouse responses fixture to block real network access outside
explicit mock blocks.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
pyproject.toml | 3 +-
tests/conftest.py | 9 +-
| 45 +-
tests/test_cache.py | 207 +++---
tests/test_node.py | 1708 ++++++++++++++++++++++++--------------------
5 files changed, 1097 insertions(+), 875 deletions(-)
diff --git a/pyproject.toml b/pyproject.toml
index 7a96d23..a160edd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -42,6 +42,7 @@ dev = [
"pyright",
"pytest",
"pytest-asyncio",
+ "responses",
"ruff",
"types-requests",
]
@@ -63,7 +64,7 @@ executionEnvironments = [
]
[tool.ruff.lint]
-extend-select = ["I"]
+extend-select = ["ARG", "I"]
[tool.ruff.format]
quote-style = "single"
diff --git a/tests/conftest.py b/tests/conftest.py
index 28fedaf..ea52d95 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -6,13 +6,20 @@ from __future__ import annotations
import email.utils
import textwrap
from email.message import EmailMessage
-from typing import Protocol
+from typing import Iterator, Protocol
import pytest
+import responses as responses_
from liblore import emlpolicy
+@pytest.fixture(autouse=True)
+def _block_network() -> Iterator[None]: # pyright: ignore[reportUnusedFunction]
+ with responses_.RequestsMock():
+ yield
+
+
class MsgFactory(Protocol):
def __call__(
self,
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index 8cec624..f1c240a 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -7,9 +7,11 @@ import gzip
import sys
from email.message import EmailMessage
from types import ModuleType
+from typing import Iterator
from unittest.mock import MagicMock, patch
import pytest
+import responses
from liblore import LibloreError
from liblore.node import LoreNode
@@ -122,7 +124,7 @@ class TestAuthenticateMsgs:
class TestAuthInFetchMethods:
@pytest.fixture()
- def auth_node(self, sample_mbox: bytes) -> LoreNode:
+ def auth_node(self) -> Iterator[tuple[LoreNode, responses.RequestsMock]]:
fake = ModuleType('authheaders')
fake.authenticate_message = MagicMock( # type: ignore[attr-defined]
return_value='Authentication-Results: liblore; dkim=pass',
@@ -133,26 +135,41 @@ class TestAuthInFetchMethods:
self._fake = fake
self._patcher = patch.dict(sys.modules, {'authheaders': fake})
self._patcher.start()
-
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
- return node
+ with responses.RequestsMock() as rsps:
+ yield node, rsps
def teardown_method(self) -> None:
if hasattr(self, '_patcher'):
self._patcher.stop()
- def test_get_thread_by_msgid(self, auth_node: LoreNode) -> None:
- msgs = auth_node.get_thread_by_msgid('first@example.com')
+ def test_get_thread_by_msgid(
+ self,
+ auth_node: tuple[LoreNode, responses.RequestsMock],
+ sample_mbox: bytes,
+ ) -> None:
+ node, rsps = auth_node
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ msgs = node.get_thread_by_msgid('first@example.com')
for msg in msgs:
assert msg['Authentication-Results'] == 'liblore; dkim=pass'
- def test_get_thread_by_query(self, auth_node: LoreNode) -> None:
- msgs = auth_node.get_thread_by_query('test query')
+ def test_get_thread_by_query(
+ self,
+ auth_node: tuple[LoreNode, responses.RequestsMock],
+ sample_mbox: bytes,
+ ) -> None:
+ node, rsps = auth_node
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ msgs = node.get_thread_by_query('test query')
for msg in msgs:
assert msg['Authentication-Results'] == 'liblore; dkim=pass'
diff --git a/tests/test_cache.py b/tests/test_cache.py
index b822e7c..c5f2a3a 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -6,9 +6,9 @@ from __future__ import annotations
import gzip
import os
from pathlib import Path
-from unittest.mock import MagicMock
import pytest
+import responses
from liblore import LoreNode, RemoteError
@@ -121,109 +121,154 @@ class TestProperties:
class TestCachedMethods:
"""Integration tests: verify cache hit avoids network, cache miss hits network."""
- def _make_node(
- self, tmp_path: Path, sample_mbox: bytes
- ) -> tuple[LoreNode, MagicMock]:
+ def _make_node(self, tmp_path: Path, _sample_mbox: bytes) -> LoreNode:
node = LoreNode(
'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
)
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
- return node, mock_session
+ return node
def test_get_mbox_by_msgid_caches(self, tmp_path: Path, sample_mbox: bytes) -> None:
- node, mock_session = self._make_node(tmp_path, sample_mbox)
- result1 = node.get_mbox_by_msgid('test@example.com')
- result2 = node.get_mbox_by_msgid('test@example.com')
- assert result1 == result2 == sample_mbox
- # Network should only be hit once
- assert mock_session.get.call_count == 1
+ with responses.RequestsMock() as rsps:
+ node = self._make_node(tmp_path, sample_mbox)
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ result1 = node.get_mbox_by_msgid('test@example.com')
+ result2 = node.get_mbox_by_msgid('test@example.com')
+ assert result1 == result2 == sample_mbox
+ # Network should only be hit once
+ assert len(rsps.calls) == 1
def test_get_mbox_by_query_caches(self, tmp_path: Path, sample_mbox: bytes) -> None:
- node, mock_session = self._make_node(tmp_path, sample_mbox)
- result1 = node.get_mbox_by_query('test query')
- result2 = node.get_mbox_by_query('test query')
- assert result1 == result2 == sample_mbox
- assert mock_session.post.call_count == 1
+ with responses.RequestsMock() as rsps:
+ node = self._make_node(tmp_path, sample_mbox)
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ result1 = node.get_mbox_by_query('test query')
+ result2 = node.get_mbox_by_query('test query')
+ assert result1 == result2 == sample_mbox
+ assert len(rsps.calls) == 1
def test_get_mbox_by_query_full_threads_separate_key(
self, tmp_path: Path, sample_mbox: bytes
) -> None:
- node, mock_session = self._make_node(tmp_path, sample_mbox)
- node.get_mbox_by_query('test', full_threads=False)
- node.get_mbox_by_query('test', full_threads=True)
- # Different full_threads values = different cache keys = two network calls
- assert mock_session.post.call_count == 2
+ with responses.RequestsMock() as rsps:
+ node = self._make_node(tmp_path, sample_mbox)
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&t=1&q=test',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ node.get_mbox_by_query('test', full_threads=False)
+ node.get_mbox_by_query('test', full_threads=True)
+ # Different full_threads values = different cache keys = two network calls
+ assert len(rsps.calls) == 2
def test_get_message_by_msgid_caches(self, tmp_path: Path) -> None:
- node = LoreNode(
- 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = b'raw email bytes'
- mock_resp.raise_for_status = MagicMock()
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- result1 = node.get_message_by_msgid('test@example.com')
- result2 = node.get_message_by_msgid('test@example.com')
- assert result1 == result2 == b'raw email bytes'
- assert mock_session.get.call_count == 1
+ with responses.RequestsMock() as rsps:
+ node = LoreNode(
+ 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/raw',
+ body=b'raw email bytes',
+ status=200,
+ )
+
+ result1 = node.get_message_by_msgid('test@example.com')
+ result2 = node.get_message_by_msgid('test@example.com')
+ assert result1 == result2 == b'raw email bytes'
+ assert len(rsps.calls) == 1
def test_cache_expired_refetches(self, tmp_path: Path, sample_mbox: bytes) -> None:
- node, mock_session = self._make_node(tmp_path, sample_mbox)
- node._cache_ttl = 1
- node.get_mbox_by_msgid('test@example.com')
- assert mock_session.get.call_count == 1
- # Backdate the cache file
- for f in tmp_path.glob('*.lore.cache'):
- os.utime(f, (0, 0))
- node.get_mbox_by_msgid('test@example.com')
- assert mock_session.get.call_count == 2
+ with responses.RequestsMock() as rsps:
+ node = self._make_node(tmp_path, sample_mbox)
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ node._cache_ttl = 1
+ node.get_mbox_by_msgid('test@example.com')
+ assert len(rsps.calls) == 1
+ # Backdate the cache file
+ for f in tmp_path.glob('*.lore.cache'):
+ os.utime(f, (0, 0))
+ node.get_mbox_by_msgid('test@example.com')
+ assert len(rsps.calls) == 2
def test_no_cache_no_files(self, tmp_path: Path, sample_mbox: bytes) -> None:
"""When cache_dir is None, no files are written."""
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.get_mbox_by_msgid('test@example.com')
- # tmp_path should be empty since we didn't set cache_dir to it
- assert list(tmp_path.iterdir()) == []
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ node.get_mbox_by_msgid('test@example.com')
+ # tmp_path should be empty since we didn't set cache_dir to it
+ assert list(tmp_path.iterdir()) == []
def test_fetch_thread_since_not_cached(
self, tmp_path: Path, sample_mbox: bytes
) -> None:
"""_fetch_thread_since should NOT be cached."""
- node, mock_session = self._make_node(tmp_path, sample_mbox)
- node._fetch_thread_since('test@example.com', 'dt:20240101..')
- node._fetch_thread_since('test@example.com', 'dt:20240101..')
- # Both calls should hit the network
- assert mock_session.post.call_count == 2
+ with responses.RequestsMock() as rsps:
+ node = self._make_node(tmp_path, sample_mbox)
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/test%40example.com/?x=m&q=dt%3A20240101..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ node._fetch_thread_since('test@example.com', 'dt:20240101..')
+ node._fetch_thread_since('test@example.com', 'dt:20240101..')
+ # Both calls should hit the network
+ assert len(rsps.calls) == 2
def test_error_not_cached(self, tmp_path: Path) -> None:
"""Network errors should not be cached."""
- node = LoreNode(
- 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 404
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- with pytest.raises(RemoteError):
- node.get_mbox_by_msgid('bad@example.com')
- # No cache file should be written
- assert list(tmp_path.glob('*.lore.cache')) == []
+ with responses.RequestsMock() as rsps:
+ node = LoreNode(
+ 'https://lore.kernel.org/all', cache_dir=str(tmp_path), cache_ttl=60
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/bad%40example.com/t.mbox.gz',
+ status=404,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/bad%40example.com/',
+ status=404,
+ )
+
+ with pytest.raises(RemoteError):
+ node.get_mbox_by_msgid('bad@example.com')
+ # No cache file should be written
+ assert list(tmp_path.glob('*.lore.cache')) == []
diff --git a/tests/test_node.py b/tests/test_node.py
index af140db..fc94d9f 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -7,15 +7,22 @@ import gzip
import os
from datetime import datetime, timezone
from email.message import EmailMessage
-from typing import cast
from unittest.mock import MagicMock, call, patch
import pytest
import requests
+import responses
from liblore import RemoteError
from liblore.node import LoreNode
+
+def request_url(rsps: responses.RequestsMock, index: int) -> str:
+ url = rsps.calls[index].request.url
+ assert url is not None
+ return url
+
+
# =====================================================================
# Session management
# =====================================================================
@@ -125,345 +132,367 @@ class TestSessionManagement:
class TestGetMboxByMsgid:
def test_returns_raw_bytes(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- result = node.get_mbox_by_msgid('first@example.com')
- assert result == sample_mbox
+ result = node.get_mbox_by_msgid('first@example.com')
+ assert result == sample_mbox
def test_http_error(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 500
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40x.com/t.mbox.gz',
+ status=500,
+ )
- with pytest.raises(RemoteError, match='Server returned an error'):
- node.get_mbox_by_msgid('test@x.com')
+ with pytest.raises(RemoteError, match='Server returned an error'):
+ node.get_mbox_by_msgid('test@x.com')
def test_404_falls_back_to_head_redirect(self, sample_mbox: bytes) -> None:
- """On 404, try HEAD against the bare origin to discover the list path."""
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
-
- # First GET returns 404
- mock_404 = MagicMock()
- mock_404.status_code = 404
-
- # HEAD follows redirect and succeeds
- mock_head = MagicMock()
- mock_head.status_code = 200
- mock_head.url = 'https://lore.kernel.org/tools/test%40example.com/'
-
- # Second GET (to resolved URL) succeeds
- mock_200 = MagicMock()
- mock_200.status_code = 200
- mock_200.content = gzip.compress(sample_mbox)
-
- mock_session.get.side_effect = [mock_404, mock_200]
- mock_session.head.return_value = mock_head
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """On 404, try HEAD against the bare origin to discover the list path."""
+ node = LoreNode('https://lore.kernel.org/all')
+ # First GET returns 404
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ status=404,
+ )
+ # HEAD follows redirect and succeeds
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/test%40example.com/',
+ status=302,
+ headers={
+ 'Location': 'https://lore.kernel.org/tools/test%40example.com/'
+ },
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/tools/test%40example.com/',
+ status=200,
+ )
+ # Second GET (to resolved URL) succeeds
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/tools/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- # Verify the HEAD was sent to the bare origin
- mock_session.head.assert_called_once()
- head_url = mock_session.head.call_args[0][0]
- assert head_url == 'https://lore.kernel.org/test%40example.com/'
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ # Verify the HEAD was sent to the bare origin
+ assert request_url(rsps, 1) == 'https://lore.kernel.org/test%40example.com/'
def test_404_no_redirect_raises(self) -> None:
- """When HEAD also 404s (no redirect), raise RemoteError."""
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
-
- mock_404 = MagicMock()
- mock_404.status_code = 404
-
- mock_head_404 = MagicMock()
- mock_head_404.status_code = 404
-
- mock_session.get.return_value = mock_404
- mock_session.head.return_value = mock_head_404
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """When HEAD also 404s (no redirect), raise RemoteError."""
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/nonexistent%40example.com/t.mbox.gz',
+ status=404,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/nonexistent%40example.com/',
+ status=404,
+ )
- with pytest.raises(RemoteError, match='Server returned an error: 404'):
- node.get_mbox_by_msgid('nonexistent@example.com')
+ with pytest.raises(RemoteError, match='Server returned an error: 404'):
+ node.get_mbox_by_msgid('nonexistent@example.com')
class TestGetMboxByQuery:
def test_returns_raw_bytes(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- result = node.get_mbox_by_query('test query')
- assert result == sample_mbox
+ result = node.get_mbox_by_query('test query')
+ assert result == sample_mbox
def test_full_threads_adds_t_param(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&t=1&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- node.get_mbox_by_query('test query', full_threads=True)
- url = mock_session.post.call_args[0][0]
- assert '&t=1&' in url
+ node.get_mbox_by_query('test query', full_threads=True)
+ url = request_url(rsps, 0)
+ assert '&t=1&' in url
def test_no_full_threads_omits_t_param(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- node.get_mbox_by_query('test query')
- url = mock_session.post.call_args[0][0]
- assert 't=1' not in url
+ node.get_mbox_by_query('test query')
+ url = request_url(rsps, 0)
+ assert 't=1' not in url
def test_http_error(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 500
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- with pytest.raises(RemoteError, match='Server returned an error'):
- node.get_mbox_by_query('test')
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test',
+ status=500,
+ )
+ with pytest.raises(RemoteError, match='Server returned an error'):
+ node.get_mbox_by_query('test')
-# =====================================================================
-# get_thread_by_msgid
-# =====================================================================
+ # =====================================================================
+ # get_thread_by_msgid
+ # =====================================================================
class TestGetThreadByMsgid:
def test_full_thread(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- msgs = node.get_thread_by_msgid('first@example.com')
- assert len(msgs) >= 1
- # Without since, fetches full thread via GET /{msgid}/t.mbox.gz
- mock_session.get.assert_called_once()
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ msgs = node.get_thread_by_msgid('first@example.com')
+ assert len(msgs) >= 1
+ # Without since, fetches full thread via GET /{msgid}/t.mbox.gz
+ assert len(rsps.calls) == 1
def test_query_contains_msgid(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.get_thread_by_msgid('first@example.com')
- call_url = mock_session.get.call_args[0][0]
- assert 'first%40example.com' in call_url or 'first@example.com' in call_url
- assert call_url.endswith('/t.mbox.gz')
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ node.get_thread_by_msgid('first@example.com')
+ call_url = request_url(rsps, 0)
+ assert 'first%40example.com' in call_url or 'first@example.com' in call_url
+ assert call_url.endswith('/t.mbox.gz')
def test_since_uses_dt_prefix(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.get_thread_by_msgid('first@example.com', since='20240101')
- call_url = mock_session.post.call_args[0][0]
- assert 'dt%3A20240101' in call_url or 'dt:20240101' in call_url
- assert 'first%40example.com' in call_url or 'first@example.com' in call_url
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/first%40example.com/?x=m&q=dt%3A20240101..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ node.get_thread_by_msgid('first@example.com', since='20240101')
+ call_url = request_url(rsps, 0)
+ assert 'dt%3A20240101' in call_url or 'dt:20240101' in call_url
+ assert 'first%40example.com' in call_url or 'first@example.com' in call_url
def test_raises_on_empty(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(b'')
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/nonexistent%40x.com/t.mbox.gz',
+ body=gzip.compress(b''),
+ status=200,
+ )
- with pytest.raises(LookupError):
- node.get_thread_by_msgid('nonexistent@x.com')
+ with pytest.raises(LookupError):
+ node.get_thread_by_msgid('nonexistent@x.com')
def test_sort_parameter(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- msgs = node.get_thread_by_msgid('first@example.com', sort=True)
- assert len(msgs) >= 1
+ msgs = node.get_thread_by_msgid('first@example.com', sort=True)
+ assert len(msgs) >= 1
-
-# =====================================================================
-# get_thread_updates_since
-# =====================================================================
+ # =====================================================================
+ # get_thread_updates_since
+ # =====================================================================
class TestGetThreadUpdatesSince:
def test_returns_messages(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
- msgs = node.get_thread_updates_since('first@example.com', since)
- assert len(msgs) >= 1
- mock_session.post.assert_called_once()
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/first%40example.com/?x=m&q=rt%3A1705320000..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+ msgs = node.get_thread_updates_since('first@example.com', since)
+ assert len(msgs) >= 1
+ assert len(rsps.calls) == 1
def test_empty_returns_empty_list(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(b'')
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/first%40example.com/?x=m&q=rt%3A1705320000..',
+ body=gzip.compress(b''),
+ status=200,
+ )
- since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
- msgs = node.get_thread_updates_since('first@example.com', since)
- assert msgs == []
+ since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+ msgs = node.get_thread_updates_since('first@example.com', since)
+ assert msgs == []
def test_converts_datetime_to_rt_epoch(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- since = datetime(2024, 3, 15, 8, 30, 45, tzinfo=timezone.utc)
- epoch = int(since.timestamp()) # 1710491445
- node.get_thread_updates_since('first@example.com', since)
- call_url = mock_session.post.call_args[0][0]
- assert f'rt%3A{epoch}' in call_url or f'rt:{epoch}' in call_url
- assert 'first%40example.com' in call_url or 'first@example.com' in call_url
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ since = datetime(2024, 3, 15, 8, 30, 45, tzinfo=timezone.utc)
+ epoch = int(since.timestamp()) # 1710491445
+ rsps.add(
+ responses.POST,
+ f'https://lore.kernel.org/all/first%40example.com/?x=m&q=rt%3A{epoch}..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ node.get_thread_updates_since('first@example.com', since)
+ call_url = request_url(rsps, 0)
+ assert f'rt%3A{epoch}' in call_url or f'rt:{epoch}' in call_url
+ assert 'first%40example.com' in call_url or 'first@example.com' in call_url
def test_with_sort(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
- msgs = node.get_thread_updates_since(
- 'first@example.com',
- since,
- sort=True,
- )
- assert len(msgs) >= 1
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/first%40example.com/?x=m&q=rt%3A1705320000..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- def test_server_error_returns_empty_list(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 404
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+ msgs = node.get_thread_updates_since(
+ 'first@example.com',
+ since,
+ sort=True,
+ )
+ assert len(msgs) >= 1
- since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
- msgs = node.get_thread_updates_since('first@example.com', since)
- assert msgs == []
+ def test_server_error_returns_empty_list(self) -> None:
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/first%40example.com/?x=m&q=rt%3A1705320000..',
+ status=404,
+ )
+ since = datetime(2024, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+ msgs = node.get_thread_updates_since('first@example.com', since)
+ assert msgs == []
-# =====================================================================
-# get_thread_by_query
-# =====================================================================
+ # =====================================================================
+ # get_thread_by_query
+ # =====================================================================
class TestGetThreadByQuery:
def test_posts_query(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- msgs = node.get_thread_by_query('test query')
- assert len(msgs) == 2
- mock_session.post.assert_called_once()
+ msgs = node.get_thread_by_query('test query')
+ assert len(msgs) == 2
+ assert len(rsps.calls) == 1
def test_query_with_date_filter(self, sample_mbox: bytes) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.get_thread_by_query('test d:20240101..')
- call_url = mock_session.post.call_args[0][0]
- assert 'd%3A20240101' in call_url or 'd:20240101' in call_url
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/?x=m&q=test+d%3A20240101..',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ node.get_thread_by_query('test d:20240101..')
+ call_url = request_url(rsps, 0)
+ assert 'd%3A20240101' in call_url or 'd:20240101' in call_url
-# =====================================================================
-# get_message_by_msgid
-# =====================================================================
+ # =====================================================================
+ # get_message_by_msgid
+ # =====================================================================
class TestGetMessageByMsgid:
def test_fetches_raw(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = b'raw email bytes'
- mock_resp.raise_for_status = MagicMock()
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40x.com/raw',
+ body=b'raw email bytes',
+ status=200,
+ )
- result = node.get_message_by_msgid('test@x.com')
- assert result == b'raw email bytes'
+ result = node.get_message_by_msgid('test@x.com')
+ assert result == b'raw email bytes'
def test_raises_remote_error(self) -> None:
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_session.get.side_effect = Exception('connection refused')
- node.set_requests_session(mock_session)
-
- with pytest.raises(RemoteError):
- node.get_message_by_msgid('test@x.com')
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40x.com/raw',
+ body=requests.ConnectionError('connection refused'),
+ )
+ with pytest.raises(RemoteError):
+ node.get_message_by_msgid('test@x.com')
-# =====================================================================
-# batch_get_thread_by_msgid
-# =====================================================================
+ # =====================================================================
+ # batch_get_thread_by_msgid
+ # =====================================================================
class TestBatchGetThreadByMsgid:
@@ -589,244 +618,287 @@ class TestBatchGetThreadByQuery:
class TestValidate:
def test_valid_url(self) -> None:
- node = LoreNode('https://lore.kernel.org/lkml')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_session.head.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.validate()
- mock_session.head.assert_called_once_with(
- 'https://lore.kernel.org/lkml/_/text/help/'
- )
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://lore.kernel.org/lkml')
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/lkml/_/text/help/',
+ status=200,
+ )
- def test_not_public_inbox(self) -> None:
- node = LoreNode('https://example.com/not-pi')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 404
- mock_session.head.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- with pytest.raises(RemoteError, match='does not appear'):
node.validate()
+ assert request_url(rsps, 0) == 'https://lore.kernel.org/lkml/_/text/help/'
- def test_connection_error(self) -> None:
- node = LoreNode('https://unreachable.example.com')
- mock_session = MagicMock()
- mock_session.head.side_effect = Exception('connection refused')
- node.set_requests_session(mock_session)
+ def test_not_public_inbox(self) -> None:
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://example.com/not-pi')
+ rsps.add(
+ responses.HEAD,
+ 'https://example.com/not-pi/_/text/help/',
+ status=404,
+ )
- with pytest.raises(RemoteError, match='Failed to reach'):
- node.validate()
+ with pytest.raises(RemoteError, match='does not appear'):
+ node.validate()
+ def test_connection_error(self) -> None:
+ with responses.RequestsMock() as rsps:
+ node = LoreNode('https://unreachable.example.com')
+ rsps.add(
+ responses.HEAD,
+ 'https://unreachable.example.com/_/text/help/',
+ body=requests.ConnectionError('connection refused'),
+ )
-# =====================================================================
-# URL fallback
-# =====================================================================
+ with pytest.raises(RemoteError, match='Failed to reach'):
+ node.validate()
+
+ # =====================================================================
+ # URL fallback
+ # =====================================================================
class TestFallback:
"""Tests for the fallback_urls feature."""
def test_no_fallbacks_unchanged(self, sample_mbox: bytes) -> None:
- """Without fallback_urls, behavior is identical to before."""
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- assert mock_session.get.call_count == 1
- url = mock_session.get.call_args[0][0]
- assert url.startswith('https://lore.kernel.org/')
+ with responses.RequestsMock() as rsps:
+ """Without fallback_urls, behavior is identical to before."""
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ assert len(rsps.calls) == 1
+ url = request_url(rsps, 0)
+ assert url.startswith('https://lore.kernel.org/')
def test_fallback_on_connection_error(self, sample_mbox: bytes) -> None:
- """Primary raises ConnectionError, fallback succeeds."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.side_effect = [
- requests.ConnectionError('refused'),
- mock_resp,
- ]
- node.set_requests_session(mock_session)
-
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- assert mock_session.get.call_count == 2
- # First call goes to the fallback (tried first)
- first_url = mock_session.get.call_args_list[0][0][0]
- assert first_url.startswith('http://mirror.local/all/')
- # Second call goes to the canonical URL
- second_url = mock_session.get.call_args_list[1][0][0]
- assert second_url.startswith('https://lore.kernel.org/all/')
+ with responses.RequestsMock() as rsps:
+ """Primary raises ConnectionError, fallback succeeds."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/all/test%40example.com/t.mbox.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ assert len(rsps.calls) == 2
+ # First call goes to the fallback (tried first)
+ first_url = request_url(rsps, 0)
+ assert first_url.startswith('http://mirror.local/all/')
+ # Second call goes to the canonical URL
+ second_url = request_url(rsps, 1)
+ assert second_url.startswith('https://lore.kernel.org/all/')
def test_fallback_on_timeout(self, sample_mbox: bytes) -> None:
- """Primary raises Timeout, fallback succeeds."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://ams.lore.kernel.org'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.side_effect = [
- requests.Timeout('timed out'),
- mock_resp,
- ]
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """Primary raises Timeout, fallback succeeds."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://ams.lore.kernel.org'],
+ )
+ rsps.add(
+ responses.GET,
+ 'https://ams.lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=requests.Timeout('timed out'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- assert mock_session.get.call_count == 2
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ assert len(rsps.calls) == 2
def test_fallback_on_5xx(self, sample_mbox: bytes) -> None:
- """Primary returns 500, fallback returns 200."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_500 = MagicMock()
- mock_500.status_code = 500
- mock_200 = MagicMock()
- mock_200.status_code = 200
- mock_200.content = gzip.compress(sample_mbox)
- mock_session.get.side_effect = [mock_500, mock_200]
- node.set_requests_session(mock_session)
-
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- assert mock_session.get.call_count == 2
+ with responses.RequestsMock() as rsps:
+ """Primary returns 500, fallback returns 200."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/all/test%40example.com/t.mbox.gz',
+ status=500,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ assert len(rsps.calls) == 2
def test_no_fallback_on_4xx(self) -> None:
- """4xx is not retriable — fallback should NOT be tried."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 404
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """4xx is not retriable — fallback should NOT be tried."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/all/test%40example.com/t.mbox.gz',
+ status=404,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/test%40example.com/',
+ status=404,
+ )
- with pytest.raises(RemoteError, match='Server returned an error'):
- node.get_mbox_by_msgid('test@example.com')
- # Only 1 call — the fallback, which returned 404, no retry
- assert mock_session.get.call_count == 1
+ with pytest.raises(RemoteError, match='Server returned an error'):
+ node.get_mbox_by_msgid('test@example.com')
+ # Fallback returns 404, then redirect discovery also returns 404.
+ assert len(rsps.calls) == 2
def test_all_hosts_fail_connection(self) -> None:
- """All origins raise ConnectionError → RemoteError."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_session.get.side_effect = requests.ConnectionError('refused')
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """All origins raise ConnectionError → RemoteError."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/all/test%40example.com/t.mbox.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=requests.ConnectionError('refused'),
+ )
- with pytest.raises(RemoteError, match='All hosts failed'):
- node.get_mbox_by_msgid('test@example.com')
- assert mock_session.get.call_count == 2
+ with pytest.raises(RemoteError, match='All hosts failed'):
+ node.get_mbox_by_msgid('test@example.com')
+ assert len(rsps.calls) == 2
def test_all_hosts_fail_5xx(self) -> None:
- """All origins return 5xx → caller gets the error response."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 503
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- # get_mbox_by_msgid checks status_code and raises RemoteError
- with pytest.raises(RemoteError, match='Server returned an error'):
- node.get_mbox_by_msgid('test@example.com')
- assert mock_session.get.call_count == 2
+ with responses.RequestsMock() as rsps:
+ """All origins return 5xx → caller gets the error response."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/all/test%40example.com/t.mbox.gz',
+ status=503,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ status=503,
+ )
+
+ # get_mbox_by_msgid checks status_code and raises RemoteError
+ with pytest.raises(RemoteError, match='Server returned an error'):
+ node.get_mbox_by_msgid('test@example.com')
+ assert len(rsps.calls) == 2
def test_all_hosts_fail_no_raise(self) -> None:
- """_fetch_thread_since path: all fail, returns empty list."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 503
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """_fetch_thread_since path: all fail, returns empty list."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.POST,
+ 'http://mirror.local/all/test%40example.com/?x=m&q=dt%3A20240101..',
+ status=503,
+ )
+ rsps.add(
+ responses.POST,
+ 'https://lore.kernel.org/all/test%40example.com/?x=m&q=dt%3A20240101..',
+ status=503,
+ )
- result = node._fetch_thread_since('test@example.com', 'dt:20240101..')
- assert result == []
- assert mock_session.post.call_count == 2
+ result = node._fetch_thread_since('test@example.com', 'dt:20240101..')
+ assert result == []
+ assert len(rsps.calls) == 2
def test_url_rewriting_preserves_path(self, sample_mbox: bytes) -> None:
- """Verify full URL rewriting with scheme change."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mymirror.local'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- # First call (fallback) succeeds
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- node.get_mbox_by_msgid('test@example.com')
- url = mock_session.get.call_args_list[0][0][0]
- assert url.startswith('http://mymirror.local/all/')
- assert url.endswith('/t.mbox.gz')
+ with responses.RequestsMock() as rsps:
+ """Verify full URL rewriting with scheme change."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mymirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mymirror.local/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+
+ node.get_mbox_by_msgid('test@example.com')
+ url = request_url(rsps, 0)
+ assert url.startswith('http://mymirror.local/all/')
+ assert url.endswith('/t.mbox.gz')
def test_url_rewriting_post(self, sample_mbox: bytes) -> None:
- """Verify URL rewriting works for POST requests too."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://ams.lore.kernel.org'],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.post.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """Verify URL rewriting works for POST requests too."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://ams.lore.kernel.org'],
+ )
+ rsps.add(
+ responses.POST,
+ 'https://ams.lore.kernel.org/all/?x=m&q=test+query',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
- node.get_mbox_by_query('test query')
- url = mock_session.post.call_args[0][0]
- assert url.startswith('https://ams.lore.kernel.org/all/')
+ node.get_mbox_by_query('test query')
+ url = request_url(rsps, 0)
+ assert url.startswith('https://ams.lore.kernel.org/all/')
def test_validate_does_not_use_fallback(self) -> None:
- """validate() hits canonical URL only, ignoring fallbacks."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_session.head.side_effect = Exception('connection refused')
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """validate() hits canonical URL only, ignoring fallbacks."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/all/_/text/help/',
+ body=requests.ConnectionError('connection refused'),
+ )
- with pytest.raises(RemoteError, match='Failed to reach'):
- node.validate()
- # Only 1 call to the canonical URL — no fallback
- assert mock_session.head.call_count == 1
- url = mock_session.head.call_args[0][0]
- assert 'lore.kernel.org' in url
+ with pytest.raises(RemoteError, match='Failed to reach'):
+ node.validate()
+ # Only 1 call to the canonical URL — no fallback
+ assert len(rsps.calls) == 1
+ url = request_url(rsps, 0)
+ assert 'lore.kernel.org' in url
def test_invalid_fallback_url_no_scheme(self) -> None:
"""Fallback URL without scheme raises LibloreError."""
@@ -857,37 +929,43 @@ class TestFallback:
assert node.hostname == 'lore.kernel.org'
def test_multiple_fallbacks_tried_in_order(self, sample_mbox: bytes) -> None:
- """With 3 fallbacks, they are tried in the configured order."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=[
- 'http://mirror1.local',
- 'https://ams.lore.kernel.org',
- ],
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.side_effect = [
- requests.ConnectionError('refused'),
- requests.ConnectionError('refused'),
- mock_resp,
- ]
- node.set_requests_session(mock_session)
-
- result = node.get_mbox_by_msgid('test@example.com')
- assert result == sample_mbox
- assert mock_session.get.call_count == 3
- urls = [c[0][0] for c in mock_session.get.call_args_list]
- assert urls[0].startswith('http://mirror1.local/')
- assert urls[1].startswith('https://ams.lore.kernel.org/')
- assert urls[2].startswith('https://lore.kernel.org/')
+ with responses.RequestsMock() as rsps:
+ """With 3 fallbacks, they are tried in the configured order."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=[
+ 'http://mirror1.local',
+ 'https://ams.lore.kernel.org',
+ ],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror1.local/all/test%40example.com/t.mbox.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://ams.lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ result = node.get_mbox_by_msgid('test@example.com')
+ assert result == sample_mbox
+ assert len(rsps.calls) == 3
+ urls = [request_url(rsps, i) for i in range(len(rsps.calls))]
+ assert urls[0].startswith('http://mirror1.local/')
+ assert urls[1].startswith('https://ams.lore.kernel.org/')
+ assert urls[2].startswith('https://lore.kernel.org/')
-# =====================================================================
-# Origin probing
-# =====================================================================
+ # =====================================================================
+ # Origin probing
+ # =====================================================================
class TestProbeOrigins:
@@ -927,49 +1005,54 @@ class TestProbeOrigins:
assert node._all_origins == origins
def test_probe_unreachable_sorted_last(self) -> None:
- """Unreachable origins get inf elapsed and sort to the end."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://dead.example.com'],
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- if 'dead' in url:
- raise requests.ConnectionError('refused')
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """Unreachable origins get inf elapsed and sort to the end."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://dead.example.com'],
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://dead.example.com/manifest.js.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
results = node.probe_origins()
- assert len(results) == 2
- # canonical should be first (reachable), dead last
- assert results[0][0] == 'https://lore.kernel.org'
- assert results[0][1] < float('inf')
- assert results[1][0] == 'https://dead.example.com'
- assert results[1][1] == float('inf')
+ assert len(results) == 2
+ # canonical should be first (reachable), dead last
+ assert results[0][0] == 'https://lore.kernel.org'
+ assert results[0][1] < float('inf')
+ assert results[1][0] == 'https://dead.example.com'
+ assert results[1][1] == float('inf')
def test_probe_4xx_treated_as_unreachable(self) -> None:
- """Origins returning 4xx are treated as unreachable."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://nomanifest.example.com'],
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- if 'nomanifest' in url:
- resp.status_code = 404
- else:
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """Origins returning 4xx are treated as unreachable."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://nomanifest.example.com'],
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://nomanifest.example.com/manifest.js.gz',
+ status=404,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
results = node.probe_origins()
- assert results[0][0] == 'https://lore.kernel.org'
- assert results[1][1] == float('inf')
+ assert results[0][0] == 'https://lore.kernel.org'
+ assert results[1][1] == float('inf')
def test_probe_single_origin_noop(self) -> None:
"""With only one origin, probe is a no-op."""
@@ -980,265 +1063,321 @@ class TestProbeOrigins:
assert node._probe_done is True
def test_probe_uses_manifest_url(self) -> None:
- """Probe hits /manifest.js.gz on each origin."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- probed_urls: list[str] = []
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- probed_urls.append(url)
- resp = MagicMock()
- resp.status_code = 200
- return resp
-
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ with responses.RequestsMock() as rsps:
+ """Probe hits /manifest.js.gz on each origin."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ probed_urls: list[str] = []
+
+ def callback(
+ request: requests.PreparedRequest,
+ ) -> tuple[int, dict[str, str], str]:
+ assert request.url is not None
+ probed_urls.append(request.url)
+ return 200, {}, ''
+
+ rsps.add_callback(
+ responses.HEAD,
+ 'http://mirror.local/manifest.js.gz',
+ callback=callback,
+ )
+ rsps.add_callback(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ callback=callback,
+ )
node.probe_origins()
- assert len(probed_urls) == 2
- assert 'http://mirror.local/manifest.js.gz' in probed_urls
- assert 'https://lore.kernel.org/manifest.js.gz' in probed_urls
+ assert len(probed_urls) == 2
+ assert 'http://mirror.local/manifest.js.gz' in probed_urls
+ assert 'https://lore.kernel.org/manifest.js.gz' in probed_urls
def test_probe_sends_user_agent(self) -> None:
- """Probe requests include the configured User-Agent."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- node.set_user_agent('myapp', '1.0')
+ with responses.RequestsMock() as rsps:
+ """Probe requests include the configured User-Agent."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ node.set_user_agent('myapp', '1.0')
- captured_headers: list[dict[str, str]] = []
+ captured_headers: list[dict[str, str]] = []
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- headers = kwargs.get('headers', {})
- assert isinstance(headers, dict)
- captured_headers.append(cast(dict[str, str], headers))
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ def callback(
+ request: requests.PreparedRequest,
+ ) -> tuple[int, dict[str, str], str]:
+ captured_headers.append(dict(request.headers))
+ return 200, {}, ''
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add_callback(
+ responses.HEAD,
+ 'http://mirror.local/manifest.js.gz',
+ callback=callback,
+ )
+ rsps.add_callback(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ callback=callback,
+ )
node.probe_origins()
- for h in captured_headers:
- assert h['User-Agent'] == 'myapp/1.0'
+ for h in captured_headers:
+ assert h['User-Agent'] == 'myapp/1.0'
def test_auto_probe_triggers_on_first_request(
self,
sample_mbox: bytes,
) -> None:
- """With auto_probe=True, first _request() triggers probe."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- auto_probe=True,
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- resp.status_code = 200
- return resp
-
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ with responses.RequestsMock() as rsps:
+ """With auto_probe=True, first _request() triggers probe."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ auto_probe=True,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://fast.example.com/all/test%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
node.get_mbox_by_msgid('test@example.com')
- assert node._probe_done is True
+ assert node._probe_done is True
def test_auto_probe_only_once(self, sample_mbox: bytes) -> None:
- """auto_probe fires only on the first request, not subsequent ones."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- auto_probe=True,
- )
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_resp.content = gzip.compress(sample_mbox)
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
-
- probe_count = 0
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- nonlocal probe_count
- probe_count += 1
- resp = MagicMock()
- resp.status_code = 200
- return resp
-
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ with responses.RequestsMock() as rsps:
+ """auto_probe fires only on the first request, not subsequent ones."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ auto_probe=True,
+ )
+ probe_count = 0
+
+ def callback(
+ _request: requests.PreparedRequest,
+ ) -> tuple[int, dict[str, str], str]:
+ nonlocal probe_count
+ probe_count += 1
+ return 200, {}, ''
+
+ rsps.add_callback(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ callback=callback,
+ )
+ rsps.add_callback(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ callback=callback,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://fast.example.com/all/first%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
+ rsps.add(
+ responses.GET,
+ 'https://fast.example.com/all/second%40example.com/t.mbox.gz',
+ body=gzip.compress(sample_mbox),
+ status=200,
+ )
node.get_mbox_by_msgid('first@example.com')
first_probe_count = probe_count
node.get_mbox_by_msgid('second@example.com')
- # Second request should NOT trigger another probe
- assert probe_count == first_probe_count
+ # Second request should NOT trigger another probe
+ assert probe_count == first_probe_count
def test_probe_cache_write_and_read(self, tmp_path: object) -> None:
- """Probe results are cached and restored on next probe call."""
- cache_dir = str(tmp_path)
- node1 = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- cache_dir=cache_dir,
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """Probe results are cached and restored on next probe call."""
+ cache_dir = str(tmp_path)
+ node1 = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ cache_dir=cache_dir,
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
with patch('liblore.node.time.monotonic') as mock_mono:
# fast: 0.1s, canonical: 0.5s
mock_mono.side_effect = [0.0, 0.1, 0.0, 0.5]
node1.probe_origins()
- expected_order = node1._all_origins[:]
+ expected_order = node1._all_origins[:]
- # New node with same origins should get cached order
- node2 = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- cache_dir=cache_dir,
- )
- # Without patching requests.head — cache should be used
- node2.probe_origins()
- assert node2._all_origins == expected_order
+ # New node with same origins should get cached order
+ node2 = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ cache_dir=cache_dir,
+ )
+ # Without patching requests.head — cache should be used
+ node2.probe_origins()
+ assert node2._all_origins == expected_order
def test_probe_cache_expired(self, tmp_path: object) -> None:
- """Expired probe cache triggers a fresh probe."""
- cache_dir = str(tmp_path)
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- cache_dir=cache_dir,
- probe_ttl=10,
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """Expired probe cache triggers a fresh probe."""
+ cache_dir = str(tmp_path)
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ cache_dir=cache_dir,
+ probe_ttl=10,
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
node.probe_origins()
- # Backdate cache file to force expiry
- import glob as glob_mod
-
- for f in glob_mod.glob(os.path.join(cache_dir, '*.lore.cache')):
- os.utime(f, (0, 0))
+ # Backdate cache file to force expiry
+ import glob as glob_mod
- probe_called = False
+ for f in glob_mod.glob(os.path.join(cache_dir, '*.lore.cache')):
+ os.utime(f, (0, 0))
- def fake_head_2(url: str, **kwargs: object) -> MagicMock:
- nonlocal probe_called
- probe_called = True
- resp = MagicMock()
- resp.status_code = 200
- return resp
-
- node._probe_done = False
- with patch('liblore.node.requests.head', side_effect=fake_head_2):
+ node._probe_done = False
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
node.probe_origins()
-
- assert probe_called
+ assert len(rsps.calls) == 4
def test_probe_cache_ignored_when_origins_change(
self,
tmp_path: object,
) -> None:
- """Cache is ignored when the set of origins differs."""
- cache_dir = str(tmp_path)
- node1 = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- cache_dir=cache_dir,
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """Cache is ignored when the set of origins differs."""
+ cache_dir = str(tmp_path)
+ node1 = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ cache_dir=cache_dir,
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
node1.probe_origins()
- # New node with DIFFERENT fallbacks
- node2 = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://other.example.com'],
- cache_dir=cache_dir,
- )
-
- probe_called = False
-
- def fake_head_2(url: str, **kwargs: object) -> MagicMock:
- nonlocal probe_called
- probe_called = True
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ # New node with DIFFERENT fallbacks
+ node2 = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://other.example.com'],
+ cache_dir=cache_dir,
+ )
- with patch('liblore.node.requests.head', side_effect=fake_head_2):
+ rsps.add(
+ responses.HEAD,
+ 'https://other.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
node2.probe_origins()
- # Different origins → cache miss → fresh probe
- assert probe_called
+ # Different origins → cache miss → fresh probe
+ assert len(rsps.calls) == 4
def test_probe_nocache_skips_cache(self, tmp_path: object) -> None:
- """nocache=True forces a live probe even when cache is fresh."""
- cache_dir = str(tmp_path)
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['https://fast.example.com'],
- cache_dir=cache_dir,
- )
-
- def fake_head(url: str, **kwargs: object) -> MagicMock:
- resp = MagicMock()
- resp.status_code = 200
- return resp
+ with responses.RequestsMock() as rsps:
+ """nocache=True forces a live probe even when cache is fresh."""
+ cache_dir = str(tmp_path)
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['https://fast.example.com'],
+ cache_dir=cache_dir,
+ )
- # First probe — populates cache
- with patch('liblore.node.requests.head', side_effect=fake_head):
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
+ # First probe — populates cache
with patch('liblore.node.time.monotonic') as mock_mono:
mock_mono.side_effect = [0.0, 0.1, 0.0, 0.5]
node.probe_origins()
- # Second probe with nocache — should do a live probe, not read cache
- probe_called = False
-
- def fake_head_2(url: str, **kwargs: object) -> MagicMock:
- nonlocal probe_called
- probe_called = True
- resp = MagicMock()
- resp.status_code = 200
- return resp
-
- node._probe_done = False
- with patch('liblore.node.requests.head', side_effect=fake_head_2):
+ # Second probe with nocache — should do a live probe, not read cache
+ node._probe_done = False
+ rsps.add(
+ responses.HEAD,
+ 'https://fast.example.com/manifest.js.gz',
+ status=200,
+ )
+ rsps.add(
+ responses.HEAD,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
with patch('liblore.node.time.monotonic') as mock_mono:
- mock_mono.side_effect = [0.0, 0.2, 0.0, 0.3]
+ mock_mono.side_effect = [0.0, 1.0, 2.0, 3.0]
results = node.probe_origins(nocache=True)
- assert probe_called
- # Results should have real timing, not 0.0
- assert all(elapsed > 0.0 for _, elapsed in results)
-
+ assert len(rsps.calls) == 4
+ # Results should have real timing, not 0.0
+ assert all(elapsed > 0.0 for _, elapsed in results)
-# =====================================================================
-# Git config integration
-# =====================================================================
+ # =====================================================================
+ # Git config integration
+ # =====================================================================
class TestFromGitConfig:
@@ -1687,71 +1826,84 @@ class TestFromGitConfigSubsections:
class TestRequest:
"""Tests for the public request() method."""
- def test_delegates_to_private_request(self, sample_mbox: bytes) -> None:
- """request() delegates to _request() with raise_on_error=True."""
- node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ def test_delegates_to_private_request(self) -> None:
+ with responses.RequestsMock() as rsps:
+ """request() delegates to _request() with raise_on_error=True."""
+ node = LoreNode('https://lore.kernel.org/all')
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
- resp = node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
- assert resp.status_code == 200
+ resp = node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
+ assert resp.status_code == 200
- def test_failover_works(self, sample_mbox: bytes) -> None:
- """First origin fails, second succeeds."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_200 = MagicMock()
- mock_200.status_code = 200
- mock_session.get.side_effect = [
- requests.ConnectionError('refused'),
- mock_200,
- ]
- node.set_requests_session(mock_session)
+ def test_failover_works(self) -> None:
+ with responses.RequestsMock() as rsps:
+ """First origin fails, second succeeds."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/manifest.js.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/manifest.js.gz',
+ status=200,
+ )
- resp = node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
- assert resp.status_code == 200
- assert mock_session.get.call_count == 2
+ resp = node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
+ assert resp.status_code == 200
+ assert len(rsps.calls) == 2
def test_raises_remote_error_when_all_fail(self) -> None:
- """RemoteError raised when every origin fails."""
- node = LoreNode(
- 'https://lore.kernel.org/all',
- fallback_urls=['http://mirror.local'],
- )
- mock_session = MagicMock()
- mock_session.get.side_effect = requests.ConnectionError('refused')
- node.set_requests_session(mock_session)
+ with responses.RequestsMock() as rsps:
+ """RemoteError raised when every origin fails."""
+ node = LoreNode(
+ 'https://lore.kernel.org/all',
+ fallback_urls=['http://mirror.local'],
+ )
+ rsps.add(
+ responses.GET,
+ 'http://mirror.local/manifest.js.gz',
+ body=requests.ConnectionError('refused'),
+ )
+ rsps.add(
+ responses.GET,
+ 'https://lore.kernel.org/manifest.js.gz',
+ body=requests.ConnectionError('refused'),
+ )
- with pytest.raises(RemoteError, match='All hosts failed'):
- node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
+ with pytest.raises(RemoteError, match='All hosts failed'):
+ node.request('GET', 'https://lore.kernel.org/manifest.js.gz')
def test_kwargs_forwarded(self) -> None:
"""Extra kwargs (e.g. timeout) are passed through."""
node = LoreNode('https://lore.kernel.org/all')
- mock_session = MagicMock()
- mock_resp = MagicMock()
- mock_resp.status_code = 200
- mock_session.get.return_value = mock_resp
- node.set_requests_session(mock_session)
+ with patch.object(
+ node, '_request', return_value=MagicMock(status_code=200)
+ ) as mock_request:
+ node.request(
+ 'GET',
+ 'https://lore.kernel.org/manifest.js.gz',
+ timeout=30,
+ )
- node.request(
+ mock_request.assert_called_once_with(
'GET',
'https://lore.kernel.org/manifest.js.gz',
+ raise_on_error=True,
timeout=30,
)
- _, kwargs = mock_session.get.call_args
- assert kwargs['timeout'] == 30
-
-# =====================================================================
-# Public API: user_agent_plus property
-# =====================================================================
+ # =====================================================================
+ # Public API: user_agent_plus property
+ # =====================================================================
class TestUserAgentPlusProperty:
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 07/14] Add ty checks to CI
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (5 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 06/14] Replace HTTP session mocks with responses Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 08/14] Drop redundant read-only property test Tamir Duberstein
` (7 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Add ty to the development dependencies and enable all ty rules. Run ty
in the b4 CI check script so its diagnostics are surfaced with the other
checks.
Use `ty check --add-ignore` to suppress existing errors.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
ci.sh | 1 +
pyproject.toml | 4 ++++
src/liblore/node.py | 2 +-
| 10 +++++-----
tests/test_node.py | 30 +++++++++++++++---------------
5 files changed, 26 insertions(+), 21 deletions(-)
diff --git a/ci.sh b/ci.sh
index 4b07fa2..c97c853 100755
--- a/ci.sh
+++ b/ci.sh
@@ -4,6 +4,7 @@ set -eu
uv run ruff format --check
uv run ruff check
+uv run ty check
uv run mypy .
uv run pyright
uv run pytest --durations=0
diff --git a/pyproject.toml b/pyproject.toml
index a160edd..37c85a7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -44,6 +44,7 @@ dev = [
"pytest-asyncio",
"responses",
"ruff",
+ "ty",
"types-requests",
]
@@ -63,6 +64,9 @@ executionEnvironments = [
{ root = "tests", reportPrivateUsage = false },
]
+[tool.ty.rules]
+all = "error"
+
[tool.ruff.lint]
extend-select = ["ARG", "I"]
diff --git a/src/liblore/node.py b/src/liblore/node.py
index 095ba4f..1b0463e 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -319,7 +319,7 @@ class LoreNode:
except ValueError:
pass
- node = cls(url, **kwargs) # type: ignore[arg-type]
+ node = cls(url, **kwargs) # type: ignore[arg-type] # ty:ignore[invalid-argument-type]
val = gitcfg.get('useragentplus')
if isinstance(val, str) and val:
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index f1c240a..b51050c 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -29,7 +29,7 @@ class TestAuthHeadersImport:
def test_ok_when_authheaders_installed(self) -> None:
fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock() # type: ignore[attr-defined]
+ fake.authenticate_message = MagicMock() # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
with patch.dict(sys.modules, {'authheaders': fake}):
node = LoreNode(add_auth_headers=True)
assert node._authheaders is not None
@@ -56,7 +56,7 @@ class TestAuthenticateMsgs:
def test_adds_header_when_enabled(self) -> None:
fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined]
+ fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
return_value='Authentication-Results: liblore; dkim=pass header.d=example.com',
)
with patch.dict(sys.modules, {'authheaders': fake}):
@@ -82,7 +82,7 @@ class TestAuthenticateMsgs:
def test_skips_empty_result(self) -> None:
fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock(return_value='') # type: ignore[attr-defined]
+ fake.authenticate_message = MagicMock(return_value='') # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
with patch.dict(sys.modules, {'authheaders': fake}):
node = LoreNode(add_auth_headers=True)
msg = EmailMessage()
@@ -95,7 +95,7 @@ class TestAuthenticateMsgs:
def test_multiple_messages(self) -> None:
fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined]
+ fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
side_effect=[
'liblore; dkim=pass',
'Authentication-Results: liblore; dkim=fail',
@@ -126,7 +126,7 @@ class TestAuthInFetchMethods:
@pytest.fixture()
def auth_node(self) -> Iterator[tuple[LoreNode, responses.RequestsMock]]:
fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined]
+ fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
return_value='Authentication-Results: liblore; dkim=pass',
)
with patch.dict(sys.modules, {'authheaders': fake}):
diff --git a/tests/test_node.py b/tests/test_node.py
index fc94d9f..f4a3495 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -500,19 +500,19 @@ class TestBatchGetThreadByMsgid:
node = LoreNode()
thread_a = [EmailMessage()]
thread_b = [EmailMessage(), EmailMessage()]
- node.get_thread_by_msgid = MagicMock(side_effect=[thread_a, thread_b]) # type: ignore[method-assign]
+ node.get_thread_by_msgid = MagicMock(side_effect=[thread_a, thread_b]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid(['a@x', 'b@x'])
assert results == [thread_a, thread_b]
- assert node.get_thread_by_msgid.call_count == 2
+ assert node.get_thread_by_msgid.call_count == 2 # ty:ignore[unresolved-attribute]
mock_sleep.assert_called_once_with(0.1)
def test_no_sleep_for_single_msgid(self) -> None:
node = LoreNode()
thread = [EmailMessage()]
- node.get_thread_by_msgid = MagicMock(return_value=thread) # type: ignore[method-assign]
+ node.get_thread_by_msgid = MagicMock(return_value=thread) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid(['only@x'])
@@ -522,7 +522,7 @@ class TestBatchGetThreadByMsgid:
def test_passes_kwargs(self) -> None:
node = LoreNode()
- node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign]
+ node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep'):
node.batch_get_thread_by_msgid(
@@ -532,7 +532,7 @@ class TestBatchGetThreadByMsgid:
since='20240101',
)
- node.get_thread_by_msgid.assert_called_once_with(
+ node.get_thread_by_msgid.assert_called_once_with( # ty:ignore[unresolved-attribute]
'a@x',
strict=False,
sort=True,
@@ -541,7 +541,7 @@ class TestBatchGetThreadByMsgid:
def test_sleep_count_matches_gaps(self) -> None:
node = LoreNode()
- node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign]
+ node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
node.batch_get_thread_by_msgid(['a@x', 'b@x', 'c@x'])
@@ -550,14 +550,14 @@ class TestBatchGetThreadByMsgid:
def test_empty_list(self) -> None:
node = LoreNode()
- node.get_thread_by_msgid = MagicMock() # type: ignore[method-assign]
+ node.get_thread_by_msgid = MagicMock() # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid([])
assert results == []
mock_sleep.assert_not_called()
- node.get_thread_by_msgid.assert_not_called()
+ node.get_thread_by_msgid.assert_not_called() # ty:ignore[unresolved-attribute]
# =====================================================================
@@ -570,19 +570,19 @@ class TestBatchGetThreadByQuery:
node = LoreNode()
result_a = [EmailMessage()]
result_b = [EmailMessage(), EmailMessage()]
- node.get_thread_by_query = MagicMock(side_effect=[result_a, result_b]) # type: ignore[method-assign]
+ node.get_thread_by_query = MagicMock(side_effect=[result_a, result_b]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query(['q1', 'q2'])
assert results == [result_a, result_b]
- assert node.get_thread_by_query.call_count == 2
+ assert node.get_thread_by_query.call_count == 2 # ty:ignore[unresolved-attribute]
mock_sleep.assert_called_once_with(0.1)
def test_no_sleep_for_single_query(self) -> None:
node = LoreNode()
result = [EmailMessage()]
- node.get_thread_by_query = MagicMock(return_value=result) # type: ignore[method-assign]
+ node.get_thread_by_query = MagicMock(return_value=result) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query(['only_query'])
@@ -592,7 +592,7 @@ class TestBatchGetThreadByQuery:
def test_sleep_count_matches_gaps(self) -> None:
node = LoreNode()
- node.get_thread_by_query = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign]
+ node.get_thread_by_query = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
node.batch_get_thread_by_query(['q1', 'q2', 'q3', 'q4'])
@@ -601,14 +601,14 @@ class TestBatchGetThreadByQuery:
def test_empty_list(self) -> None:
node = LoreNode()
- node.get_thread_by_query = MagicMock() # type: ignore[method-assign]
+ node.get_thread_by_query = MagicMock() # type: ignore[method-assign] # ty:ignore[invalid-assignment]
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query([])
assert results == []
mock_sleep.assert_not_called()
- node.get_thread_by_query.assert_not_called()
+ node.get_thread_by_query.assert_not_called() # ty:ignore[unresolved-attribute]
# =====================================================================
@@ -1928,7 +1928,7 @@ class TestUserAgentPlusProperty:
"""Property has no setter — assignment raises AttributeError."""
node = LoreNode()
with pytest.raises(AttributeError):
- node.user_agent_plus = 'nope' # type: ignore[misc]
+ node.user_agent_plus = 'nope' # type: ignore[misc] # ty:ignore[invalid-assignment]
# =====================================================================
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 08/14] Drop redundant read-only property test
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (6 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 07/14] Add ty checks to CI Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 09/14] Type from_git_config keyword arguments Tamir Duberstein
` (6 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Remove the user_agent_plus assignment test. It's not clear to me what
the point of this test is, and doing this already produces type checking
errors.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
tests/test_node.py | 6 ------
1 file changed, 6 deletions(-)
diff --git a/tests/test_node.py b/tests/test_node.py
index f4a3495..4abb4f4 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -1924,12 +1924,6 @@ class TestUserAgentPlusProperty:
assert node.user_agent_plus == 'my-tracking-uuid'
- def test_read_only(self) -> None:
- """Property has no setter — assignment raises AttributeError."""
- node = LoreNode()
- with pytest.raises(AttributeError):
- node.user_agent_plus = 'nope' # type: ignore[misc] # ty:ignore[invalid-assignment]
-
# =====================================================================
# Public API: origins property
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 09/14] Type from_git_config keyword arguments
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (7 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 08/14] Drop redundant read-only property test Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 10/14] Add authheaders stub and typed callable Tamir Duberstein
` (5 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Define a TypedDict for LoreNode constructor keyword arguments and
annotate from_git_config with an unpacked kwargs signature.
This removes the constructor-call type suppressions while keeping
git-config defaults and caller overrides unchanged.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
pyproject.toml | 1 +
src/liblore/node.py | 18 ++++++++++++++++--
2 files changed, 17 insertions(+), 2 deletions(-)
diff --git a/pyproject.toml b/pyproject.toml
index 37c85a7..387757c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,6 +46,7 @@ dev = [
"ruff",
"ty",
"types-requests",
+ "typing-extensions",
]
[tool.pytest.ini_options]
diff --git a/src/liblore/node.py b/src/liblore/node.py
index 1b0463e..f9817c1 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -17,6 +17,7 @@ import types
import urllib.parse
from datetime import datetime, timezone
from email.message import EmailMessage
+from typing import TYPE_CHECKING, TypedDict
import requests
@@ -29,6 +30,19 @@ from liblore.utils import (
logger = logging.getLogger(__name__)
+if TYPE_CHECKING:
+ from typing_extensions import Unpack
+
+
+class _LoreNodeInitKwargs(TypedDict, total=False):
+ fallback_urls: list[str] | None
+ auto_probe: bool
+ probe_timeout: float
+ probe_ttl: int
+ add_auth_headers: bool
+ cache_dir: str | None
+ cache_ttl: int
+
def _get_config_from_git(
regexp: str,
@@ -212,7 +226,7 @@ class LoreNode:
def from_git_config(
cls,
url: str = 'https://lore.kernel.org/all',
- **kwargs: object,
+ **kwargs: Unpack[_LoreNodeInitKwargs],
) -> LoreNode:
"""Create a :class:`LoreNode` using settings from git config.
@@ -319,7 +333,7 @@ class LoreNode:
except ValueError:
pass
- node = cls(url, **kwargs) # type: ignore[arg-type] # ty:ignore[invalid-argument-type]
+ node = cls(url, **kwargs)
val = gitcfg.get('useragentplus')
if isinstance(val, str) and val:
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 10/14] Add authheaders stub and typed callable
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (8 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 09/14] Type from_git_config keyword arguments Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 11/14] Replace batch mocks with subclasses Tamir Duberstein
` (4 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Add a generated authheaders stub under typings so the type checkers can
resolve the package. Then store authenticate_message directly on
LoreNode as a typed callable and update the authheaders tests to use
that shape.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
src/liblore/node.py | 26 +++++++++++++++++++-------
| 31 ++++++++++++++++++-------------
| 13 +++++++++++++
3 files changed, 50 insertions(+), 20 deletions(-)
diff --git a/src/liblore/node.py b/src/liblore/node.py
index f9817c1..8935dbf 100644
--- a/src/liblore/node.py
+++ b/src/liblore/node.py
@@ -13,11 +13,10 @@ import os
import re
import subprocess
import time
-import types
import urllib.parse
from datetime import datetime, timezone
from email.message import EmailMessage
-from typing import TYPE_CHECKING, TypedDict
+from typing import TYPE_CHECKING, Protocol, TypedDict
import requests
@@ -44,6 +43,19 @@ class _LoreNodeInitKwargs(TypedDict, total=False):
cache_ttl: int
+class _AuthenticateMessage(Protocol):
+ def __call__(
+ self,
+ msg: bytes,
+ authserv_id: str,
+ *,
+ dkim: bool = ...,
+ dmarc: bool = ...,
+ arc: bool = ...,
+ spf: bool = ...,
+ ) -> str: ...
+
+
def _get_config_from_git(
regexp: str,
multivals: list[str] | None = None,
@@ -206,12 +218,12 @@ class LoreNode:
if cache_dir is not None:
os.makedirs(cache_dir, exist_ok=True)
- self._authheaders: types.ModuleType | None = None
+ self._authenticate_message: _AuthenticateMessage | None = None
if add_auth_headers:
try:
- import authheaders # type: ignore[import-untyped]
+ import authheaders
- self._authheaders = authheaders
+ self._authenticate_message = authheaders.authenticate_message
except ImportError:
raise LibloreError(
'authheaders library is required for add_auth_headers. '
@@ -721,11 +733,11 @@ class LoreNode:
def _authenticate_msgs(self, msgs: list[EmailMessage]) -> None:
"""Add Authentication-Results headers via authheaders."""
- if self._authheaders is None:
+ if self._authenticate_message is None:
return
for msg in msgs:
msg_bytes = msg.as_bytes()
- auth_result = self._authheaders.authenticate_message(
+ auth_result = self._authenticate_message(
msg_bytes,
'liblore',
dkim=True,
--git a/tests/test_auth_headers.py b/tests/test_auth_headers.py
index b51050c..e149863 100644
--- a/tests/test_auth_headers.py
+++ b/tests/test_auth_headers.py
@@ -14,7 +14,12 @@ import pytest
import responses
from liblore import LibloreError
-from liblore.node import LoreNode
+from liblore.node import LoreNode, _AuthenticateMessage
+
+
+class _FakeAuthHeaders(ModuleType):
+ authenticate_message: _AuthenticateMessage
+
# =====================================================================
# Import-time validation
@@ -28,16 +33,16 @@ class TestAuthHeadersImport:
LoreNode(add_auth_headers=True)
def test_ok_when_authheaders_installed(self) -> None:
- fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock() # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
+ fake = _FakeAuthHeaders('authheaders')
+ fake.authenticate_message = MagicMock()
with patch.dict(sys.modules, {'authheaders': fake}):
node = LoreNode(add_auth_headers=True)
- assert node._authheaders is not None
+ assert node._authenticate_message is not None
node.close()
def test_default_is_disabled(self) -> None:
node = LoreNode()
- assert node._authheaders is None
+ assert node._authenticate_message is None
node.close()
@@ -55,8 +60,8 @@ class TestAuthenticateMsgs:
assert 'Authentication-Results' not in msg
def test_adds_header_when_enabled(self) -> None:
- fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
+ fake = _FakeAuthHeaders('authheaders')
+ fake.authenticate_message = MagicMock(
return_value='Authentication-Results: liblore; dkim=pass header.d=example.com',
)
with patch.dict(sys.modules, {'authheaders': fake}):
@@ -81,8 +86,8 @@ class TestAuthenticateMsgs:
node.close()
def test_skips_empty_result(self) -> None:
- fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock(return_value='') # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
+ fake = _FakeAuthHeaders('authheaders')
+ fake.authenticate_message = MagicMock(return_value='')
with patch.dict(sys.modules, {'authheaders': fake}):
node = LoreNode(add_auth_headers=True)
msg = EmailMessage()
@@ -94,8 +99,8 @@ class TestAuthenticateMsgs:
node.close()
def test_multiple_messages(self) -> None:
- fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
+ fake = _FakeAuthHeaders('authheaders')
+ fake.authenticate_message = MagicMock(
side_effect=[
'liblore; dkim=pass',
'Authentication-Results: liblore; dkim=fail',
@@ -125,8 +130,8 @@ class TestAuthenticateMsgs:
class TestAuthInFetchMethods:
@pytest.fixture()
def auth_node(self) -> Iterator[tuple[LoreNode, responses.RequestsMock]]:
- fake = ModuleType('authheaders')
- fake.authenticate_message = MagicMock( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
+ fake = _FakeAuthHeaders('authheaders')
+ fake.authenticate_message = MagicMock(
return_value='Authentication-Results: liblore; dkim=pass',
)
with patch.dict(sys.modules, {'authheaders': fake}):
--git a/typings/authheaders/__init__.pyi b/typings/authheaders/__init__.pyi
new file mode 100644
index 0000000..ed4f61e
--- /dev/null
+++ b/typings/authheaders/__init__.pyi
@@ -0,0 +1,13 @@
+"""
+This type stub file was generated by pyright.
+"""
+
+def authenticate_message(
+ msg: bytes,
+ authserv_id: str,
+ *,
+ dkim: bool = ...,
+ dmarc: bool = ...,
+ arc: bool = ...,
+ spf: bool = ...,
+) -> str: ...
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 11/14] Replace batch mocks with subclasses
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (9 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 10/14] Add authheaders stub and typed callable Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 12/14] Use CompletedProcess in git config tests Tamir Duberstein
` (3 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Replace method reassignment in the batch helper tests with small
LoreNode subclasses that override the fetch methods and record calls.
This removes the type-checking suppressions without adding more
patch-based mocking.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
tests/test_node.py | 88 +++++++++++++++++++++++++++++++-----------------------
1 file changed, 51 insertions(+), 37 deletions(-)
diff --git a/tests/test_node.py b/tests/test_node.py
index 4abb4f4..73c79a3 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -12,6 +12,7 @@ from unittest.mock import MagicMock, call, patch
import pytest
import requests
import responses
+from typing_extensions import override
from liblore import RemoteError
from liblore.node import LoreNode
@@ -496,24 +497,38 @@ class TestGetMessageByMsgid:
class TestBatchGetThreadByMsgid:
+ class _Node(LoreNode):
+ def __init__(self, results: list[list[EmailMessage]]) -> None:
+ super().__init__()
+ self.calls: list[tuple[str, bool, bool, str | None]] = []
+ self._results = results
+
+ @override
+ def get_thread_by_msgid(
+ self,
+ msgid: str,
+ *,
+ strict: bool = True,
+ sort: bool = False,
+ since: str | None = None,
+ ) -> list[EmailMessage]:
+ self.calls.append((msgid, strict, sort, since))
+ return self._results.pop(0)
+
def test_returns_ordered_results(self) -> None:
- node = LoreNode()
thread_a = [EmailMessage()]
thread_b = [EmailMessage(), EmailMessage()]
- node.get_thread_by_msgid = MagicMock(side_effect=[thread_a, thread_b]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([thread_a, thread_b])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid(['a@x', 'b@x'])
assert results == [thread_a, thread_b]
- assert node.get_thread_by_msgid.call_count == 2 # ty:ignore[unresolved-attribute]
+ assert len(node.calls) == 2
mock_sleep.assert_called_once_with(0.1)
def test_no_sleep_for_single_msgid(self) -> None:
- node = LoreNode()
thread = [EmailMessage()]
- node.get_thread_by_msgid = MagicMock(return_value=thread) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([thread])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid(['only@x'])
@@ -521,9 +536,7 @@ class TestBatchGetThreadByMsgid:
mock_sleep.assert_not_called()
def test_passes_kwargs(self) -> None:
- node = LoreNode()
- node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([[EmailMessage()]])
with patch('liblore.node.time.sleep'):
node.batch_get_thread_by_msgid(
['a@x'],
@@ -532,32 +545,23 @@ class TestBatchGetThreadByMsgid:
since='20240101',
)
- node.get_thread_by_msgid.assert_called_once_with( # ty:ignore[unresolved-attribute]
- 'a@x',
- strict=False,
- sort=True,
- since='20240101',
- )
+ assert node.calls == [('a@x', False, True, '20240101')]
def test_sleep_count_matches_gaps(self) -> None:
- node = LoreNode()
- node.get_thread_by_msgid = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([[EmailMessage()], [EmailMessage()], [EmailMessage()]])
with patch('liblore.node.time.sleep') as mock_sleep:
node.batch_get_thread_by_msgid(['a@x', 'b@x', 'c@x'])
assert mock_sleep.call_args_list == [call(0.1), call(0.1)]
def test_empty_list(self) -> None:
- node = LoreNode()
- node.get_thread_by_msgid = MagicMock() # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_msgid([])
assert results == []
mock_sleep.assert_not_called()
- node.get_thread_by_msgid.assert_not_called() # ty:ignore[unresolved-attribute]
+ assert node.calls == []
# =====================================================================
@@ -566,24 +570,36 @@ class TestBatchGetThreadByMsgid:
class TestBatchGetThreadByQuery:
+ class _Node(LoreNode):
+ def __init__(self, results: list[list[EmailMessage]]) -> None:
+ super().__init__()
+ self.calls: list[tuple[str, bool]] = []
+ self._results = results
+
+ @override
+ def get_thread_by_query(
+ self,
+ query: str,
+ *,
+ full_threads: bool = False,
+ ) -> list[EmailMessage]:
+ self.calls.append((query, full_threads))
+ return self._results.pop(0)
+
def test_returns_ordered_results(self) -> None:
- node = LoreNode()
result_a = [EmailMessage()]
result_b = [EmailMessage(), EmailMessage()]
- node.get_thread_by_query = MagicMock(side_effect=[result_a, result_b]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([result_a, result_b])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query(['q1', 'q2'])
assert results == [result_a, result_b]
- assert node.get_thread_by_query.call_count == 2 # ty:ignore[unresolved-attribute]
+ assert len(node.calls) == 2
mock_sleep.assert_called_once_with(0.1)
def test_no_sleep_for_single_query(self) -> None:
- node = LoreNode()
result = [EmailMessage()]
- node.get_thread_by_query = MagicMock(return_value=result) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([result])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query(['only_query'])
@@ -591,24 +607,22 @@ class TestBatchGetThreadByQuery:
mock_sleep.assert_not_called()
def test_sleep_count_matches_gaps(self) -> None:
- node = LoreNode()
- node.get_thread_by_query = MagicMock(return_value=[EmailMessage()]) # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node(
+ [[EmailMessage()], [EmailMessage()], [EmailMessage()], [EmailMessage()]]
+ )
with patch('liblore.node.time.sleep') as mock_sleep:
node.batch_get_thread_by_query(['q1', 'q2', 'q3', 'q4'])
assert mock_sleep.call_args_list == [call(0.1), call(0.1), call(0.1)]
def test_empty_list(self) -> None:
- node = LoreNode()
- node.get_thread_by_query = MagicMock() # type: ignore[method-assign] # ty:ignore[invalid-assignment]
-
+ node = self._Node([])
with patch('liblore.node.time.sleep') as mock_sleep:
results = node.batch_get_thread_by_query([])
assert results == []
mock_sleep.assert_not_called()
- node.get_thread_by_query.assert_not_called() # ty:ignore[unresolved-attribute]
+ assert node.calls == []
# =====================================================================
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 12/14] Use CompletedProcess in git config tests
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (10 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 11/14] Replace batch mocks with subclasses Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 13/14] Update README for uv-based dev checks Tamir Duberstein
` (2 subsequent siblings)
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Replace MagicMock subprocess results with real
subprocess.CompletedProcess values in the git config helper tests. This
keeps the behavior the same while giving the mocks a concrete,
type-checked shape.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
tests/test_node.py | 77 +++++++++++++++++++++++++++++++-----------------------
1 file changed, 45 insertions(+), 32 deletions(-)
diff --git a/tests/test_node.py b/tests/test_node.py
index 73c79a3..89be909 100644
--- a/tests/test_node.py
+++ b/tests/test_node.py
@@ -5,6 +5,7 @@ from __future__ import annotations
import gzip
import os
+import subprocess
from datetime import datetime, timezone
from email.message import EmailMessage
from unittest.mock import MagicMock, call, patch
@@ -1557,15 +1558,17 @@ class TestGetConfigFromGit:
"""Parses git config -z output correctly."""
from liblore.node import _get_config_from_git
- mock_result = MagicMock()
- mock_result.returncode = 0
- mock_result.stdout = (
- 'lore.fallback\nhttps://tor.lore.kernel.org\x00'
- 'lore.fallback\nhttps://sea.lore.kernel.org\x00'
- 'lore.autoprobe\ntrue\x00'
- 'lore.probettl\n7200\x00'
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=0,
+ stdout=(
+ 'lore.fallback\nhttps://tor.lore.kernel.org\x00'
+ 'lore.fallback\nhttps://sea.lore.kernel.org\x00'
+ 'lore.autoprobe\ntrue\x00'
+ 'lore.probettl\n7200\x00'
+ ),
)
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_config_from_git(r'^lore\.', multivals=['fallback'])
assert cfg == {
@@ -1593,10 +1596,12 @@ class TestGetConfigFromGit:
"""Returns empty dict when no keys match the regexp."""
from liblore.node import _get_config_from_git
- mock_result = MagicMock()
- mock_result.returncode = 1
- mock_result.stdout = ''
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=1,
+ stdout='',
+ )
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_config_from_git(r'^lore\.')
assert cfg == {}
@@ -1605,10 +1610,12 @@ class TestGetConfigFromGit:
"""A key without a value (no newline) defaults to 'true'."""
from liblore.node import _get_config_from_git
- mock_result = MagicMock()
- mock_result.returncode = 0
- mock_result.stdout = 'lore.autoprobe\x00'
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=0,
+ stdout='lore.autoprobe\x00',
+ )
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_config_from_git(r'^lore\.')
assert cfg == {'autoprobe': 'true'}
@@ -1621,15 +1628,17 @@ class TestGetSubsectionConfig:
"""Parses keys from [liblore "https://lore.kernel.org"]."""
from liblore.node import _get_subsection_config
- mock_result = MagicMock()
- mock_result.returncode = 0
- mock_result.stdout = (
- 'liblore.https://lore.kernel.org.fallback\nhttps://tor.lore.kernel.org\x00'
- 'liblore.https://lore.kernel.org.fallback\nhttps://sea.lore.kernel.org\x00'
- 'liblore.https://lore.kernel.org.autoprobe\ntrue\x00'
- 'liblore.https://lore.kernel.org.useragentplus\nmyuuid\x00'
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=0,
+ stdout=(
+ 'liblore.https://lore.kernel.org.fallback\nhttps://tor.lore.kernel.org\x00'
+ 'liblore.https://lore.kernel.org.fallback\nhttps://sea.lore.kernel.org\x00'
+ 'liblore.https://lore.kernel.org.autoprobe\ntrue\x00'
+ 'liblore.https://lore.kernel.org.useragentplus\nmyuuid\x00'
+ ),
)
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_subsection_config(
'liblore',
'https://lore.kernel.org',
@@ -1649,10 +1658,12 @@ class TestGetSubsectionConfig:
"""Subsection names with dots (URLs) are parsed correctly."""
from liblore.node import _get_subsection_config
- mock_result = MagicMock()
- mock_result.returncode = 0
- mock_result.stdout = 'liblore.https://subspace.kernel.org.fallback\nhttps://mirror.example.com\x00'
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=0,
+ stdout='liblore.https://subspace.kernel.org.fallback\nhttps://mirror.example.com\x00',
+ )
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_subsection_config(
'liblore',
'https://subspace.kernel.org',
@@ -1667,10 +1678,12 @@ class TestGetSubsectionConfig:
"""Returns empty dict when no keys match the subsection."""
from liblore.node import _get_subsection_config
- mock_result = MagicMock()
- mock_result.returncode = 1
- mock_result.stdout = ''
- with patch('liblore.node.subprocess.run', return_value=mock_result):
+ result = subprocess.CompletedProcess[str](
+ args=['git'],
+ returncode=1,
+ stdout='',
+ )
+ with patch('liblore.node.subprocess.run', return_value=result):
cfg = _get_subsection_config(
'liblore',
'https://nonexistent.example.com',
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 13/14] Update README for uv-based dev checks
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (11 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 12/14] Use CompletedProcess in git config tests Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-13 22:08 ` [PATCH v2 14/14] Add b4 send configuration Tamir Duberstein
2026-04-14 17:28 ` [PATCH v2 00/14] Harden local type checking and test mocking Konstantin Ryabitsev
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Document the development workflow in terms of `uv` and mention `pyright`,
`ty check`, and `ruff format`.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
README.md | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git a/README.md b/README.md
index 3709d7e..1c3863e 100644
--- a/README.md
+++ b/README.md
@@ -613,25 +613,28 @@ except LibloreError:
Install with development dependencies:
```shell
-pip install -e ".[dev]"
+uv sync --all-extras --all-groups
```
Run the test suite:
```shell
-pytest
+uv run --all-extras --all-groups pytest
```
Type checking:
```shell
-mypy src/liblore/ --strict
+uv run --all-extras --all-groups ty check
+uv run --all-extras --all-groups mypy .
+uv run --all-extras --all-groups pyright
```
Linting:
```shell
-ruff check src/liblore/
+uv run --all-extras --all-groups ruff format --check
+uv run --all-extras --all-groups ruff check
```
## Bug Reports
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* [PATCH v2 14/14] Add b4 send configuration
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (12 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 13/14] Update README for uv-based dev checks Tamir Duberstein
@ 2026-04-13 22:08 ` Tamir Duberstein
2026-04-14 17:28 ` [PATCH v2 00/14] Harden local type checking and test mocking Konstantin Ryabitsev
14 siblings, 0 replies; 16+ messages in thread
From: Tamir Duberstein @ 2026-04-13 22:08 UTC (permalink / raw)
To: Kernel.org Tools; +Cc: Konstantin Ryabitsev, Tamir Duberstein
Populate the repository b4 config with the lore masks and default send
settings used for kernel.org submissions.
Signed-off-by: Tamir Duberstein <tamird@gmail.com>
---
.b4-config | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/.b4-config b/.b4-config
new file mode 100644
index 0000000..5084661
--- /dev/null
+++ b/.b4-config
@@ -0,0 +1,7 @@
+[b4]
+ midmask = https://lore.kernel.org/%s
+ linkmask = https://patch.msgid.link/%s
+ send-series-to = Kernel.org Tools <tools@kernel.org>
+ send-series-cc = Konstantin Ryabitsev <konstantin@linuxfoundation.org>
+ send-endpoint-web = https://lkml.kernel.org/_b4_submit
+ send-prefixes = liblore
--
2.53.0
^ permalink raw reply related [flat|nested] 16+ messages in thread* Re: [PATCH v2 00/14] Harden local type checking and test mocking
2026-04-13 22:08 [PATCH v2 00/14] Harden local type checking and test mocking Tamir Duberstein
` (13 preceding siblings ...)
2026-04-13 22:08 ` [PATCH v2 14/14] Add b4 send configuration Tamir Duberstein
@ 2026-04-14 17:28 ` Konstantin Ryabitsev
14 siblings, 0 replies; 16+ messages in thread
From: Konstantin Ryabitsev @ 2026-04-14 17:28 UTC (permalink / raw)
To: Kernel.org Tools, Tamir Duberstein
On Mon, 13 Apr 2026 18:08:14 -0400, Tamir Duberstein wrote:
> Harden local type checking and test mocking
>
> This series raises the local quality floor by making the type-checking
> tooling stricter and by removing test patterns that rely on unsafe or
> poorly typed mocks.
>
> It starts by wiring the repository into b4 CI checks, then adds ruff,
> pyright, and ty so the same checks run consistently in one place. After
> that, it removes suppressions incrementally by making the code and tests
> more explicit and better typed: typed keyword forwarding in
> from_git_config, a typed authheaders integration, and test-local
> subclasses instead of method reassignment.
>
> [...]
Applied, thanks!
[01/14] Add ci script and mypy suppressions
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=828e718aa0a4
[02/14] Type make_msg and drop test suppressions
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=b2981347f362
[03/14] Add ruff import checks to b4 CI
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=ac6e2df890e8
[04/14] Add ruff format check to CI
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=100116fefeef
[05/14] Add pyright strict checks to CI
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=6109a737d052
[06/14] Replace HTTP session mocks with responses
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=5445b53cd074
[07/14] Add ty checks to CI
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=7cd032b465fc
[08/14] Drop redundant read-only property test
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=53041a41e451
[09/14] Type from_git_config keyword arguments
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=18603217b450
[10/14] Add authheaders stub and typed callable
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=c9e8cf46e133
[11/14] Replace batch mocks with subclasses
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=bdca5ed2752e
[12/14] Use CompletedProcess in git config tests
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=f993dde25931
[13/14] Update README for uv-based dev checks
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=204122fd6ad6
[14/14] Add b4 send configuration
https://git.kernel.org/pub/scm/utils/liblore/liblore.git/commit/?id=9ba4ddb7e4aa
Best regards,
--
KR
^ permalink raw reply [flat|nested] 16+ messages in thread