86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""榜单快照：具名采集、失败保留失败态、观察差异不推断因果。"""
|
|
|
|
import pytest
|
|
|
|
from muse.资料研究.接口 import 资料服务
|
|
from muse.资料研究.模型 import 榜单读取结果, 资料错误
|
|
|
|
pytestmark = pytest.mark.数据库
|
|
|
|
|
|
class 合成榜单读取器:
|
|
"""按剧本返回的合成读取器；不访问网络。"""
|
|
|
|
def __init__(self, 剧本: list[榜单读取结果]):
|
|
self.剧本 = list(剧本)
|
|
|
|
def 抓取(self, url: str, 方式: str) -> 榜单读取结果:
|
|
return self.剧本.pop(0)
|
|
|
|
|
|
@pytest.fixture
|
|
def 服务():
|
|
return 资料服务()
|
|
|
|
|
|
def test_采集成功失败与观察差异__17b01e(服务, 应用测试库):
|
|
from muse.共享.调用身份 import 用途
|
|
|
|
库 = 应用测试库[用途.生产]
|
|
with 库.连接() as 连, 连.transaction():
|
|
服务.登记榜单(连, "rank-qd", "起点月票榜", "https://example.com/rank", "html")
|
|
# 第一次：失败保留失败态，不伪造数据。
|
|
失败 = 服务.采集榜单快照(
|
|
连, "rank-qd", 合成榜单读取器([榜单读取结果(成功=False, 失败原因="上游 503")])
|
|
)
|
|
assert 失败["status"] == "failed" and 失败["failure_reason"] == "上游 503"
|
|
快照集 = 服务.列出快照(连, "rank-qd")
|
|
assert 快照集[0]["status"] == "failed" and 快照集[0]["items"] is None
|
|
# 第二、三次：成功快照带时刻与条目。
|
|
第一 = 服务.采集榜单快照(
|
|
连,
|
|
"rank-qd",
|
|
合成榜单读取器(
|
|
[
|
|
榜单读取结果(
|
|
成功=True,
|
|
items=(
|
|
{"作品": "甲", "名次": 1},
|
|
{"作品": "乙", "名次": 2},
|
|
),
|
|
)
|
|
]
|
|
),
|
|
)
|
|
第二 = 服务.采集榜单快照(
|
|
连,
|
|
"rank-qd",
|
|
合成榜单读取器(
|
|
[
|
|
榜单读取结果(
|
|
成功=True,
|
|
items=(
|
|
{"作品": "乙", "名次": 1},
|
|
{"作品": "丙", "名次": 2},
|
|
{"作品": "甲", "名次": 3},
|
|
),
|
|
)
|
|
]
|
|
),
|
|
)
|
|
assert 第一["status"] == "ok" and 第二["条目数"] == 3
|
|
差异 = 服务.快照观察差异(连, "rank-qd")
|
|
变化 = {d["作品"]: d["变化"] for d in 差异["差异"]}
|
|
assert 变化["乙"] == 1 and 变化["甲"] == -2 and 变化["丙"] is None
|
|
# 观察差异不携带任何 run/效果归因字段。
|
|
assert all("run" not in str(键).lower() for 键 in 差异)
|
|
|
|
|
|
def test_未登记榜单拒绝采集__17b02f(服务, 应用测试库):
|
|
from muse.共享.调用身份 import 用途
|
|
|
|
库 = 应用测试库[用途.生产]
|
|
with 库.连接() as 连:
|
|
with pytest.raises(资料错误, match="榜单未登记"):
|
|
服务.采集榜单快照(连, "rank-none", 合成榜单读取器([]))
|