"""cookie 自动续期(2026-09-03):缓存优先级 + 撞过期自动重登重跑。 机制::func:`_load_cookie` 优先读缓存文件(自动续期成功时写入, MEDIA_DIR/.yk_cookie_cache)——比 compose 注入的 YK_COOKIE 新;实体运行撞 :class:`CookieExpiredError` 时 :func:`_run_with_auto_relogin` 调浏览器表单 重登拿新 cookie、落缓存、整实体重跑一次(幂等,水位增量天然安全)。 二次过期不再重试(防死循环)。 """ from __future__ import annotations import os import pytest from scripts import scrape_via_page as sv @pytest.fixture def cache_file(tmp_path, monkeypatch): """把 cookie 缓存指到临时目录。""" path = tmp_path / ".yk_cookie_cache" monkeypatch.setenv("YK_COOKIE_CACHE", str(path)) return path # ── 缓存:读/写/优先级 ────────────────────────────────────────────────────── # def test_load_cookie_prefers_cache_over_env(cache_file, monkeypatch): monkeypatch.setenv("YK_COOKIE", "env-cookie") cache_file.write_text("cache-cookie", encoding="utf-8") assert sv._load_cookie() == "cache-cookie" def test_load_cookie_falls_back_to_env_when_no_cache(cache_file, monkeypatch): monkeypatch.setenv("YK_COOKIE", "env-cookie") assert sv._load_cookie() == "env-cookie" def test_save_then_load_roundtrip(cache_file): sv._save_cookie_cache("fresh-cookie") assert cache_file.read_text(encoding="utf-8").strip() == "fresh-cookie" assert sv._load_cookie() == "fresh-cookie" def test_save_cache_unwritable_path_does_not_raise(monkeypatch): # 持久化是尽力而为:路径不可写(本地无挂载等)不能炸,回退 env 语义仍在 monkeypatch.setenv("YK_COOKIE_CACHE", "/nonexistent-root/x/y/.cookie") sv._save_cookie_cache("whatever") # 不应抛 # ── 撞过期自动重登重跑 ───────────────────────────────────────────────────── # async def test_retry_after_relogin(cache_file): calls: list[str] = [] async def entity(cookie: str) -> str: calls.append(cookie) if cookie == "old": raise sv.CookieExpiredError("云客 cookie 已过期") return f"ok-with-{cookie}" async def fake_relogin() -> str: return "new" result = await sv._run_with_auto_relogin(entity, "old", relogin=fake_relogin) assert result == "ok-with-new" assert calls == ["old", "new"] # 旧 cookie 失败一次,新 cookie 重跑成功 assert sv._load_cookie() == "new" # 新 cookie 已落缓存 async def test_no_relogin_when_other_error(cache_file): async def entity(cookie: str): raise RuntimeError("业务炸了") async def fake_relogin() -> str: raise AssertionError("不应触发重登") with pytest.raises(RuntimeError, match="业务炸了"): await sv._run_with_auto_relogin(entity, "old", relogin=fake_relogin) async def test_second_expiry_propagates_no_loop(cache_file): """重登后仍过期 → 直接抛出,绝不无限循环。""" async def always_expired(cookie: str): raise sv.CookieExpiredError("还是过期") async def fake_relogin() -> str: return "renewed" with pytest.raises(sv.CookieExpiredError): await sv._run_with_auto_relogin(always_expired, "old", relogin=fake_relogin) async def test_relogin_failure_propagates(cache_file): async def entity(cookie: str): raise sv.CookieExpiredError("过期") async def failing_relogin() -> str: raise sv.CookieExpiredError("自动重登失败(无账密/验证码)") with pytest.raises(sv.CookieExpiredError, match="自动重登失败"): await sv._run_with_auto_relogin(entity, "old", relogin=failing_relogin)