File size: 13,231 Bytes
eebf5c4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 |
"""
Test suite for chart endpoints
Validates rate limit history and freshness history endpoints
"""
import pytest
import requests as R
from datetime import datetime, timedelta
# Base URL for API (adjust if running on different port/host)
BASE = "http://localhost:7860"
class TestRateLimitHistory:
"""Test suite for /api/charts/rate-limit-history endpoint"""
def test_rate_limit_default(self):
"""Test rate limit history with default parameters"""
r = R.get(f"{BASE}/api/charts/rate-limit-history")
r.raise_for_status()
data = r.json()
# Validate response structure
assert isinstance(data, list), "Response should be a list"
if len(data) > 0:
# Validate first series object
s = data[0]
assert "provider" in s, "Series should have provider field"
assert "hours" in s, "Series should have hours field"
assert "series" in s, "Series should have series field"
assert "meta" in s, "Series should have meta field"
# Validate hours field
assert s["hours"] == 24, "Default hours should be 24"
# Validate series points
assert isinstance(s["series"], list), "series should be a list"
assert len(s["series"]) == 24, "Should have 24 data points for 24 hours"
# Validate each point
for point in s["series"]:
assert "t" in point, "Point should have timestamp (t)"
assert "pct" in point, "Point should have percentage (pct)"
assert 0 <= point["pct"] <= 100, f"Percentage should be 0-100, got {point['pct']}"
# Validate timestamp format
try:
datetime.fromisoformat(point["t"].replace('Z', '+00:00'))
except ValueError:
pytest.fail(f"Invalid timestamp format: {point['t']}")
# Validate meta
meta = s["meta"]
assert "limit_type" in meta, "Meta should have limit_type"
assert "limit_value" in meta, "Meta should have limit_value"
def test_rate_limit_48h_subset(self):
"""Test rate limit history with custom time range and provider selection"""
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"hours": 48, "providers": "coingecko,cmc"}
)
r.raise_for_status()
data = r.json()
assert isinstance(data, list), "Response should be a list"
assert len(data) <= 2, "Should have at most 2 providers (coingecko, cmc)"
for series in data:
assert series["hours"] == 48, "Should have 48 hours of data"
assert len(series["series"]) == 48, "Should have 48 data points"
assert series["provider"] in ["coingecko", "cmc"], "Provider should match requested"
def test_rate_limit_hours_clamping(self):
"""Test that hours parameter is properly clamped to valid range"""
# Test lower bound (should clamp to 1)
r = R.get(f"{BASE}/api/charts/rate-limit-history", params={"hours": 0})
assert r.status_code in [200, 422], "Should handle hours=0"
# Test upper bound (should clamp to 168)
r = R.get(f"{BASE}/api/charts/rate-limit-history", params={"hours": 999})
assert r.status_code in [200, 422], "Should handle hours=999"
def test_rate_limit_invalid_provider(self):
"""Test rejection of invalid provider names"""
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": "invalid_provider_xyz"}
)
# Should return 400 for invalid provider
assert r.status_code in [400, 404], "Should reject invalid provider names"
def test_rate_limit_max_providers(self):
"""Test that provider list is limited to max 5"""
# Request more than 5 providers
providers_list = ",".join([f"provider{i}" for i in range(10)])
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": providers_list}
)
# Should either succeed with max 5 or reject invalid providers
if r.status_code == 200:
data = r.json()
assert len(data) <= 5, "Should limit to max 5 providers"
def test_rate_limit_response_time(self):
"""Test that endpoint responds within performance target (< 200ms for 24h)"""
import time
start = time.time()
r = R.get(f"{BASE}/api/charts/rate-limit-history")
duration_ms = (time.time() - start) * 1000
r.raise_for_status()
# Allow 500ms for dev environment (more generous than production target)
assert duration_ms < 500, f"Response took {duration_ms:.0f}ms (target < 500ms)"
class TestFreshnessHistory:
"""Test suite for /api/charts/freshness-history endpoint"""
def test_freshness_default(self):
"""Test freshness history with default parameters"""
r = R.get(f"{BASE}/api/charts/freshness-history")
r.raise_for_status()
data = r.json()
# Validate response structure
assert isinstance(data, list), "Response should be a list"
if len(data) > 0:
# Validate first series object
s = data[0]
assert "provider" in s, "Series should have provider field"
assert "hours" in s, "Series should have hours field"
assert "series" in s, "Series should have series field"
assert "meta" in s, "Series should have meta field"
# Validate hours field
assert s["hours"] == 24, "Default hours should be 24"
# Validate series points
assert isinstance(s["series"], list), "series should be a list"
assert len(s["series"]) == 24, "Should have 24 data points for 24 hours"
# Validate each point
for point in s["series"]:
assert "t" in point, "Point should have timestamp (t)"
assert "staleness_min" in point, "Point should have staleness_min"
assert "ttl_min" in point, "Point should have ttl_min"
assert "status" in point, "Point should have status"
assert point["staleness_min"] >= 0, "Staleness should be non-negative"
assert point["ttl_min"] > 0, "TTL should be positive"
assert point["status"] in ["fresh", "aging", "stale"], f"Invalid status: {point['status']}"
# Validate timestamp format
try:
datetime.fromisoformat(point["t"].replace('Z', '+00:00'))
except ValueError:
pytest.fail(f"Invalid timestamp format: {point['t']}")
# Validate meta
meta = s["meta"]
assert "category" in meta, "Meta should have category"
assert "default_ttl" in meta, "Meta should have default_ttl"
def test_freshness_72h_subset(self):
"""Test freshness history with custom time range and provider selection"""
r = R.get(
f"{BASE}/api/charts/freshness-history",
params={"hours": 72, "providers": "coingecko,binance"}
)
r.raise_for_status()
data = r.json()
assert isinstance(data, list), "Response should be a list"
assert len(data) <= 2, "Should have at most 2 providers"
for series in data:
assert series["hours"] == 72, "Should have 72 hours of data"
assert len(series["series"]) == 72, "Should have 72 data points"
assert series["provider"] in ["coingecko", "binance"], "Provider should match requested"
def test_freshness_hours_clamping(self):
"""Test that hours parameter is properly clamped to valid range"""
# Test lower bound (should clamp to 1)
r = R.get(f"{BASE}/api/charts/freshness-history", params={"hours": 0})
assert r.status_code in [200, 422], "Should handle hours=0"
# Test upper bound (should clamp to 168)
r = R.get(f"{BASE}/api/charts/freshness-history", params={"hours": 999})
assert r.status_code in [200, 422], "Should handle hours=999"
def test_freshness_invalid_provider(self):
"""Test rejection of invalid provider names"""
r = R.get(
f"{BASE}/api/charts/freshness-history",
params={"providers": "foo,bar"}
)
# Should return 400 for invalid providers
assert r.status_code in [400, 404], "Should reject invalid provider names"
def test_freshness_status_derivation(self):
"""Test that status is correctly derived from staleness and TTL"""
r = R.get(f"{BASE}/api/charts/freshness-history")
r.raise_for_status()
data = r.json()
if len(data) > 0:
for series in data:
ttl = series["meta"]["default_ttl"]
for point in series["series"]:
staleness = point["staleness_min"]
status = point["status"]
# Validate status derivation logic
if staleness <= ttl:
expected = "fresh"
elif staleness <= ttl * 2:
expected = "aging"
else:
expected = "stale"
# Allow for edge case where staleness is 999 (no data)
if staleness == 999.0:
assert status == "stale", "No data should be marked as stale"
else:
assert status == expected, f"Status mismatch: staleness={staleness}, ttl={ttl}, expected={expected}, got={status}"
def test_freshness_response_time(self):
"""Test that endpoint responds within performance target (< 200ms for 24h)"""
import time
start = time.time()
r = R.get(f"{BASE}/api/charts/freshness-history")
duration_ms = (time.time() - start) * 1000
r.raise_for_status()
# Allow 500ms for dev environment
assert duration_ms < 500, f"Response took {duration_ms:.0f}ms (target < 500ms)"
class TestSecurityValidation:
"""Test security and validation measures"""
def test_sql_injection_prevention(self):
"""Test that SQL injection attempts are safely handled"""
malicious_providers = "'; DROP TABLE providers; --"
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": malicious_providers}
)
# Should reject or safely handle malicious input
assert r.status_code in [400, 404, 500], "Should reject SQL injection attempts"
def test_xss_prevention(self):
"""Test that XSS attempts are safely handled"""
malicious_providers = "<script>alert('xss')</script>"
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": malicious_providers}
)
# Should reject or safely handle malicious input
assert r.status_code in [400, 404], "Should reject XSS attempts"
def test_parameter_type_validation(self):
"""Test that invalid parameter types are rejected"""
# Test invalid hours type
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"hours": "invalid"}
)
assert r.status_code == 422, "Should reject invalid parameter type"
class TestEdgeCases:
"""Test edge cases and boundary conditions"""
def test_empty_provider_list(self):
"""Test behavior with empty provider list"""
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": ""}
)
r.raise_for_status()
data = r.json()
# Should return default providers or empty list
assert isinstance(data, list), "Should return list even with empty providers param"
def test_whitespace_handling(self):
"""Test that whitespace in provider names is properly handled"""
r = R.get(
f"{BASE}/api/charts/rate-limit-history",
params={"providers": " coingecko , cmc "}
)
# Should handle whitespace gracefully
if r.status_code == 200:
data = r.json()
for series in data:
assert series["provider"].strip() == series["provider"], "Provider names should be trimmed"
def test_concurrent_requests(self):
"""Test that endpoint handles concurrent requests safely"""
import concurrent.futures
def make_request():
r = R.get(f"{BASE}/api/charts/rate-limit-history")
r.raise_for_status()
return r.json()
# Make 5 concurrent requests
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(make_request) for _ in range(5)]
results = [f.result() for f in concurrent.futures.as_completed(futures)]
# All should succeed
assert len(results) == 5, "All concurrent requests should succeed"
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])
|