86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
"""sync_utils 纯函数：取值清洗、日期格式化、限频重试语义。"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
|
||
from app.data import sync_utils
|
||
|
||
|
||
def test_s_clean():
|
||
assert sync_utils.s_clean(" 平安银行 ") == "平安银行"
|
||
assert sync_utils.s_clean("") is None
|
||
assert sync_utils.s_clean(" ") is None
|
||
assert sync_utils.s_clean(None) is None
|
||
assert sync_utils.s_clean(float("nan")) is None # pandas NaN
|
||
|
||
|
||
def test_f_clean():
|
||
assert sync_utils.f_clean("3.14") == 3.14
|
||
assert sync_utils.f_clean(2) == 2.0
|
||
assert sync_utils.f_clean(float("nan")) is None
|
||
assert sync_utils.f_clean(None) is None
|
||
assert sync_utils.f_clean("abc") is None
|
||
assert sync_utils.f_clean(math.inf) == math.inf # inf 非 NaN，原样保留
|
||
|
||
|
||
def test_d8_iso():
|
||
assert sync_utils.d8_iso("20240102") == "2024-01-02"
|
||
assert sync_utils.d8_iso(20240102) == "2024-01-02"
|
||
assert sync_utils.d8_iso(None) is None
|
||
assert sync_utils.d8_iso("") is None
|
||
|
||
|
||
def test_fresh():
|
||
now = sync_utils.utcnow()
|
||
assert sync_utils.fresh(now, days=7) is True
|
||
assert sync_utils.fresh(now - timedelta(days=8), days=7) is False
|
||
assert sync_utils.fresh(None, days=7) is False
|
||
|
||
|
||
def test_call_retry_passes_through_args():
|
||
calls = []
|
||
|
||
def fn(a, b=0):
|
||
calls.append((a, b))
|
||
return a + b
|
||
|
||
assert sync_utils.call_retry(fn, 1, b=2) == 3
|
||
assert calls == [(1, 2)]
|
||
|
||
|
||
def test_call_retry_rate_limit_retries_once(monkeypatch):
|
||
"""「每分钟」级频率超限等 62s 重试一次；重试成功则返回结果。"""
|
||
monkeypatch.setattr(sync_utils.time, "sleep", lambda s: None)
|
||
calls = []
|
||
|
||
def fn():
|
||
calls.append(1)
|
||
if len(calls) == 1:
|
||
raise RuntimeError("抱歉，您每分钟最多访问该接口5次")
|
||
return "ok"
|
||
|
||
assert sync_utils.call_retry(fn) == "ok"
|
||
assert len(calls) == 2
|
||
|
||
|
||
def test_call_retry_hourly_limit_raises(monkeypatch):
|
||
"""小时级限频不重试，直接抛出。"""
|
||
monkeypatch.setattr(sync_utils.time, "sleep", lambda s: None)
|
||
|
||
def fn():
|
||
raise RuntimeError("您每小时最多访问该接口10次")
|
||
|
||
with pytest.raises(RuntimeError):
|
||
sync_utils.call_retry(fn)
|
||
|
||
|
||
def test_call_retry_other_errors_raise():
|
||
def fn():
|
||
raise ValueError("数据源故障")
|
||
|
||
with pytest.raises(ValueError):
|
||
sync_utils.call_retry(fn)
|