diff --git a/README.md b/README.md index 4ab03c5..594eade 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ All constructor arguments after `token` / `chat_id` are keyword-only. |---|---|---| | `token` | `None` | Bot token; falls back to `TG_TOKEN`. | | `chat_id` | `None` | Target chat; falls back to `TG_CHAT_ID`. | +| `topic_id` | `None` | Forum topic id; falls back to `TG_TOPIC_ID`. When set on a forum supergroup, messages go to that topic; when unset, to the group's General topic. | | `level` | `logging.WARNING` | Minimum level (passed to `setLevel`). | | `batch_size` | `1` | Max records per message (`>= 1`). `1` = send each record immediately. | | `flush_interval` | `5.0` | Max seconds a partial batch waits (`>= 0`). | @@ -85,6 +86,7 @@ All constructor arguments after `token` / `chat_id` are keyword-only. |---|---| | `TG_TOKEN` | `token` argument is omitted. | | `TG_CHAT_ID` | `chat_id` argument is omitted. | +| `TG_TOPIC_ID` | `topic_id` argument is omitted. | ## Batching diff --git a/tests/test_config_stats.py b/tests/test_config_stats.py index 9e64775..2ea16ff 100644 --- a/tests/test_config_stats.py +++ b/tests/test_config_stats.py @@ -3,11 +3,12 @@ from __future__ import annotations import queue +from typing import cast import pytest from tg_logging_handler import queue_policy -from tg_logging_handler.config import resolve_credentials +from tg_logging_handler.config import resolve_credentials, resolve_topic_id from tg_logging_handler.exceptions import TelegramConfigError from tg_logging_handler.stats import HandlerStats, StatsCollector @@ -71,6 +72,48 @@ def test_invalid_long_token_reveals_only_ends(self) -> None: assert "mmmm" not in message # the middle stays hidden +class TestResolveTopicId: + def test_explicit_int_passes_through(self) -> None: + assert resolve_topic_id(42) == 42 + + def test_none_returns_none_without_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("TG_TOPIC_ID", raising=False) + assert resolve_topic_id(None) is None + + def test_env_string_is_parsed(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "42") + assert resolve_topic_id(None) == 42 + + def test_explicit_arg_wins_over_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "7") + assert resolve_topic_id(42) == 42 + + def test_empty_env_is_missing(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "") + assert resolve_topic_id(None) is None + + def test_invalid_env_string_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "not-a-number") + with pytest.raises(ValueError, match="TG_TOPIC_ID must be an integer"): + resolve_topic_id(None) + + @pytest.mark.parametrize("bad", [True, False, 3.5, "42"]) + def test_non_int_explicit_values_are_rejected(self, bad: object) -> None: + # bool is an int subclass, floats compare fine, and str would crash on + # the range comparison; all must fail with a clean ValueError. + with pytest.raises(ValueError, match="topic_id must be an integer"): + resolve_topic_id(bad) # type: ignore[arg-type] # deliberate misuse + + def test_rejection_message_names_the_type(self) -> None: + with pytest.raises(ValueError, match="got bool"): + resolve_topic_id(True) + + def test_non_positive_is_rejected(self) -> None: + for bad in (0, -1, -100): + with pytest.raises(ValueError, match="positive"): + resolve_topic_id(bad) + + class TestStats: def test_initial_snapshot_is_zero(self) -> None: assert StatsCollector().snapshot() == HandlerStats() @@ -96,6 +139,31 @@ def test_snapshot_does_not_reflect_later_changes(self) -> None: assert stats.snapshot().sent == 1 +class _ScriptedQueue: + """Duck-typed stand-in for ``queue.Queue`` with per-call scripted outcomes. + + Only the two methods ``put_drop_oldest`` touches are provided. ``puts`` + entries: ``"full"`` makes ``put_nowait`` raise ``queue.Full``, anything + else succeeds. ``gets`` entries: ``"empty"`` makes ``get_nowait`` raise + ``queue.Empty``, anything else returns an evictable item. This lets tests + exercise the concurrency-race branches that a real queue can never reach + single-threaded. + """ + + def __init__(self, puts: list[str], gets: list[str]) -> None: + self._puts = iter(puts) + self._gets = iter(gets) + + def put_nowait(self, item: object) -> None: + if next(self._puts) == "full": + raise queue.Full + + def get_nowait(self) -> object: + if next(self._gets) == "empty": + raise queue.Empty + return "evicted" + + class TestQueuePolicy: def test_put_drop_newest_succeeds_when_space(self) -> None: q: queue.Queue[object] = queue.Queue(maxsize=1) @@ -129,6 +197,44 @@ def test_put_drop_oldest_no_eviction_when_space(self) -> None: assert result.enqueued is True assert result.dropped == 0 + def test_put_drop_oldest_handles_queue_draining_between_evict_and_retry( + self, + ) -> None: + # Race path: between a failed put and the eviction get, another + # producer drains the queue, so get_nowait raises Empty and the loop + # must retry instead of crashing. Scripted via a fake queue. + fake = cast( + "queue.Queue[object]", + _ScriptedQueue(puts=["full", "full", "ok"], gets=["empty", "item"]), + ) + result = queue_policy.put_drop_oldest(fake, "new") + assert result.enqueued is True + assert result.dropped == 1 # one real eviction happened, then room appeared + + def test_put_drop_oldest_gives_up_when_slot_keeps_refilling(self) -> None: + # Race path: the freed slot is instantly refilled by a competing + # producer on every attempt, so all evictions fail and the incoming + # item is dropped (and counted) after the bounded attempt budget. + fake = cast( + "queue.Queue[object]", + _ScriptedQueue(puts=["full", "full", "full", "full"], gets=["item"] * 3), + ) + result = queue_policy.put_drop_oldest(fake, "new") + assert result.enqueued is False + assert result.dropped == 4 # 3 evicted victims + the incoming record + + def test_put_drop_oldest_lands_item_after_attempt_budget(self) -> None: + # Race path: the slot keeps refilling for all three attempts, but the + # final (post-budget) put succeeds, so the item lands but the victims + # are still counted. + fake = cast( + "queue.Queue[object]", + _ScriptedQueue(puts=["full", "full", "full", "ok"], gets=["item"] * 3), + ) + result = queue_policy.put_drop_oldest(fake, "new") + assert result.enqueued is True + assert result.dropped == 3 + def test_put_block_enqueues(self) -> None: q: queue.Queue[object] = queue.Queue(maxsize=1) result = queue_policy.put_block(q, 1) diff --git a/tests/test_handler.py b/tests/test_handler.py index 6fa1c4f..bf5f9f5 100644 --- a/tests/test_handler.py +++ b/tests/test_handler.py @@ -96,6 +96,58 @@ def test_level_kwarg_applied(fast_handler_factory: HandlerFactory) -> None: assert handler.level == logging.ERROR +def test_topic_id_validation_raises() -> None: + for bad in (0, -1, -100): + with pytest.raises(ValueError): + TelegramLoggingHandler( + token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, topic_id=bad + ) + + +def test_topic_id_none_is_allowed(fast_handler_factory: HandlerFactory) -> None: + handler = fast_handler_factory(topic_id=None) # None is allowed + handler.close() + + +def test_topic_id_from_env_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "42") + handler = TelegramLoggingHandler( + token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, api_base_url=API_BASE + ) + assert handler._sender._message_thread_id == 42 + handler.close() + + +def test_topic_id_arg_wins_over_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "7") + handler = TelegramLoggingHandler( + token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, topic_id=42, api_base_url=API_BASE + ) + assert handler._sender._message_thread_id == 42 + handler.close() + + +def test_topic_id_invalid_env_raises(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TG_TOPIC_ID", "not-a-number") + with pytest.raises(ValueError): + TelegramLoggingHandler( + token=TEST_TOKEN, chat_id=TEST_CHAT_ID, validate=False, api_base_url=API_BASE + ) + + +def test_topic_id_reaches_the_wire( + fast_handler_factory: HandlerFactory, mock_api: respx.MockRouter, wait_for: WaitFor +) -> None: + import json + + handler = fast_handler_factory(topic_id=42) + handler.emit(_record("topic log")) + assert wait_for(lambda: handler.stats.sent == 1) + route = next(r for r in mock_api.routes if "sendMessage" in str(r.pattern)) + body = json.loads(route.calls[0].request.content.decode()) + assert body["message_thread_id"] == 42 + + # --- emit contract ---------------------------------------------------------- diff --git a/tests/test_sender_retry.py b/tests/test_sender_retry.py index 8aea4f2..120a412 100644 --- a/tests/test_sender_retry.py +++ b/tests/test_sender_retry.py @@ -129,6 +129,64 @@ def test_parse_mode_included_when_set() -> None: assert body["parse_mode"] == "HTML" +@respx.mock +def test_message_thread_id_included_when_set() -> None: + route = respx.post(SEND_URL).mock(return_value=_ok()) + sleep = _RecordingSleep() + sender = TelegramSender(TOKEN, CHAT, API_BASE, message_thread_id=42, sleep=sleep) + sender.send_with_retry("hi") + body = json.loads(route.calls[0].request.content.decode()) + assert body["message_thread_id"] == 42 + + +@respx.mock +def test_message_thread_id_omitted_when_not_set() -> None: + route = respx.post(SEND_URL).mock(return_value=_ok()) + sleep = _RecordingSleep() + _sender(sleep).send_with_retry("hi") + body = json.loads(route.calls[0].request.content.decode()) + assert "message_thread_id" not in body # backward compatible: no topic field + + +@respx.mock +def test_thread_not_found_400_is_a_permanent_failure() -> None: + # Wrong or closed topic id: Telegram answers "message thread not found". + # It is a permanent 4xx, so it must not be retried and must count as failed. + route = respx.post(SEND_URL).mock( + return_value=httpx.Response(400, text="Bad Request: message thread not found") + ) + sleep = _RecordingSleep() + outcome = TelegramSender( + TOKEN, CHAT, API_BASE, message_thread_id=999, sleep=sleep + ).send_with_retry("hi") + assert outcome.delivered is False + assert outcome.retries == 0 + assert route.call_count == 1 + assert sleep.calls == [] + + +@respx.mock +def test_429_heads_up_notice_targets_the_topic() -> None: + # The one-time 429 notice must land in the same topic as the logs, not + # fall back to the General topic. + route = respx.post(SEND_URL).mock( + side_effect=[ + httpx.Response(429, headers={"retry-after": "1"}), + _ok(), # the heads-up notice POST + _ok(), # the retried original message + ] + ) + sleep = _RecordingSleep() + sender = TelegramSender(TOKEN, CHAT, API_BASE, message_thread_id=42, sleep=sleep) + outcome = sender.send_with_retry("real log line") + assert outcome.delivered is True + notice = [ + c for c in route.calls if _NOTICE_MARKER in json.loads(c.request.content.decode())["text"] + ] + assert len(notice) == 1 + assert json.loads(notice[0].request.content.decode())["message_thread_id"] == 42 + + @respx.mock def test_429_waits_are_capped_so_a_stuck_429_cannot_spin_forever() -> None: # 11 consecutive 429s (cap is 10) must give up, not loop forever. diff --git a/tg_logging_handler/config.py b/tg_logging_handler/config.py index 13cfd70..bd06b5c 100644 --- a/tg_logging_handler/config.py +++ b/tg_logging_handler/config.py @@ -10,7 +10,7 @@ from ._constants import DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT from .exceptions import TelegramConfigError -__all__ = ["resolve_credentials", "validate_token"] +__all__ = ["resolve_credentials", "resolve_topic_id", "validate_token"] def resolve_credentials( @@ -51,6 +51,39 @@ def resolve_credentials( return token, str(resolved_chat) +def resolve_topic_id(topic_id: int | None) -> int | None: + """Resolve the forum topic id from args, falling back to ``TG_TOPIC_ID``. + + Args: + topic_id: Explicit topic id, or ``None`` to read ``TG_TOPIC_ID``. + + Returns: + A positive ``int`` topic id, or ``None`` when neither is set + (messages then go to the group's General topic). + + Raises: + ValueError: If the value is not an integer or not positive. + """ + if topic_id is not None: + if type(topic_id) is not int: + # bool is an int subclass and floats compare fine, but both are + # always mistakes here; anything else (e.g. str) would crash on + # the range comparison below. Reject non-ints up front. + raise ValueError(f"topic_id must be an integer, got {type(topic_id).__name__}") + value = topic_id + else: + raw = os.environ.get("TG_TOPIC_ID") + if raw is None or raw == "": + return None + try: + value = int(raw) + except ValueError as exc: + raise ValueError(f"TG_TOPIC_ID must be an integer, got {raw!r}") from exc + if value < 1: + raise ValueError("topic_id must be a positive integer") + return value + + def _looks_like_bot_token(token: str) -> bool: """Heuristic: Telegram bot tokens are ``':'``.""" return re.fullmatch(r"\d+:.+", token) is not None diff --git a/tg_logging_handler/handler.py b/tg_logging_handler/handler.py index cfe059e..b41dc9f 100644 --- a/tg_logging_handler/handler.py +++ b/tg_logging_handler/handler.py @@ -14,7 +14,7 @@ from typing import Literal from . import queue_policy -from .config import resolve_credentials, validate_token +from .config import resolve_credentials, resolve_topic_id, validate_token from .sender import TelegramSender from .stats import HandlerStats, StatsCollector from .worker import SHUTDOWN, WorkerThread @@ -31,6 +31,9 @@ class TelegramLoggingHandler(logging.Handler): Args: token: Bot token. Falls back to ``TG_TOKEN``. Raises if neither is set. chat_id: Target chat id. Falls back to ``TG_CHAT_ID``. Raises if neither is set. + topic_id: Target forum topic id. Falls back to ``TG_TOPIC_ID``. When set + and the chat is a forum supergroup, messages go to that topic; when + unset they go to the group's General topic. level: Minimum level, passed to ``setLevel``. Defaults to ``WARNING``. batch_size: Max records per outgoing message (``>= 1``). flush_interval: Max seconds a partial batch waits (``>= 0``). @@ -52,6 +55,7 @@ def __init__( token: str | None = None, chat_id: str | int | None = None, *, + topic_id: int | None = None, level: int | str = logging.WARNING, batch_size: int = 1, flush_interval: float = 5.0, @@ -76,6 +80,7 @@ def __init__( raise ValueError("queue_maxsize must be >= 0") resolved_token, resolved_chat = resolve_credentials(token, chat_id) + resolved_topic = resolve_topic_id(topic_id) if validate: validate_token(resolved_token, api_base_url) @@ -96,6 +101,7 @@ def __init__( api_base_url, parse_mode=parse_mode, max_retries=max_retries, + message_thread_id=resolved_topic, ) self._worker = WorkerThread( record_queue=self._queue, diff --git a/tg_logging_handler/sender.py b/tg_logging_handler/sender.py index d41c00d..5c48b5d 100644 --- a/tg_logging_handler/sender.py +++ b/tg_logging_handler/sender.py @@ -59,12 +59,14 @@ def __init__( api_base_url: str, parse_mode: str | None = None, max_retries: int = 3, + message_thread_id: int | None = None, sleep: Callable[[float], None] = time.sleep, ) -> None: self._url = f"{api_base_url}/bot{token}/sendMessage" self._chat_id = chat_id self._parse_mode = parse_mode self._max_retries = max_retries + self._message_thread_id = message_thread_id self._sleep = sleep self._client: httpx.Client | None = None # One-time heads-up when rate limiting first starts, so a 429 storm does @@ -142,6 +144,8 @@ def _notify_rate_limited(self) -> None: "rate limit; recommend checking it manually." ) payload = {"chat_id": self._chat_id, "text": text, "disable_web_page_preview": True} + if self._message_thread_id is not None: + payload["message_thread_id"] = self._message_thread_id try: self._get_client().post(self._url, json=payload) except httpx.HTTPError as exc: @@ -159,6 +163,8 @@ def _send_once(self, text: str) -> None: "text": text, "disable_web_page_preview": True, } + if self._message_thread_id is not None: + payload["message_thread_id"] = self._message_thread_id if self._parse_mode is not None: payload["parse_mode"] = self._parse_mode