q792602257andClaude Opus 5
e2875f0c00
- try_relogin 的并发去重原本读 status().logged_in，但重登成功后的 reload() 会把它重置成 None，排队在 site 锁上的调用方一律判「还没人登上」，N 个并发 调用会串行触发 N 次真实登录（各自最长 relogin_timeout）。改用 _relogin_epochs 计数：等锁期间 epoch 变过就真探测一次，已登录即跳过；仍未登录说明这轮站点侧 就是登不上（验证码/密码错/风控），直接失败，不在同一波并发里重复触发。 原并发测试的桩自相矛盾（login_one 返回成功、探针页始终回未登录），断言只能 松到 count >= 1；桩改为登录成功时翻转探针页，断言收紧到 count == 1。 - check_order_status / list_recent_orders 执行中掉登录此前会被静默吞掉：订单页 被踢到 SSO 后既不报错也没订单号，_parse_order_status 返回 found=False 被 _monitor_order 当成「订单还没反映出来」继续轮询（默认 3 小时一轮）， _parse_order_list 则退化成空列表让 verify_on_site 转 unknown 卡住等人工。 新增 SiteInteractor._read_with_relogin_retry 外壳：只读操作中途判定掉登录时 重登一次并整个重跑，第二次仍失败抛 NotLoggedInError。只给读操作用——写操作 中途掉登录不能重跑（上次动作可能已在站点侧生效），这条边界在两边文档里写明。 - 判据是新增的 auth_site.looks_logged_out：与探针页上权威的 is_logged_in 分开， 它是业务页上的单边启发式（返回 False 不代表登录着），只用于「判错最多多花一次 重登」的重试决策。刻意排除 session/upgrade——那是已登录时的站点风控复核密码， 不是 cookie 过期，误判会把风控当掉登录去重登。 判据里「掉登录会跳到 SSO 域」这一步没有真实探测证据（要复现得先让一份真实登录态 过期），是按站点通行行为的推断，已在常量注释标注；新增测试用替身页面，不是真实 站点 HTML。399 测试全绿（仓库未配 ruff/flake8/mypy，只跑了 pytest）。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
454 lines
17 KiB
Python
454 lines
17 KiB
Python
"""登录态会话测试：cookie 加载、失效探测、重新加载、自动重登
|
|
|
|
全部用 httpx.MockTransport 拦截，不触达真实站点、不需要真实账号。
|
|
登录态文件写在 tmp_path 下，不碰仓库里的 .auth/。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.shared.config import Settings
|
|
from app.shared.errors import NotLoggedInError, UpstreamRequestError
|
|
from app.trading.core import auth_site
|
|
from app.trading.services import auth_session as auth_session_mod
|
|
from app.trading.services.auth_session import AuthSession
|
|
|
|
# 购物车页的两种形态：含未登录标记 = 未登录，不含 = 已登录
|
|
CART_LOGGED_OUT = f"<html><body>買い物かご {auth_site.RAKUTEN_LOGGED_OUT_MARKER}</body></html>"
|
|
CART_LOGGED_IN = "<html><body>買い物かご 商品が1点入っています</body></html>"
|
|
|
|
|
|
def make_settings(tmp_path: Path, **overrides) -> Settings:
|
|
base = {
|
|
"auth_state_dir": str(tmp_path / "auth"),
|
|
"request_timeout_seconds": 5.0,
|
|
}
|
|
base.update(overrides)
|
|
return Settings(**base)
|
|
|
|
|
|
def write_state(settings: Settings, site: str, cookies: list[dict]) -> Path:
|
|
"""伪造一份 Playwright storage_state 落盘"""
|
|
path = settings.auth_state_path / auth_site.profile(site).state_filename
|
|
path.write_text(
|
|
json.dumps({"cookies": cookies, "origins": []}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
async def build_session(settings: Settings, handler) -> AuthSession:
|
|
"""构建 AuthSession 并把 client 换成 MockTransport 版本
|
|
|
|
换掉 client 会丢掉 start() 时灌进去的 cookie，因此重新走一次 reload
|
|
把登录态读回新客户端；reload 同时清空探测缓存，正好是测试想要的干净起点。
|
|
"""
|
|
session = AuthSession(settings)
|
|
await session.start()
|
|
for name in session.sites:
|
|
auth = session._sites[name]
|
|
await auth.client.aclose()
|
|
auth.client = httpx.AsyncClient(
|
|
transport=httpx.MockTransport(handler),
|
|
follow_redirects=True,
|
|
)
|
|
session.reload(name)
|
|
return session
|
|
|
|
|
|
# ---- cookie 加载 ----
|
|
|
|
|
|
async def test_loads_cookies_from_state_file(tmp_path):
|
|
"""storage_state 里的 cookie 应被灌进 httpx 客户端"""
|
|
settings = make_settings(tmp_path)
|
|
write_state(
|
|
settings,
|
|
"rakuten",
|
|
[{"name": "SESSION", "value": "abc", "domain": ".rakuten.co.jp", "path": "/"}],
|
|
)
|
|
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_IN))
|
|
try:
|
|
names = {c.name for c in session.client("rakuten").cookies.jar}
|
|
assert "SESSION" in names
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_starts_without_state_file(tmp_path):
|
|
"""登录态文件不存在时仍能启动，只是报告未登录——服务不应因此起不来"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
status = session.status("rakuten")
|
|
assert status.state_file_exists is False
|
|
assert status.logged_in is None # 尚未探测
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_corrupted_state_file_is_tolerated(tmp_path):
|
|
"""登录态文件损坏时降级为无 cookie，不抛异常"""
|
|
settings = make_settings(tmp_path)
|
|
path = settings.auth_state_path / auth_site.profile("rakuten").state_filename
|
|
path.write_text("{ not json", encoding="utf-8")
|
|
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
assert not list(session.client("rakuten").cookies.jar)
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
# ---- 登录态探测 ----
|
|
|
|
|
|
async def test_rakuten_detects_logged_out_by_marker(tmp_path):
|
|
"""购物车页出现未登录文案即判定未登录"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
status = await session.check("rakuten")
|
|
assert status.logged_in is False
|
|
assert "未登录" in status.detail
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_rakuten_detects_logged_in(tmp_path):
|
|
"""购物车页没有未登录文案即判定已登录"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_IN))
|
|
try:
|
|
status = await session.check("rakuten")
|
|
assert status.logged_in is True
|
|
assert status.checked_at is not None
|
|
# 序列化后暴露给 API 的是相对时长，不是单调时钟原值
|
|
assert status.to_dict()["checked_age_seconds"] is not None
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_error_status_counts_as_logged_out(tmp_path):
|
|
"""探测页返回 4xx/5xx 时保守判定为未登录，不放行下单"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(503, text="oops"))
|
|
try:
|
|
status = await session.check("rakuten")
|
|
assert status.logged_in is False
|
|
assert "503" in status.detail
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_network_error_raises_upstream(tmp_path):
|
|
"""网络异常与「确实未登录」是两回事，应抛错而不是静默判未登录"""
|
|
settings = make_settings(tmp_path)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("boom")
|
|
|
|
session = await build_session(settings, handler)
|
|
try:
|
|
with pytest.raises(UpstreamRequestError):
|
|
await session.check("rakuten")
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
# ---- 下单前置校验 ----
|
|
|
|
|
|
async def test_require_logged_in_raises_when_logged_out(tmp_path):
|
|
"""relogin_enabled=False 时，未登录直接抛 5001，且标记为不可重试
|
|
|
|
登录需要人工过验证码，自动重试没有意义，必须让上游停下来。
|
|
"""
|
|
settings = make_settings(tmp_path, relogin_enabled=False)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
with pytest.raises(NotLoggedInError) as excinfo:
|
|
await session.require_logged_in("rakuten")
|
|
assert excinfo.value.err_code == 5001
|
|
assert excinfo.value.retryable is False
|
|
assert excinfo.value.status_code == 401
|
|
assert "scripts/login.py" in excinfo.value.message
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_require_logged_in_passes_when_logged_in(tmp_path):
|
|
"""已登录时放行，不抛异常"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_IN))
|
|
try:
|
|
await session.require_logged_in("rakuten") # 不应抛出
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
# ---- 自动重登 ----
|
|
|
|
|
|
def _patch_relogin(monkeypatch, *, return_value: bool, sleep_quick: bool = True):
|
|
"""把 login_runner.login_one 替换成可控桩
|
|
|
|
login_one 在 AuthSession.try_relogin 里通过 from import 触发，所以 patch
|
|
的对象要是 login_runner 模块上的 login_one。
|
|
"""
|
|
calls: list[dict] = []
|
|
|
|
async def fake_login_one(account, settings, *, timeout_seconds=..., progress=None):
|
|
calls.append({
|
|
"site": account.site,
|
|
"account_id": account.id,
|
|
"username": account.username, # 仅测试中校验不泄密给日志
|
|
"timeout_seconds": timeout_seconds,
|
|
})
|
|
if sleep_quick:
|
|
await asyncio.sleep(0) # 让协程调度
|
|
return return_value
|
|
|
|
# AuthSession.try_relogin 内部用 `from ... import login_runner` 形式
|
|
from app.trading.services import login_runner
|
|
monkeypatch.setattr(login_runner, "login_one", fake_login_one)
|
|
return calls
|
|
|
|
|
|
def _patch_accounts_file(monkeypatch, tmp_path, *, has_rakuten: bool = True):
|
|
"""让 login_runner 读到一份假的 account.yaml"""
|
|
accounts = {"rakuten": []} if has_rakuten else {}
|
|
if has_rakuten:
|
|
accounts["rakuten"].append({
|
|
"id": "testacct",
|
|
"username": "u@example.com",
|
|
"password": "pw",
|
|
"user_data_dir": str(tmp_path / "ud"),
|
|
"state_filename": "rakuten_state.json",
|
|
"default": True,
|
|
})
|
|
yaml_text = "".join(
|
|
f"{site}:\n" + "".join(
|
|
f" - id: {r['id']}\n username: {r['username']}\n password: {r['password']}\n"
|
|
f" user_data_dir: {r['user_data_dir']}\n state_filename: {r['state_filename']}\n"
|
|
f" default: true\n"
|
|
for r in recs
|
|
)
|
|
for site, recs in accounts.items()
|
|
)
|
|
from app.trading.services import login_runner
|
|
fake_path = tmp_path / "account.yaml"
|
|
fake_path.write_text(yaml_text, encoding="utf-8")
|
|
monkeypatch.setattr(login_runner, "accounts_file_path", lambda: fake_path)
|
|
|
|
|
|
async def test_require_logged_in_tries_relogin_and_passes(tmp_path, monkeypatch):
|
|
"""relogin_enabled=True：失效 → login_one 成功 → check 通过 → 放行
|
|
|
|
login_one 模拟重登成功：落盘一份新 storage_state，并把 MockTransport 切到
|
|
「已登录」分支，让 require_logged_in 第二次 check 通过。
|
|
"""
|
|
_patch_accounts_file(monkeypatch, tmp_path)
|
|
|
|
state = {"logged_in": False}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if state["logged_in"]:
|
|
return httpx.Response(200, text=CART_LOGGED_IN)
|
|
return httpx.Response(200, text=CART_LOGGED_OUT)
|
|
|
|
settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=5)
|
|
session = await build_session(settings, handler)
|
|
|
|
from app.trading.services import login_runner
|
|
|
|
async def fake_login_one(account, s, *, timeout_seconds=300, progress=None):
|
|
# 模拟重登成功：落盘新 storage_state（让 reload 能读到），翻转 check 行为
|
|
write_state(
|
|
settings,
|
|
"rakuten",
|
|
[{"name": "FRESH", "value": "xyz", "domain": ".rakuten.co.jp", "path": "/"}],
|
|
)
|
|
state["logged_in"] = True
|
|
return True
|
|
|
|
monkeypatch.setattr(login_runner, "login_one", fake_login_one)
|
|
|
|
try:
|
|
await session.require_logged_in("rakuten") # 不应抛出
|
|
assert session.status("rakuten").logged_in is True
|
|
# 重登后的 cookie 已灌进 client
|
|
names = {c.name for c in session.client("rakuten").cookies.jar}
|
|
assert "FRESH" in names
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_require_logged_in_raises_when_relogin_fails(tmp_path, monkeypatch):
|
|
"""relogin_enabled=True 但 login_one 失败 → 抛 NotLoggedInError"""
|
|
_patch_accounts_file(monkeypatch, tmp_path)
|
|
_patch_relogin(monkeypatch, return_value=False)
|
|
|
|
settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=5)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
with pytest.raises(NotLoggedInError):
|
|
await session.require_logged_in("rakuten")
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_require_logged_in_falls_back_when_no_accounts_file(tmp_path, monkeypatch):
|
|
"""relogin_enabled=True 但 account.yaml 不存在 → try_relogin 返回 False，抛 NotLoggedInError"""
|
|
from app.trading.services import login_runner
|
|
monkeypatch.setattr(login_runner, "accounts_file_path", lambda: tmp_path / "missing.yaml")
|
|
|
|
settings = make_settings(tmp_path, relogin_enabled=True)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
with pytest.raises(NotLoggedInError):
|
|
await session.require_logged_in("rakuten")
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_try_relogin_is_disabled_when_flag_off(tmp_path):
|
|
"""relogin_enabled=False：try_relogin 直接返回 False，不读 account.yaml"""
|
|
settings = make_settings(tmp_path, relogin_enabled=False)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
ok = await session.try_relogin("rakuten")
|
|
assert ok is False
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_try_relogin_serializes_concurrent_calls_same_site(tmp_path, monkeypatch):
|
|
"""同 site 并发触发**只跑一次** login_one：锁串行化 + 排队者复核登录态后跳过
|
|
|
|
不能只断言「没并发执行」：重登成功后的 reload() 会把 logged_in 重置成 None，
|
|
排队者如果读缓存去重就会一律判「还没人登上」，于是 N 个并发调用串行触发 N 次
|
|
真实登录（每次最长 relogin_timeout）。这里断言 count == 1 就是钉住这一点。
|
|
"""
|
|
_patch_accounts_file(monkeypatch, tmp_path)
|
|
|
|
invocations = {"count": 0, "in_flight_max": 0, "current": 0}
|
|
state = {"logged_in": False}
|
|
from app.trading.services import login_runner
|
|
|
|
async def counting_login_one(account, s, *, timeout_seconds=300, progress=None):
|
|
invocations["current"] += 1
|
|
invocations["in_flight_max"] = max(invocations["in_flight_max"], invocations["current"])
|
|
await asyncio.sleep(0.05) # 故意拉长，让并发请求有机会叠上来
|
|
invocations["current"] -= 1
|
|
invocations["count"] += 1
|
|
# 落盘新 storage_state，让 reload 有东西可读
|
|
write_state(
|
|
make_settings(tmp_path), # 用同一路径
|
|
"rakuten",
|
|
[{"name": "X", "value": "1", "domain": ".rakuten.co.jp", "path": "/"}],
|
|
)
|
|
# 登录成功后探针页也必须随之翻转，否则桩自相矛盾（login_one 说成功、站点说
|
|
# 没登上），测不出真实行为
|
|
state["logged_in"] = True
|
|
return True
|
|
|
|
monkeypatch.setattr(login_runner, "login_one", counting_login_one)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
text = CART_LOGGED_IN if state["logged_in"] else CART_LOGGED_OUT
|
|
return httpx.Response(200, text=text)
|
|
|
|
settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=10)
|
|
session = await build_session(settings, handler)
|
|
|
|
try:
|
|
# 第一次状态为 logged_in=None，三个并发都进入 try_relogin
|
|
results = await asyncio.gather(
|
|
session.try_relogin("rakuten"),
|
|
session.try_relogin("rakuten"),
|
|
session.try_relogin("rakuten"),
|
|
)
|
|
assert invocations["count"] == 1
|
|
# login_one 永远没并发执行（同账号同 user_data_dir，撞锁会失败）
|
|
assert invocations["in_flight_max"] == 1
|
|
# 三个调用方都拿到「已登录」结论：一个真登录，两个复核后跳过
|
|
assert results == [True, True, True]
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_try_relogin_does_not_retry_after_queued_failure(tmp_path, monkeypatch):
|
|
"""一次重登失败后，排在锁上的并发调用不再重复触发同一个注定失败的登录
|
|
|
|
站点侧登不上（验证码/密码错/风控）时，把每个调用方各卡一个 relogin_timeout
|
|
不会改变结果，只会让整批任务慢几倍。
|
|
"""
|
|
_patch_accounts_file(monkeypatch, tmp_path)
|
|
|
|
calls = {"count": 0}
|
|
from app.trading.services import login_runner
|
|
|
|
async def failing_login_one(account, s, *, timeout_seconds=300, progress=None):
|
|
calls["count"] += 1
|
|
await asyncio.sleep(0.05) # 让其余并发调用叠到锁上
|
|
return False
|
|
|
|
monkeypatch.setattr(login_runner, "login_one", failing_login_one)
|
|
|
|
settings = make_settings(tmp_path, relogin_enabled=True, relogin_timeout_seconds=10)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_OUT))
|
|
try:
|
|
results = await asyncio.gather(
|
|
session.try_relogin("rakuten"),
|
|
session.try_relogin("rakuten"),
|
|
session.try_relogin("rakuten"),
|
|
)
|
|
assert results == [False, False, False]
|
|
assert calls["count"] == 1
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
# ---- 重新加载 ----
|
|
|
|
|
|
async def test_reload_picks_up_new_cookies(tmp_path):
|
|
"""人工重新登录后 reload 应换上新 cookie 并清掉旧的探测结论"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_IN))
|
|
try:
|
|
await session.check("rakuten")
|
|
assert session.status("rakuten").logged_in is True
|
|
|
|
write_state(
|
|
settings,
|
|
"rakuten",
|
|
[{"name": "FRESH", "value": "xyz", "domain": ".rakuten.co.jp", "path": "/"}],
|
|
)
|
|
count = session.reload("rakuten")
|
|
|
|
assert count == 1
|
|
assert "FRESH" in {c.name for c in session.client("rakuten").cookies.jar}
|
|
# 重载后旧结论必须作废，避免拿过期判断放行下单
|
|
assert session.status("rakuten").logged_in is None
|
|
finally:
|
|
await session.close()
|
|
|
|
|
|
async def test_unknown_site_rejected(tmp_path):
|
|
"""未知站点名应明确报错，不静默返回空状态"""
|
|
settings = make_settings(tmp_path)
|
|
session = await build_session(settings, lambda r: httpx.Response(200, text=CART_LOGGED_IN))
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
session.status("mercari")
|
|
finally:
|
|
await session.close()
|