120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
"""统一数据服务测试。"""
|
||
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
from typing import List, Tuple
|
||
|
||
import pytest
|
||
|
||
from sidecar.config import SidecarConfig
|
||
from sidecar.models import KlineBar, Snapshot
|
||
from sidecar.service import UnifiedDataService
|
||
from sidecar.storage import DuckDBStore
|
||
|
||
|
||
class FakeSource:
|
||
"""测试用假数据源。"""
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
功能说明：初始化测试数据源计数器。
|
||
参数说明：无。
|
||
返回值说明：无返回值。
|
||
注意事项：仅用于单元测试。
|
||
"""
|
||
self.kline_calls = 0
|
||
self.snapshot_calls = 0
|
||
|
||
def fetch_kline(self, symbol: str, period: str, limit: int) -> List[KlineBar]:
|
||
"""
|
||
功能说明：返回固定 K 线测试数据。
|
||
参数说明：symbol 为股票代码，period 为周期，limit 为条数。
|
||
返回值说明：返回 KlineBar 列表。
|
||
注意事项：会记录调用次数用于验证缓存。
|
||
"""
|
||
self.kline_calls += 1
|
||
return [
|
||
KlineBar(symbol, period, date(2026, 6, 22), 1, 2, 1, 2, 100, 200, "fake"),
|
||
KlineBar(symbol, period, date(2026, 6, 23), 2, 3, 2, 3, 200, 300, "fake"),
|
||
][:limit]
|
||
|
||
def fetch_snapshot(self, symbol: str) -> Snapshot:
|
||
"""
|
||
功能说明：返回固定快照测试数据。
|
||
参数说明：symbol 为股票代码。
|
||
返回值说明：返回 Snapshot。
|
||
注意事项：会记录调用次数用于验证 TTL 缓存。
|
||
"""
|
||
self.snapshot_calls += 1
|
||
return Snapshot(
|
||
symbol=symbol,
|
||
name="测试股",
|
||
trade_time=datetime(2026, 6, 23, 15, 0, 0),
|
||
price=10,
|
||
previous_close=9,
|
||
open=9.5,
|
||
high=10.5,
|
||
low=9.4,
|
||
volume=1000,
|
||
amount=10000,
|
||
change=1,
|
||
change_percent=11.11,
|
||
turnover_rate=2,
|
||
pe_ttm=8,
|
||
pe_static=9,
|
||
pb=1.2,
|
||
market_cap=100,
|
||
float_market_cap=80,
|
||
limit_up=9.9,
|
||
limit_down=8.1,
|
||
source="fake",
|
||
)
|
||
|
||
|
||
@pytest.fixture()
|
||
def service(tmp_path: Path) -> Tuple[UnifiedDataService, FakeSource]:
|
||
"""
|
||
功能说明：创建使用临时 DuckDB 的统一服务。
|
||
参数说明：tmp_path 为 pytest 临时目录。
|
||
返回值说明：返回服务实例和假数据源。
|
||
注意事项：用例间数据库完全隔离。
|
||
"""
|
||
source = FakeSource()
|
||
store = DuckDBStore(str(tmp_path / "test.duckdb"))
|
||
config = SidecarConfig(db_path=str(tmp_path / "test.duckdb"), snapshot_ttl_seconds=60)
|
||
return UnifiedDataService(store, config, source, [source]), source
|
||
|
||
|
||
def test_2_1_kline_fetches_then_reads_cache(service: Tuple[UnifiedDataService, FakeSource]) -> None:
|
||
"""
|
||
功能说明：验证 K 线首次远端获取后可从 DuckDB 缓存读取。
|
||
参数说明：service 为测试夹具。
|
||
返回值说明：无返回值。
|
||
注意事项：用例编号 2-1，第二次 limit 小于缓存数量时不再调用远端。
|
||
"""
|
||
data_service, source = service
|
||
|
||
first = data_service.get_kline("600000", limit=2)
|
||
second = data_service.get_kline("sh600000", limit=1)
|
||
|
||
assert len(first) == 2
|
||
assert len(second) == 1
|
||
assert second[0].trade_date == date(2026, 6, 23)
|
||
assert source.kline_calls == 1
|
||
|
||
|
||
def test_2_2_snapshot_uses_ttl_cache(service: Tuple[UnifiedDataService, FakeSource]) -> None:
|
||
"""
|
||
功能说明：验证快照在 TTL 内复用 DuckDB 缓存。
|
||
参数说明：service 为测试夹具。
|
||
返回值说明：无返回值。
|
||
注意事项：用例编号 2-2，第二次读取不得再次调用远端。
|
||
"""
|
||
data_service, source = service
|
||
|
||
first = data_service.get_snapshot("600000")
|
||
second = data_service.get_snapshot("sh600000")
|
||
|
||
assert first.price == 10
|
||
assert second.price == 10
|
||
assert source.snapshot_calls == 1 |