81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from pipeline import FakeSource, FatalError, RetryableError, TokenBucket, run_sync, with_retry
|
|
|
|
|
|
def make_rows(n, start=0, ts_prefix="2026-10-23T10:00:", status="paid"):
|
|
return [{"id": f"o{i:03d}", "amount_cents": i, "status": status, "updated_at": f"{ts_prefix}{i:02d}"} for i in range(start, start + n)]
|
|
|
|
|
|
def test_resume_from_watermark_and_idempotent():
|
|
conn = sqlite3.connect(":memory:")
|
|
src = FakeSource(make_rows(30))
|
|
assert run_sync(conn, src, page_size=10, sleep=lambda s: None).written == 30
|
|
again = run_sync(conn, src, page_size=10, sleep=lambda s: None)
|
|
assert again.fetched == 0 # 水位线生效，不重复拉取
|
|
src.rows += make_rows(5, start=30)
|
|
assert run_sync(conn, src, page_size=10, sleep=lambda s: None).written == 5
|
|
assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 35
|
|
|
|
|
|
def test_older_version_does_not_overwrite_newer():
|
|
conn = sqlite3.connect(":memory:")
|
|
run_sync(conn, FakeSource([{"id": "a", "amount_cents": 1, "status": "refunded", "updated_at": "2026-10-23T12:00:00"}]), job="j1")
|
|
# 另一个来源（或重放）送来旧版本
|
|
run_sync(conn, FakeSource([{"id": "a", "amount_cents": 1, "status": "paid", "updated_at": "2026-10-23T09:00:00"}]), job="j2")
|
|
assert conn.execute("SELECT status FROM orders WHERE id='a'").fetchone()[0] == "refunded"
|
|
|
|
|
|
def test_quality_gate_rolls_back_and_alerts():
|
|
conn = sqlite3.connect(":memory:")
|
|
rows = make_rows(10) + make_rows(3, start=10, status="???") # 3/13 坏数据，超过 5%
|
|
alerts = []
|
|
r = run_sync(conn, FakeSource(rows), max_dead_ratio=0.05, alert=alerts.append)
|
|
assert not r.committed and alerts
|
|
assert conn.execute("SELECT COUNT(*) FROM orders").fetchone()[0] == 0 # 整批回滚
|
|
assert conn.execute("SELECT COUNT(*) FROM watermarks").fetchone()[0] == 0 # 水位线没动
|
|
|
|
|
|
def test_retry_only_retryable_and_respects_retry_after():
|
|
sleeps = []
|
|
calls = iter([RetryableError("429", retry_after=2.0), RetryableError("timeout"), ["ok"]])
|
|
|
|
def fn():
|
|
v = next(calls)
|
|
if isinstance(v, Exception):
|
|
raise v
|
|
return v
|
|
|
|
assert with_retry(fn, sleep=sleeps.append) == ["ok"]
|
|
assert sleeps[0] == 2.0 and 0 <= sleeps[1] <= 0.4
|
|
|
|
def fatal():
|
|
raise FatalError("401")
|
|
|
|
with pytest.raises(FatalError):
|
|
with_retry(fatal, sleep=sleeps.append)
|
|
|
|
|
|
def test_retry_gives_up():
|
|
def always():
|
|
raise RetryableError("503")
|
|
|
|
with pytest.raises(RetryableError):
|
|
with_retry(always, attempts=3, sleep=lambda s: None)
|
|
|
|
|
|
def test_token_bucket_limits_rate():
|
|
clock = [0.0]
|
|
slept = []
|
|
|
|
def sleep(s):
|
|
slept.append(s)
|
|
clock[0] += s
|
|
|
|
b = TokenBucket(rate=2, capacity=2, now=lambda: clock[0], sleep=sleep)
|
|
for _ in range(6):
|
|
b.acquire()
|
|
assert clock[0] == pytest.approx(2.0) # 2 个初始令牌 + 4 个以每秒 2 个的速度补充 = 2 秒
|