| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354 |
- from __future__ import annotations
- import io
- import tempfile
- import unittest
- from contextlib import redirect_stdout
- from pathlib import Path
- from unittest.mock import patch
- from src.data import bootstrap, status
- class CliTests(unittest.TestCase):
- def test_bootstrap_requires_all_flag(self) -> None:
- with self.assertRaises(SystemExit):
- bootstrap.main([])
- def test_status_renders_manifest(self) -> None:
- temp_dir = tempfile.TemporaryDirectory()
- self.addCleanup(temp_dir.cleanup)
- root = Path(temp_dir.name)
- (root / "src" / "data").mkdir(parents=True, exist_ok=True)
- manifest = {
- "provider": "fake",
- "generated_at": "2026-01-01T00:00:00+00:00",
- "common_sample": {"start_date": "2020-01-01", "end_date": "2020-01-31"},
- "instruments": {
- "sse50": {
- "name": "上证50",
- "actual_start": "2020-01-01",
- "layers": {
- "clean": {"end_date": "2020-01-31"},
- "features": {"end_date": "2020-01-31"},
- },
- }
- },
- }
- class PipelineStub:
- def status_snapshot(self):
- return manifest
- buf = io.StringIO()
- with patch("src.data.status.build_pipeline", return_value=PipelineStub()):
- with redirect_stdout(buf):
- exit_code = status.main([])
- output = buf.getvalue()
- self.assertEqual(exit_code, 0)
- self.assertIn("provider: fake", output)
- self.assertIn("[sse50] 上证50", output)
- if __name__ == "__main__":
- unittest.main()
|