"""fetcher 核心逻辑测试：路径提取、模板渲染、请求/重试策略。""" import pytest from app.fetcher import extract_balance, fetch_balance, render_template # ---------- extract_balance ---------- class TestExtractBalance: def test_dot_path(self): assert extract_balance({"data": {"balance": 12.5}}, "data.balance") == 12.5 def test_array_index(self): assert extract_balance({"data": [{"balance": 3.0}]}, "data[0].balance") == 3.0 def test_dollar_prefix(self): assert extract_balance({"a": {"b": 7}}, "$.a.b") == 7 def test_root_key(self): assert extract_balance({"balance": 1}, "balance") == 1 def test_nested_array(self): data = {"list": [[1, 2], [3, 4]]} assert extract_balance(data, "list[1][0]") == 3 def test_string_number(self): assert extract_balance({"data": {"balance": "9.99"}}, "data.balance") == 9.99 def test_missing_key_raises(self): with pytest.raises(ValueError): extract_balance({"data": {}}, "data.balance") def test_index_out_of_range_raises(self): with pytest.raises(ValueError): extract_balance({"data": []}, "data[0].balance") def test_non_numeric_raises(self): with pytest.raises(ValueError): extract_balance({"data": {"balance": "abc"}}, "data.balance") def test_bool_rejected(self): with pytest.raises(ValueError): extract_balance({"data": {"balance": True}}, "data.balance") # ---------- render_template ---------- class TestRenderTemplate: def test_replace(self): assert render_template("https://x?key={{apiKey}}&a=1", "sk-123") == "https://x?key=sk-123&a=1" def test_multiple(self): assert render_template("{{apiKey}}/{{apiKey}}", "k") == "k/k" # ---------- fetch_balance（mock requests） ---------- class FakeResponse: def __init__(self, status_code=200, text="", json_data=None): self.status_code = status_code self.text = text self._json = json_data def json(self): if self._json is None: raise ValueError("No JSON object could be decoded") return self._json PLATFORM_GET = { "method": "GET", "url": "https://x.test/api?key={{apiKey}}", "headers": {"Authorization": "Bearer {{apiKey}}"}, "body": "", "balance_path": "data.balance", } class TestFetchBalance: def test_get_success(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append((url, headers)) assert url == "https://x.test/api?key=sk-1" assert headers["Authorization"] == "Bearer sk-1" return FakeResponse(200, json_data={"data": {"balance": 42.5}}) monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "sk-1", retry_count=2, timeout=10) assert result.ok and result.balance == 42.5 assert len(calls) == 1 def test_post_with_body(self, monkeypatch): platform = dict(PLATFORM_GET, method="POST", body='{"api_key": "{{apiKey}}"}') captured = {} def fake_post(url, headers=None, data=None, timeout=10): captured["data"] = data return FakeResponse(200, json_data={"data": {"balance": 1}}) monkeypatch.setattr("requests.post", fake_post) result = fetch_balance(platform, "sk-2", retry_count=2, timeout=10) assert result.ok assert captured["data"] == '{"api_key": "sk-2"}' def test_401_retries_then_auth_error(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append(1) return FakeResponse(401, text="unauthorized") monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "bad", retry_count=2, timeout=10) assert not result.ok and result.auth_error assert len(calls) == 3 # 401 也按重试次数确认后再判定 def test_transient_401_then_success(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append(1) return FakeResponse(401) if len(calls) == 1 else FakeResponse(200, json_data={"data": {"balance": 6.6}}) monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "k", retry_count=2, timeout=10) assert result.ok and result.balance == 6.6 assert len(calls) == 2 # 瞬时 401 重试后成功，不误判禁用 def test_5xx_retries_then_success(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append(1) return FakeResponse(500) if len(calls) < 3 else FakeResponse(200, json_data={"data": {"balance": 5}}) monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "k", retry_count=2, timeout=10) assert result.ok and result.balance == 5 assert len(calls) == 3 def test_5xx_all_fail(self, monkeypatch): monkeypatch.setattr("requests.get", lambda *a, **k: FakeResponse(503)) result = fetch_balance(PLATFORM_GET, "k", retry_count=2, timeout=10) assert not result.ok and not result.auth_error def test_network_error_retries(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append(1) import requests if len(calls) < 3: raise requests.ConnectionError("boom") return FakeResponse(200, json_data={"data": {"balance": 8}}) monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "k", retry_count=2, timeout=10) assert result.ok and result.balance == 8 def test_other_4xx_no_retry(self, monkeypatch): calls = [] def fake_get(url, headers=None, timeout=10): calls.append(1) return FakeResponse(404, text="not found") monkeypatch.setattr("requests.get", fake_get) result = fetch_balance(PLATFORM_GET, "k", retry_count=2, timeout=10) assert not result.ok and not result.auth_error assert "404" in result.error assert len(calls) == 1 def test_bad_json_path(self, monkeypatch): monkeypatch.setattr("requests.get", lambda *a, **k: FakeResponse(200, json_data={"x": 1})) result = fetch_balance(PLATFORM_GET, "k", retry_count=0, timeout=10) assert not result.ok assert "路径不存在" in result.error def test_invalid_json_body(self, monkeypatch): monkeypatch.setattr("requests.get", lambda *a, **k: FakeResponse(200, text="")) result = fetch_balance(PLATFORM_GET, "k", retry_count=0, timeout=10) assert not result.ok assert "JSON" in result.error