import logging import subprocess import unittest from datetime import date, datetime, time, timedelta, timezone from unittest.mock import patch import ndx_daily_report as report class ParseTradeDatetimeTests(unittest.TestCase): def test_nasdaq_date_only_timestamp_represents_market_close(self) -> None: parsed = report.parse_trade_datetime("Jun 5, 2026", "America/New_York") self.assertEqual(parsed.date().isoformat(), "2026-06-05") self.assertEqual(parsed.time(), time(16, 0)) def test_naive_iso_timestamp_uses_trade_timezone(self) -> None: parsed = report.parse_trade_datetime("2026-06-05T16:01:00", "America/New_York") self.assertEqual(parsed.hour, 16) self.assertIsNotNone(parsed.tzinfo) def test_date_only_timestamp_passes_closed_market_validation(self) -> None: parsed = report.parse_trade_datetime("Jun 5, 2026", "America/New_York") with patch.object(report, "datetime", wraps=datetime) as mocked_datetime: mocked_datetime.now.return_value = datetime(2026, 6, 5, 20, 0, tzinfo=parsed.tzinfo) report.validate_complete_regular_close(parsed, "Closed", "America/New_York") def test_date_only_timestamp_is_incomplete_before_market_close(self) -> None: parsed = report.parse_trade_datetime("Jun 5, 2026", "America/New_York") with patch.object(report, "datetime", wraps=datetime) as mocked_datetime: mocked_datetime.now.return_value = datetime(2026, 6, 5, 10, 0, tzinfo=parsed.tzinfo) with self.assertRaises(report.IncompleteMarketData): report.validate_complete_regular_close(parsed, "Closed", "America/New_York") class ReplayReportTests(unittest.TestCase): def setUp(self) -> None: self.beijing_tz = timezone(timedelta(hours=8)) def test_china_rest_day_does_not_trigger_replay(self) -> None: now = datetime(2026, 6, 19, 8, 0, tzinfo=self.beijing_tz) needed = report.should_replay_report(now, {}, True, False, True, 8) self.assertFalse(needed) def test_first_workday_after_holiday_triggers_replay(self) -> None: now = datetime(2026, 6, 22, 8, 0, tzinfo=self.beijing_tz) needed = report.should_replay_report(now, {}, True, True, False, 8) self.assertTrue(needed) def test_makeup_workday_after_holiday_triggers_replay(self) -> None: now = datetime(2026, 1, 4, 8, 0, tzinfo=self.beijing_tz) needed = report.should_replay_report(now, {}, True, True, False, 8) self.assertTrue(needed) def test_ordinary_workday_does_not_trigger_replay(self) -> None: now = datetime(2026, 6, 23, 8, 0, tzinfo=self.beijing_tz) needed = report.should_replay_report(now, {}, True, True, True, 8) self.assertFalse(needed) def test_replay_waits_until_delivery_hour(self) -> None: now = datetime(2026, 6, 22, 7, 59, tzinfo=self.beijing_tz) needed = report.should_replay_report(now, {}, True, True, False, 8) self.assertFalse(needed) def test_replay_is_not_repeated_after_delivery(self) -> None: now = datetime(2026, 6, 22, 13, 0, tzinfo=self.beijing_tz) needed = report.should_replay_report( now, {"2026-06-22": "2026-06-19"}, True, True, False, 8, ) self.assertFalse(needed) class DeliveryScheduleTests(unittest.TestCase): def test_friday_close_is_scheduled_for_saturday_beijing_time(self) -> None: trade_tz = report.get_timezone("America/New_York") delivery_tz = report.get_timezone("Asia/Shanghai") scheduled = report.next_delivery_time_after_close( "2026-06-05", trade_tz, delivery_tz, 8, ) self.assertEqual(scheduled.date(), date(2026, 6, 6)) self.assertEqual(scheduled.time(), time(8, 0)) def test_us_close_report_schedule_is_independent_of_china_holiday(self) -> None: trade_tz = report.get_timezone("America/New_York") delivery_tz = report.get_timezone("Asia/Shanghai") scheduled = report.next_delivery_time_after_close( "2026-06-18", trade_tz, delivery_tz, 8, ) self.assertEqual(scheduled.date(), date(2026, 6, 19)) class HolidayApiTests(unittest.TestCase): def setUp(self) -> None: self.config = { "report": { "holiday_api_v2": "https://api.haoshenqi.top/holiday", "holiday_api_timeout_seconds": 5, }, "network": {"transport": "urllib", "use_environment_proxy": False}, } self.logger = logging.getLogger("holiday-api-test") self.beijing_tz = timezone(timedelta(hours=8)) @patch("ndx_daily_report.fetch_holiday_v2_status") def test_v2_requests_today_and_yesterday(self, fetch_status) -> None: fetch_status.side_effect = [True, False] now = datetime(2026, 6, 22, 8, 0, tzinfo=self.beijing_tz) result = report.get_china_workday_pair(self.config, now, self.logger) self.assertEqual(result, (True, False)) self.assertEqual( [call.args[1] for call in fetch_status.call_args_list], [ "https://api.haoshenqi.top/holiday/today", "https://api.haoshenqi.top/holiday/yesterday", ], ) @patch("ndx_daily_report.fetch_holiday_v2_status", side_effect=RuntimeError("offline")) def test_api_failure_falls_back_to_weekdays(self, _fetch_status) -> None: monday = datetime(2026, 6, 22, 8, 0, tzinfo=self.beijing_tz) result = report.get_china_workday_pair(self.config, monday, self.logger) self.assertEqual(result, (True, False)) @patch("ndx_daily_report.http_text", return_value="工作".encode()) def test_v2_work_response_is_workday(self, _http_text) -> None: self.assertTrue( report.fetch_holiday_v2_status( self.config, "https://api.haoshenqi.top/holiday/today", 5, ) ) @patch("ndx_daily_report.http_text", return_value="休息".encode()) def test_v2_rest_response_is_not_workday(self, _http_text) -> None: self.assertFalse( report.fetch_holiday_v2_status( self.config, "https://api.haoshenqi.top/holiday/today", 5, ) ) @patch("ndx_daily_report.http_text", return_value=b"unknown") def test_unknown_v2_response_is_rejected(self, _http_text) -> None: with self.assertRaises(ValueError): report.fetch_holiday_v2_status( self.config, "https://api.haoshenqi.top/holiday/today", 5, ) class QQBotTokenTests(unittest.TestCase): def setUp(self) -> None: self.config = { "qqbot": { "appid": "config-appid", "appsecret": "config-secret", "token_api": "https://bots.qq.com/app/getAppAccessToken", }, "network": {"transport": "urllib", "use_environment_proxy": False}, } self.logger = logging.getLogger("qqbot-token-test") @patch.dict( "os.environ", {"QQBOT_APPID": "env-appid", "QQBOT_SECRET": "env-secret"}, clear=False, ) def test_environment_credentials_override_config(self) -> None: self.assertEqual( report.get_qqbot_credentials(self.config), ("env-appid", "env-secret"), ) self.assertEqual( report.get_qqbot_credential_source(self.config), "environment", ) @patch("ndx_daily_report.http_json") def test_token_request_uses_official_field_names(self, http_json) -> None: http_json.return_value = {"access_token": "token", "expires_in": "7200"} token = report.get_qqbot_token(self.config, self.logger) self.assertEqual(token, "token") self.assertEqual( http_json.call_args.kwargs["body"], {"appId": "config-appid", "clientSecret": "config-secret"}, ) @patch("ndx_daily_report.http_json") def test_token_error_includes_qq_error_code_and_message(self, http_json) -> None: http_json.return_value = {"code": 100007, "message": "appid invalid"} self.config["qqbot"]["token_retry_count"] = 0 with self.assertRaisesRegex( ValueError, "code=100007, message=appid invalid", ): report.get_qqbot_token(self.config, self.logger) @patch("ndx_daily_report.time.sleep") @patch("ndx_daily_report.http_json") def test_token_request_retries_one_transient_error(self, http_json, sleep) -> None: http_json.side_effect = [ {"code": 100007, "message": "appid invalid"}, {"access_token": "token", "expires_in": "7200"}, ] token = report.get_qqbot_token(self.config, self.logger) self.assertEqual(token, "token") self.assertEqual(http_json.call_count, 2) sleep.assert_called_once_with(1) def test_legacy_appkey_remains_supported(self) -> None: config = {"qqbot": {"appid": "appid", "appkey": "legacy-secret"}} self.assertEqual( report.get_qqbot_credentials(config), ("appid", "legacy-secret"), ) self.assertEqual( report.get_qqbot_credential_source(config), "config.appkey(legacy)", ) class CurlRequestTests(unittest.TestCase): @patch("ndx_daily_report.subprocess.run") @patch("ndx_daily_report.shutil.which", return_value="/usr/bin/curl") def test_curl_uses_http1_and_retries_by_default(self, _which, run) -> None: run.return_value = subprocess.CompletedProcess([], 0, stdout=b"ok", stderr=b"") result = report.curl_request("GET", "https://example.com", None, 15, None, {}) self.assertEqual(result, b"ok") command = run.call_args.args[0] self.assertIn("--http1.1", command) self.assertEqual(command[command.index("--retry") + 1], "2") self.assertIn("--retry-all-errors", command) @patch("ndx_daily_report.subprocess.run") @patch("ndx_daily_report.shutil.which", return_value="/usr/bin/curl") def test_curl_does_not_retry_post_requests(self, _which, run) -> None: run.return_value = subprocess.CompletedProcess([], 0, stdout=b"{}", stderr=b"") report.curl_request("POST", "https://example.com", {"value": 1}, 15, None, {}) command = run.call_args.args[0] self.assertNotIn("--retry", command) self.assertIn("Content-Type: application/json", command) if __name__ == "__main__": unittest.main()