293 lines
11 KiB
Python
293 lines
11 KiB
Python
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()
|