dynamic dashboard initial
This commit is contained in:
169
tools/test_api_client.py
Normal file
169
tools/test_api_client.py
Normal file
@@ -0,0 +1,169 @@
|
||||
"""
|
||||
Acceptance tests for API Client (fetch_devices).
|
||||
|
||||
Tests:
|
||||
1. API running: fetch_devices() returns list with test_lampe_1, test_lampe_2
|
||||
2. API down: fetch_devices() returns empty list without crash
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
import subprocess
|
||||
from apps.ui.api_client import fetch_devices
|
||||
|
||||
|
||||
def test_api_running():
|
||||
"""Test 1: API is running -> fetch_devices() returns devices."""
|
||||
print("Test 1: API running")
|
||||
print("-" * 60)
|
||||
|
||||
devices = fetch_devices("http://localhost:8001")
|
||||
|
||||
if not devices:
|
||||
print(" ✗ FAILED: Expected devices, got empty list")
|
||||
return False
|
||||
|
||||
print(f" ✓ Received {len(devices)} devices")
|
||||
|
||||
# Check structure
|
||||
for device in devices:
|
||||
device_id = device.get("device_id")
|
||||
device_type = device.get("type")
|
||||
name = device.get("name")
|
||||
|
||||
if not device_id:
|
||||
print(f" ✗ FAILED: Device missing 'device_id': {device}")
|
||||
return False
|
||||
|
||||
if not device_type:
|
||||
print(f" ✗ FAILED: Device missing 'type': {device}")
|
||||
return False
|
||||
|
||||
if not name:
|
||||
print(f" ✗ FAILED: Device missing 'name': {device}")
|
||||
return False
|
||||
|
||||
print(f" • {device_id} (type={device_type}, name={name})")
|
||||
|
||||
# Check for expected devices
|
||||
device_ids = {d["device_id"] for d in devices}
|
||||
expected = {"test_lampe_1", "test_lampe_2"}
|
||||
|
||||
if not expected.issubset(device_ids):
|
||||
missing = expected - device_ids
|
||||
print(f" ✗ FAILED: Missing devices: {missing}")
|
||||
return False
|
||||
|
||||
print(" ✓ All expected devices present")
|
||||
print()
|
||||
return True
|
||||
|
||||
|
||||
def test_api_down():
|
||||
"""Test 2: API is down -> fetch_devices() returns empty list without crash."""
|
||||
print("Test 2: API down (invalid port)")
|
||||
print("-" * 60)
|
||||
|
||||
try:
|
||||
devices = fetch_devices("http://localhost:9999")
|
||||
|
||||
if devices != []:
|
||||
print(f" ✗ FAILED: Expected empty list, got {len(devices)} devices")
|
||||
return False
|
||||
|
||||
print(" ✓ Returns empty list without crash")
|
||||
print()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ✗ FAILED: Exception raised: {e}")
|
||||
print()
|
||||
return False
|
||||
|
||||
|
||||
def test_api_timeout():
|
||||
"""Test 3: API timeout -> fetch_devices() returns empty list."""
|
||||
print("Test 3: API timeout")
|
||||
print("-" * 60)
|
||||
|
||||
# Use httpbin.org delay endpoint to simulate slow API
|
||||
try:
|
||||
start = time.time()
|
||||
devices = fetch_devices("https://httpbin.org/delay/5") # 5s delay, but 3s timeout
|
||||
elapsed = time.time() - start
|
||||
|
||||
if devices != []:
|
||||
print(f" ✗ FAILED: Expected empty list, got {len(devices)} devices")
|
||||
return False
|
||||
|
||||
if elapsed >= 4.0:
|
||||
print(f" ✗ FAILED: Timeout not enforced (took {elapsed:.1f}s)")
|
||||
return False
|
||||
|
||||
print(f" ✓ Returns empty list after timeout ({elapsed:.1f}s)")
|
||||
print()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" ✗ FAILED: Exception raised: {e}")
|
||||
print()
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Run all acceptance tests."""
|
||||
print("=" * 60)
|
||||
print("Testing API Client (fetch_devices)")
|
||||
print("=" * 60)
|
||||
print()
|
||||
|
||||
# Check if API is running
|
||||
print("Prerequisites: Checking if API is running on port 8001...")
|
||||
try:
|
||||
devices = fetch_devices("http://localhost:8001")
|
||||
if devices:
|
||||
print(f"✓ API is running ({len(devices)} devices found)")
|
||||
print()
|
||||
else:
|
||||
print("⚠ API might not be running (no devices returned)")
|
||||
print(" Start API with: poetry run uvicorn apps.api.main:app --port 8001")
|
||||
print()
|
||||
except Exception as e:
|
||||
print(f"✗ Cannot reach API: {e}")
|
||||
print(" Start API with: poetry run uvicorn apps.api.main:app --port 8001")
|
||||
print()
|
||||
sys.exit(1)
|
||||
|
||||
results = []
|
||||
|
||||
# Test 1: API running
|
||||
results.append(("API running", test_api_running()))
|
||||
|
||||
# Test 2: API down
|
||||
results.append(("API down", test_api_down()))
|
||||
|
||||
# Test 3: Timeout
|
||||
results.append(("API timeout", test_api_timeout()))
|
||||
|
||||
# Summary
|
||||
print("=" * 60)
|
||||
passed = sum(1 for _, result in results if result)
|
||||
total = len(results)
|
||||
print(f"Results: {passed}/{total} tests passed")
|
||||
print("=" * 60)
|
||||
|
||||
for name, result in results:
|
||||
status = "✓" if result else "✗"
|
||||
print(f" {status} {name}")
|
||||
|
||||
if passed == total:
|
||||
print()
|
||||
print("All tests passed!")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print()
|
||||
print("Some tests failed.")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user