修复 NDX 日期解析和网络重试

2026-06-06 13:04:47 +08:00
parent 792922e06f
commit e2af2eff79
3 changed files with 88 additions and 3 deletions
@@ -0,0 +1,54 @@
+import subprocess
+import unittest
+from datetime import time
+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")
+
+        report.validate_complete_regular_close(parsed, "Closed", "America/New_York")
+
+
+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)
+
+
+if __name__ == "__main__":
+    unittest.main()