216 lines
7.6 KiB
Python
216 lines
7.6 KiB
Python
"""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
|
||
|
||
|
||
class FakeProvider:
|
||
"""测试用 provider：URL 带 key、余额在 data.balance。"""
|
||
|
||
method = "GET"
|
||
|
||
def __init__(self, path="data.balance"):
|
||
self.path = path
|
||
|
||
def build_request(self, api_key):
|
||
return (
|
||
"https://x.test/api?key=" + api_key,
|
||
{"Authorization": "Bearer " + api_key},
|
||
None,
|
||
)
|
||
|
||
def extract_balance(self, data):
|
||
from app.expr import evaluate_balance
|
||
|
||
return evaluate_balance(data, self.path)
|
||
|
||
|
||
class FakePostProvider(FakeProvider):
|
||
method = "POST"
|
||
|
||
def build_request(self, api_key):
|
||
return ("https://x.test/api", {}, '{"api_key": "' + api_key + '"}')
|
||
|
||
|
||
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(FakeProvider(), "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):
|
||
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(FakePostProvider(), "sk-2", retry_count=2, timeout=10)
|
||
assert result.ok
|
||
assert captured["data"] == '{"api_key": "sk-2"}'
|
||
|
||
def test_post_content_type_auto(self, monkeypatch):
|
||
captured = {}
|
||
|
||
def fake_post(url, headers=None, data=None, timeout=10):
|
||
captured["headers"] = headers
|
||
return FakeResponse(200, json_data={"data": {"balance": 1}})
|
||
|
||
monkeypatch.setattr("requests.post", fake_post)
|
||
fetch_balance(FakePostProvider(), "sk-2", retry_count=0, timeout=10)
|
||
assert captured["headers"].get("Content-Type") == "application/json"
|
||
|
||
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(FakeProvider(), "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(FakeProvider(), "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(FakeProvider(), "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(FakeProvider(), "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(FakeProvider(), "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(FakeProvider(), "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(FakeProvider(), "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="<html>"))
|
||
result = fetch_balance(FakeProvider(), "k", retry_count=0, timeout=10)
|
||
assert not result.ok
|
||
assert "JSON" in result.error
|