From cd6fa419372378aa20cd17c8a3a9c37cf0faf4c4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:59:52 +0800 Subject: [PATCH 1/3] fix runtime target schedules and notification acknowledgements Co-Authored-By: Codex --- .../common/port_adapters.py | 6 +-- src/quant_platform_kit/common/ports.py | 2 +- .../common/runtime_target.py | 51 +++++++++++++++++++ .../notifications/channel.py | 4 +- .../notifications/cycle_channel.py | 42 ++++++++------- .../notifications/events.py | 13 ++--- .../notifications/telegram.py | 18 +++++++ tests/test_common_port_adapters.py | 9 +++- tests/test_cycle_channel.py | 22 ++++++++ tests/test_notification_channels.py | 22 ++++++++ tests/test_notification_events.py | 14 +++++ tests/test_runtime_target.py | 30 +++++++++++ tests/test_telegram.py | 19 +++++++ 13 files changed, 221 insertions(+), 31 deletions(-) create mode 100644 tests/test_cycle_channel.py create mode 100644 tests/test_notification_channels.py diff --git a/src/quant_platform_kit/common/port_adapters.py b/src/quant_platform_kit/common/port_adapters.py index c5d2eac..a6c5983 100644 --- a/src/quant_platform_kit/common/port_adapters.py +++ b/src/quant_platform_kit/common/port_adapters.py @@ -9,10 +9,10 @@ @dataclass(frozen=True) class CallableNotificationPort(NotificationPort): - sender: Callable[[str], None] + sender: Callable[[str], bool | None] - def send_text(self, message: str) -> None: - self.sender(message) + def send_text(self, message: str) -> bool | None: + return self.sender(message) @dataclass(frozen=True) diff --git a/src/quant_platform_kit/common/ports.py b/src/quant_platform_kit/common/ports.py index 7237b51..96e0929 100644 --- a/src/quant_platform_kit/common/ports.py +++ b/src/quant_platform_kit/common/ports.py @@ -25,7 +25,7 @@ def submit_order(self, order: OrderIntent) -> ExecutionReport: class NotificationPort(Protocol): - def send_text(self, message: str) -> None: + def send_text(self, message: str) -> bool | None: """Send a plain-text notification.""" diff --git a/src/quant_platform_kit/common/runtime_target.py b/src/quant_platform_kit/common/runtime_target.py index ec1adb3..002b2aa 100644 --- a/src/quant_platform_kit/common/runtime_target.py +++ b/src/quant_platform_kit/common/runtime_target.py @@ -4,6 +4,7 @@ from collections.abc import Iterable from dataclasses import dataclass, asdict from typing import Any, Mapping +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @dataclass(frozen=True) @@ -15,6 +16,10 @@ class RuntimeTarget: account_selector: tuple[str, ...] = () account_scope: str | None = None service_name: str | None = None + market: str | None = None + market_calendar: str | None = None + market_timezone: str | None = None + scheduler: dict[str, Any] | None = None execution_windows: dict[str, Any] | None = None @property @@ -23,6 +28,9 @@ def execution_mode(self) -> str: def to_dict(self) -> dict[str, object]: payload = asdict(self) + for field in ("market", "market_calendar", "market_timezone", "scheduler"): + if payload.get(field) is None: + payload.pop(field, None) payload["execution_mode"] = self.execution_mode return payload @@ -59,6 +67,29 @@ def _normalize_account_selector(value: Iterable[str] | str | None) -> tuple[str, return tuple(normalized) +def _normalize_market_metadata( + *, + market: str | None, + market_calendar: str | None, + market_timezone: str | None, +) -> tuple[str | None, str | None, str | None]: + values = ( + _normalize_optional_string(market), + _normalize_optional_string(market_calendar), + _normalize_optional_string(market_timezone), + ) + if any(values) and not all(values): + raise ValueError( + "market metadata must include market, market_calendar, and market_timezone together" + ) + if values[2] is not None: + try: + ZoneInfo(values[2]) + except (ZoneInfoNotFoundError, ValueError) as exc: + raise ValueError(f"invalid market_timezone: {values[2]!r}") from exc + return values + + def build_runtime_target( *, platform_id: str, @@ -68,8 +99,17 @@ def build_runtime_target( account_selector: Iterable[str] | str | None = None, account_scope: str | None = None, service_name: str | None = None, + market: str | None = None, + market_calendar: str | None = None, + market_timezone: str | None = None, + scheduler: Mapping[str, Any] | None = None, execution_windows: Mapping[str, Any] | None = None, ) -> RuntimeTarget: + normalized_market, normalized_calendar, normalized_timezone = _normalize_market_metadata( + market=market, + market_calendar=market_calendar, + market_timezone=market_timezone, + ) return RuntimeTarget( platform_id=str(platform_id).strip(), strategy_profile=str(strategy_profile).strip(), @@ -78,6 +118,10 @@ def build_runtime_target( account_selector=_normalize_account_selector(account_selector), account_scope=_normalize_optional_string(account_scope), service_name=_normalize_optional_string(service_name), + market=normalized_market, + market_calendar=normalized_calendar, + market_timezone=normalized_timezone, + scheduler=dict(scheduler) if scheduler is not None else None, execution_windows=dict(execution_windows) if execution_windows is not None else None, ) @@ -145,6 +189,9 @@ def resolve_runtime_target_from_env( execution_windows = payload.get("execution_windows") if execution_windows is not None and not isinstance(execution_windows, dict): raise ValueError("RUNTIME_TARGET_JSON.execution_windows must be an object when present") + scheduler = payload.get("scheduler") + if scheduler is not None and not isinstance(scheduler, dict): + raise ValueError("RUNTIME_TARGET_JSON.scheduler must be an object when present") return build_runtime_target( platform_id=resolved_platform_id, @@ -154,5 +201,9 @@ def resolve_runtime_target_from_env( account_selector=payload.get("account_selector"), account_scope=payload.get("account_scope"), service_name=payload.get("service_name"), + market=payload.get("market"), + market_calendar=payload.get("market_calendar"), + market_timezone=payload.get("market_timezone"), + scheduler=scheduler, execution_windows=execution_windows, ) diff --git a/src/quant_platform_kit/notifications/channel.py b/src/quant_platform_kit/notifications/channel.py index 031f9e4..f96e5fd 100644 --- a/src/quant_platform_kit/notifications/channel.py +++ b/src/quant_platform_kit/notifications/channel.py @@ -233,9 +233,9 @@ def send_message( ) -> bool: from .telegram import send_telegram_message return send_telegram_message( - chat_id=chat_id, + chat_ids=chat_id, text=text, - token=token, + bot_token=token, api_base_url=api_base_url, parse_mode=parse_mode, ) diff --git a/src/quant_platform_kit/notifications/cycle_channel.py b/src/quant_platform_kit/notifications/cycle_channel.py index 348670f..2758989 100644 --- a/src/quant_platform_kit/notifications/cycle_channel.py +++ b/src/quant_platform_kit/notifications/cycle_channel.py @@ -1,8 +1,8 @@ """Cycle notification sender factory. Provides a single entry point for building channel-agnostic notification -senders. Each sender is a ``Callable[[str], None]`` that accepts a fully -rendered message string and delivers it to the configured channel. +senders. Each sender is a ``Callable[[str], bool]`` that accepts a fully +rendered message string and returns whether the transport acknowledged it. Supported channels: - telegram — Telegram Bot API (default) @@ -58,8 +58,8 @@ def build_cycle_sender( telegram_chat_id: str | None = None, webhook_url: str | None = None, printer: Any = print, -) -> Callable[[str], None]: - """Build a ``send_message(message: str) -> None`` callback for *channel*. +) -> Callable[[str], bool]: + """Build a ``send_message(message: str) -> bool`` callback for *channel*. Args: channel: One of ``"telegram"``, ``"wecom"``, ``"dingtalk"``, @@ -163,23 +163,23 @@ def _build_telegram_sender( token: str | None, chat_id: str | None, printer: Any = print, -) -> Callable[[str], None]: +) -> Callable[[str], bool]: from .telegram import send_telegram_message resolved_token = str(token or "").strip() resolved_chat_id = str(chat_id or "").strip() - def send_message(message: str) -> None: + def send_message(message: str) -> bool: if not resolved_token: printer("Cycle sender: telegram token not configured", flush=True) - return + return False if not resolved_chat_id: printer("Cycle sender: telegram chat_id not configured", flush=True) - return + return False text = str(message or "").strip() if not text: - return - send_telegram_message( + return False + return send_telegram_message( bot_token=resolved_token, chat_ids=resolved_chat_id, text=text, @@ -199,7 +199,7 @@ def _build_webhook_sender( channel: str, webhook_url: str | None, printer: Any = print, -) -> Callable[[str], None]: +) -> Callable[[str], bool]: from .webhook import ( WEBHOOK_PROVIDER_DINGTALK, WEBHOOK_PROVIDER_FEISHU, @@ -213,25 +213,31 @@ def _build_webhook_sender( resolved_url = str(webhook_url or "").strip() - def send_message(message: str) -> None: + def send_message(message: str) -> bool: nonlocal resolved_url if not resolved_url: printer(f"Cycle sender: webhook URL not configured for {channel}", flush=True) - return + return False text = str(message or "").strip() if not text: - return + return False if channel == WEBHOOK_PROVIDER_WECOM: - send_wecom_webhook(resolved_url, text, printer=printer) + return send_wecom_webhook(resolved_url, text, printer=printer) elif channel == WEBHOOK_PROVIDER_DINGTALK: - send_dingtalk_webhook(resolved_url, text, printer=printer) + return send_dingtalk_webhook(resolved_url, text, printer=printer) elif channel == WEBHOOK_PROVIDER_FEISHU: - send_feishu_webhook(resolved_url, text, printer=printer) + return send_feishu_webhook(resolved_url, text, printer=printer) elif channel == WEBHOOK_PROVIDER_SERVERCHAN: # ServerChan: first line → title, remainder → body lines = text.split("\n", 1) title = lines[0] body = lines[1] if len(lines) > 1 else "" - send_serverchan_webhook(resolved_url, title=title, body=body, printer=printer) + return send_serverchan_webhook( + resolved_url, + title=title, + body=body, + printer=printer, + ) + return False return send_message diff --git a/src/quant_platform_kit/notifications/events.py b/src/quant_platform_kit/notifications/events.py index 8cc190e..29267a9 100644 --- a/src/quant_platform_kit/notifications/events.py +++ b/src/quant_platform_kit/notifications/events.py @@ -19,10 +19,10 @@ class NotificationPublisher: """Publish rendered notifications to the configured sinks.""" log_message: Callable[[str], None] - send_message: Callable[[str], None] + send_message: Callable[[str], bool | None] - def publish(self, notification: RenderedNotification) -> None: - publish_rendered_notification( + def publish(self, notification: RenderedNotification) -> bool: + return publish_rendered_notification( notification, log_message=self.log_message, send_message=self.send_message, @@ -33,12 +33,13 @@ def publish_rendered_notification( notification: RenderedNotification, *, log_message: Callable[[str], None], - send_message: Callable[[str], None], -) -> None: + send_message: Callable[[str], bool | None], +) -> bool: """Write the detailed log copy and send the compact user notification.""" detailed = str(notification.detailed_text or "").strip() compact = str(notification.compact_text or "").strip() if detailed: log_message(detailed) if compact: - send_message(compact) + return send_message(compact) is not False + return True diff --git a/src/quant_platform_kit/notifications/telegram.py b/src/quant_platform_kit/notifications/telegram.py index 241b032..8342454 100644 --- a/src/quant_platform_kit/notifications/telegram.py +++ b/src/quant_platform_kit/notifications/telegram.py @@ -122,18 +122,36 @@ def _request_succeeded( printer, chat_id: str, ) -> bool: + raw_response = b"" try: with request_opener(request, timeout=timeout) as response: status = getattr(response, "status", None) if status is None: status = response.getcode() status = int(status) + read_response = getattr(response, "read", None) + if callable(read_response): + raw_response = read_response() except Exception as exc: printer(f"Telegram send failed for {chat_id}: {redact_sensitive_text(exc)}", flush=True) return False if status < 200 or status >= 300: printer(f"Telegram send failed for {chat_id}: HTTP {status}", flush=True) return False + if raw_response: + try: + response_payload = json.loads(raw_response.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + response_payload = None + if isinstance(response_payload, dict) and response_payload.get("ok") is False: + description = redact_sensitive_text( + response_payload.get("description") or "negative API acknowledgement" + ) + printer( + f"Telegram send failed for {chat_id}: {description}", + flush=True, + ) + return False return True diff --git a/tests/test_common_port_adapters.py b/tests/test_common_port_adapters.py index 972a65e..2204ea8 100644 --- a/tests/test_common_port_adapters.py +++ b/tests/test_common_port_adapters.py @@ -4,7 +4,7 @@ import unittest from quant_platform_kit.common.models import PricePoint, PriceSeries, QuoteSnapshot -from quant_platform_kit.common.port_adapters import CallableMarketDataPort +from quant_platform_kit.common.port_adapters import CallableMarketDataPort, CallableNotificationPort class CallableMarketDataPortTests(unittest.TestCase): @@ -59,5 +59,12 @@ def test_get_price_series_raises_when_not_configured(self) -> None: port.get_price_series("SOXL") +class CallableNotificationPortTests(unittest.TestCase): + def test_send_text_propagates_delivery_result(self) -> None: + port = CallableNotificationPort(lambda _message: False) + + self.assertIs(port.send_text("rebalance"), False) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_cycle_channel.py b/tests/test_cycle_channel.py new file mode 100644 index 0000000..947a8f6 --- /dev/null +++ b/tests/test_cycle_channel.py @@ -0,0 +1,22 @@ +from quant_platform_kit.notifications import cycle_channel +from quant_platform_kit.notifications import telegram + + +def test_telegram_cycle_sender_propagates_transport_failure(monkeypatch): + monkeypatch.setattr( + telegram, + "send_telegram_message", + lambda **_kwargs: False, + ) + sender = cycle_channel.build_cycle_sender( + telegram_token="token", + telegram_chat_id="chat", + ) + + assert sender("rebalance") is False + + +def test_telegram_cycle_sender_fails_when_credentials_are_missing(): + sender = cycle_channel.build_cycle_sender() + + assert sender("rebalance") is False diff --git a/tests/test_notification_channels.py b/tests/test_notification_channels.py new file mode 100644 index 0000000..e65e77f --- /dev/null +++ b/tests/test_notification_channels.py @@ -0,0 +1,22 @@ +from quant_platform_kit.notifications import telegram +from quant_platform_kit.notifications.channel import TelegramChatChannel + + +def test_telegram_chat_channel_uses_current_sender_contract(monkeypatch): + observed = {} + + def fake_send(**kwargs): + observed.update(kwargs) + return False + + monkeypatch.setattr(telegram, "send_telegram_message", fake_send) + + sent = TelegramChatChannel().send_message( + "chat", + "message", + token="token", + ) + + assert sent is False + assert observed["bot_token"] == "token" + assert observed["chat_ids"] == "chat" diff --git a/tests/test_notification_events.py b/tests/test_notification_events.py index aca5428..d4b6b40 100644 --- a/tests/test_notification_events.py +++ b/tests/test_notification_events.py @@ -42,3 +42,17 @@ def test_notification_publisher_uses_configured_sinks(): assert logs == ["log"] assert sends == ["send"] + + +def test_notification_publisher_propagates_explicit_delivery_failure(): + publisher = NotificationPublisher( + log_message=lambda _message: None, + send_message=lambda _message: False, + ) + + assert ( + publisher.publish( + RenderedNotification(detailed_text="log", compact_text="send"), + ) + is False + ) diff --git a/tests/test_runtime_target.py b/tests/test_runtime_target.py index 548f7e2..8509a2c 100644 --- a/tests/test_runtime_target.py +++ b/tests/test_runtime_target.py @@ -56,6 +56,9 @@ def test_resolve_runtime_target_from_env_prefers_structured_json(self) -> None: '{"platform_id":"longbridge","strategy_profile":"global_etf_rotation",' '"dry_run_only":true,"deployment_selector":"HK","account_selector":["HK"],' '"account_scope":"HK","service_name":"longbridge-quant-hk-service",' + '"market":"HK","market_calendar":"XHKG",' + '"market_timezone":"Asia/Hong_Kong",' + '"scheduler":{"timezone":"Asia/Hong_Kong","main_time":"45 15 * * *"},' '"execution_mode":"paper","execution_windows":{"precheck":{"enabled":true,' '"offset_minutes":15,"mode":"notify_only"},"execution":{"enabled":true,' '"offset_minutes":15,"mode":"paper"}}}' @@ -71,6 +74,10 @@ def test_resolve_runtime_target_from_env_prefers_structured_json(self) -> None: self.assertEqual(target.account_selector, ("HK",)) self.assertEqual(target.account_scope, "HK") self.assertEqual(target.service_name, "longbridge-quant-hk-service") + self.assertEqual(target.market, "HK") + self.assertEqual(target.market_calendar, "XHKG") + self.assertEqual(target.market_timezone, "Asia/Hong_Kong") + self.assertEqual(target.scheduler["main_time"], "45 15 * * *") self.assertTrue(target.execution_windows["precheck"]["enabled"]) self.assertEqual(target.execution_windows["execution"]["offset_minutes"], 15) @@ -86,6 +93,29 @@ def test_resolve_runtime_target_from_env_rejects_mismatched_execution_mode(self) expected_platform_id="schwab", ) + def test_resolve_runtime_target_from_env_rejects_partial_market_metadata(self) -> None: + with self.assertRaisesRegex(ValueError, "market metadata must include"): + resolve_runtime_target_from_env( + env={ + "RUNTIME_TARGET_JSON": ( + '{"platform_id":"schwab","strategy_profile":"tqqq_growth_income",' + '"dry_run_only":false,"market":"US"}' + ) + }, + expected_platform_id="schwab", + ) + + def test_build_runtime_target_rejects_invalid_market_timezone(self) -> None: + with self.assertRaisesRegex(ValueError, "invalid market_timezone"): + build_runtime_target( + platform_id="schwab", + strategy_profile="tqqq_growth_income", + dry_run_only=False, + market="US", + market_calendar="NYSE", + market_timezone="../New_York", + ) + def test_resolve_runtime_target_from_env_rejects_mismatched_platform(self) -> None: with self.assertRaisesRegex( ValueError, diff --git a/tests/test_telegram.py b/tests/test_telegram.py index ef48b3a..a17f9ad 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -30,6 +30,25 @@ def test_send_telegram_message_calls_telegram_api(self) -> None: self.assertEqual(request.full_url, "https://api.telegram.org/bottoken/sendMessage") self.assertEqual(request.method, "POST") + def test_send_telegram_message_rejects_negative_api_ack(self) -> None: + fake_response = MagicMock(status=200) + fake_response.read.return_value = json.dumps( + {"ok": False, "description": "chat not found"} + ).encode("utf-8") + fake_context = MagicMock() + fake_context.__enter__.return_value = fake_response + fake_context.__exit__.return_value = None + + sent = send_telegram_message( + "token", + "123", + "hello world", + opener=lambda *_args, **_kwargs: fake_context, + printer=lambda *_args, **_kwargs: None, + ) + + self.assertFalse(sent) + if __name__ == "__main__": unittest.main() From 765e5c9d8cc194a725d9564caf2817cd7a42d9b1 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:03:33 +0800 Subject: [PATCH 2/3] fix CI Ruff rule stability Co-Authored-By: Codex --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 0feb4ff..efdf5c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ quant_platform_kit = ["schemas/*.json"] target-version = "py310" [tool.ruff.lint] +select = ["E4", "E7", "E9", "F"] ignore = ["E701", "E702", "E741", "F841", "F821", "B008", "F401", "F601", "E402"] [tool.coverage.run] From 48cbf5719104e3101baf0706699e4be5de149539 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:07:45 +0800 Subject: [PATCH 3/3] fix review compatibility and strict Telegram ack Co-Authored-By: Codex --- .../common/runtime_target.py | 2 +- .../notifications/telegram.py | 32 ++++++++++--------- tests/test_runtime_target.py | 18 +++++++++++ ..._strategy_plugin_telegram_notifications.py | 3 ++ tests/test_telegram.py | 19 +++++++++++ 5 files changed, 58 insertions(+), 16 deletions(-) diff --git a/src/quant_platform_kit/common/runtime_target.py b/src/quant_platform_kit/common/runtime_target.py index 002b2aa..0b38c11 100644 --- a/src/quant_platform_kit/common/runtime_target.py +++ b/src/quant_platform_kit/common/runtime_target.py @@ -16,11 +16,11 @@ class RuntimeTarget: account_selector: tuple[str, ...] = () account_scope: str | None = None service_name: str | None = None + execution_windows: dict[str, Any] | None = None market: str | None = None market_calendar: str | None = None market_timezone: str | None = None scheduler: dict[str, Any] | None = None - execution_windows: dict[str, Any] | None = None @property def execution_mode(self) -> str: diff --git a/src/quant_platform_kit/notifications/telegram.py b/src/quant_platform_kit/notifications/telegram.py index 8342454..f75c60c 100644 --- a/src/quant_platform_kit/notifications/telegram.py +++ b/src/quant_platform_kit/notifications/telegram.py @@ -138,21 +138,23 @@ def _request_succeeded( if status < 200 or status >= 300: printer(f"Telegram send failed for {chat_id}: HTTP {status}", flush=True) return False - if raw_response: - try: - response_payload = json.loads(raw_response.decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError): - response_payload = None - if isinstance(response_payload, dict) and response_payload.get("ok") is False: - description = redact_sensitive_text( - response_payload.get("description") or "negative API acknowledgement" - ) - printer( - f"Telegram send failed for {chat_id}: {description}", - flush=True, - ) - return False - return True + try: + response_payload = json.loads(raw_response.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + response_payload = None + if isinstance(response_payload, dict) and response_payload.get("ok") is True: + return True + description = ( + response_payload.get("description") + if isinstance(response_payload, dict) + else None + ) + printer( + "Telegram send failed for " + f"{chat_id}: {redact_sensitive_text(description or 'missing affirmative API acknowledgement')}", + flush=True, + ) + return False def _telegram_send_message_endpoint(api_base_url: str | None, bot_token: str) -> str: diff --git a/tests/test_runtime_target.py b/tests/test_runtime_target.py index 8509a2c..f461d30 100644 --- a/tests/test_runtime_target.py +++ b/tests/test_runtime_target.py @@ -3,6 +3,7 @@ import unittest from quant_platform_kit.common.runtime_target import ( + RuntimeTarget, build_runtime_context_fields, build_runtime_target, resolve_runtime_target_from_env, @@ -10,6 +11,23 @@ class RuntimeTargetTests(unittest.TestCase): + def test_runtime_target_preserves_existing_execution_windows_positional_slot(self) -> None: + windows = {"execution": {"enabled": True, "mode": "paper"}} + + target = RuntimeTarget( + "longbridge", + "global_etf_rotation", + True, + "HK", + ("HK",), + "HK", + "longbridge-quant-hk-service", + windows, + ) + + self.assertEqual(target.execution_windows, windows) + self.assertIsNone(target.market) + def test_build_runtime_target_normalizes_selectors_and_mode(self) -> None: target = build_runtime_target( platform_id=" longbridge ", diff --git a/tests/test_strategy_plugin_telegram_notifications.py b/tests/test_strategy_plugin_telegram_notifications.py index e6ec562..b0f6eb8 100644 --- a/tests/test_strategy_plugin_telegram_notifications.py +++ b/tests/test_strategy_plugin_telegram_notifications.py @@ -55,6 +55,9 @@ def _auto_consumable_signal(): class _FakeResponse: status = 200 + def read(self): + return b'{"ok": true}' + def __enter__(self): return self diff --git a/tests/test_telegram.py b/tests/test_telegram.py index a17f9ad..9746912 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -49,6 +49,25 @@ def test_send_telegram_message_rejects_negative_api_ack(self) -> None: self.assertFalse(sent) + def test_send_telegram_message_requires_affirmative_api_ack(self) -> None: + for payload in (b"not-json", b"{}", b'{"ok": 0}'): + with self.subTest(payload=payload): + fake_response = MagicMock(status=200) + fake_response.read.return_value = payload + fake_context = MagicMock() + fake_context.__enter__.return_value = fake_response + fake_context.__exit__.return_value = None + + sent = send_telegram_message( + "token", + "123", + "hello world", + opener=lambda *_args, **_kwargs: fake_context, + printer=lambda *_args, **_kwargs: None, + ) + + self.assertFalse(sent) + if __name__ == "__main__": unittest.main()