import time from unittest.mock import MagicMock import pytest from govcrawler.cookies import ( InMemoryCookieStore, get_default_store, reset_default_store, ) from govcrawler.cookies import store as store_mod # ---------- InMemory store ---------- @pytest.fixture(autouse=True) def _env(monkeypatch): monkeypatch.setenv("DB_URL", "postgresql+psycopg://x/x") monkeypatch.setenv("USER_AGENT", "TestBot/1.0") reset_default_store() yield reset_default_store() def test_inmemory_set_get_roundtrip(): s = InMemoryCookieStore() s.set("gdqy.gov.cn", {"ctct_session": "abc123", "x": "y"}) got = s.get("gdqy.gov.cn") assert got == {"ctct_session": "abc123", "x": "y"} def test_inmemory_case_insensitive_host(): s = InMemoryCookieStore() s.set("GDQY.Gov.Cn", {"a": "1"}) assert s.get("gdqy.gov.cn") == {"a": "1"} def test_inmemory_ttl_expires(monkeypatch): s = InMemoryCookieStore() s.set("x.com", {"a": "1"}, ttl_s=1) t0 = time.time() monkeypatch.setattr(time, "time", lambda: t0 + 5) assert s.get("x.com") is None def test_inmemory_invalidate(): s = InMemoryCookieStore() s.set("x.com", {"a": "1"}) s.invalidate("x.com") assert s.get("x.com") is None def test_inmemory_empty_cookies_noop(): s = InMemoryCookieStore() s.set("x.com", {}) assert s.get("x.com") is None # ---------- Backend selection ---------- def test_default_store_inmemory_when_valkey_url_empty(monkeypatch): monkeypatch.setenv("VALKEY_URL", "") reset_default_store() store = get_default_store() assert isinstance(store, InMemoryCookieStore) def test_default_store_valkey_when_url_set(monkeypatch): monkeypatch.setenv("VALKEY_URL", "redis://localhost:6379/0") reset_default_store() # Don't actually connect; redis-py's from_url is lazy store = get_default_store() assert store.__class__.__name__ == "ValkeyCookieStore" # ---------- ValkeyCookieStore degradation ---------- def test_valkey_get_network_error_returns_none(): s = store_mod.ValkeyCookieStore.__new__(store_mod.ValkeyCookieStore) fake_client = MagicMock() fake_client.get.side_effect = ConnectionError("down") s._client = fake_client assert s.get("x.com") is None def test_valkey_set_network_error_silently_drops(): s = store_mod.ValkeyCookieStore.__new__(store_mod.ValkeyCookieStore) fake_client = MagicMock() fake_client.set.side_effect = ConnectionError("down") s._client = fake_client # Must not raise — failing open is the whole point s.set("x.com", {"a": "1"}) # ---------- Chain invalidation on Tier-2 failure ---------- def test_chain_invalidates_on_fallback(monkeypatch): from govcrawler.fetcher import chain from govcrawler.fetcher.browser import FetchResult invalidated: list[str] = [] fake_store = MagicMock() fake_store.invalidate.side_effect = lambda h: invalidated.append(h) monkeypatch.setattr(chain, "get_default_store", lambda: fake_store) monkeypatch.setattr( chain, "fetch_html_http", lambda url, **kw: FetchResult( url=url, final_url=url, status=412, html="请稍候", fetched_at=0.0, duration_ms=5, is_challenge=True, strategy="httpx", ), ) monkeypatch.setattr( chain, "fetch_html_browser", lambda url, **kw: FetchResult( url=url, final_url=url, status=200, html="" + "x" * 1000 + "", fetched_at=0.0, duration_ms=50, is_challenge=False, strategy="playwright", ), ) fr = chain.fetch_html("https://www.gdqy.gov.cn/page") assert fr.strategy == "playwright" assert invalidated == ["www.gdqy.gov.cn"] def test_chain_no_invalidate_on_clean_tier2(monkeypatch): from govcrawler.fetcher import chain from govcrawler.fetcher.browser import FetchResult fake_store = MagicMock() monkeypatch.setattr(chain, "get_default_store", lambda: fake_store) monkeypatch.setattr( chain, "fetch_html_http", lambda url, **kw: FetchResult( url=url, final_url=url, status=200, html="" + "x" * 1000 + "", fetched_at=0.0, duration_ms=10, is_challenge=False, strategy="httpx", ), ) fr = chain.fetch_html("https://a.com/x") assert fr.strategy == "httpx" fake_store.invalidate.assert_not_called() # ---------- httpx injects cookies from pool ---------- def test_http_client_injects_pool_cookies(monkeypatch): from govcrawler.fetcher import http_client as hc from govcrawler.cookies import store as store_mod # Force in-memory store for the test — when the dev env happens to have # VALKEY_URL set, get_default_store() returns ValkeyCookieStore and the # test connects to a redis that may not be running. The behaviour we # actually verify (httpx forwards pool cookies into the Client kwargs) # is identical for both store implementations. store = store_mod.InMemoryCookieStore() monkeypatch.setattr(store_mod, "_default", store, raising=False) monkeypatch.setattr( "govcrawler.cookies.get_default_store", lambda: store, raising=False ) monkeypatch.setattr( "govcrawler.fetcher.http_client.get_default_store", lambda: store, raising=False, ) # Pre-load pool store.set("x.com", {"session": "ABCDEF"}) captured: dict = {} class _Ctx: def __enter__(self_inner): return self_inner def __exit__(self_inner, *a): pass def get(self_inner, url): resp = MagicMock() resp.status_code = 200 resp.text = "" + "x" * 1000 + "" resp.url = url resp.headers = {"content-type": "text/html"} return resp def fake_client(**kw): captured.update(kw) return _Ctx() monkeypatch.setattr(hc.httpx, "Client", fake_client) hc.fetch_html_http("https://x.com/page") assert captured.get("cookies") == {"session": "ABCDEF"}