Add endpoint tests and consolidate test directory
- Add HTTP endpoint tests to mimic browser interactions - Move old test files from test/ to tests/ directory - Add comprehensive endpoint tests for tabs, profiles, presets, patterns - Add README documenting test structure and how to run tests
This commit is contained in:
79
tests/README.md
Normal file
79
tests/README.md
Normal file
@@ -0,0 +1,79 @@
|
||||
# Tests
|
||||
|
||||
This directory contains tests for the LED Controller project.
|
||||
|
||||
## Directory Structure
|
||||
|
||||
- `test_endpoints.py` - HTTP endpoint tests that mimic web browser requests (runs against 192.168.4.1)
|
||||
- `test_ws.py` - WebSocket tests
|
||||
- `test_p2p.py` - ESP-NOW P2P tests
|
||||
- `models/` - Model unit tests
|
||||
- `web.py` - Local development web server
|
||||
|
||||
## Running Tests
|
||||
|
||||
### Browser Tests (Real Browser Automation)
|
||||
|
||||
Tests the web interface in an actual browser using Selenium:
|
||||
|
||||
```bash
|
||||
python tests/test_browser.py
|
||||
```
|
||||
|
||||
These tests:
|
||||
- Open a real Chrome browser
|
||||
- Navigate to the device at 192.168.4.1
|
||||
- Interact with UI elements (buttons, forms, modals)
|
||||
- Test complete user workflows
|
||||
- Verify visual elements and interactions
|
||||
|
||||
**Requirements:**
|
||||
```bash
|
||||
pip install selenium
|
||||
# Also need ChromeDriver installed and in PATH
|
||||
# Download from: https://chromedriver.chromium.org/
|
||||
```
|
||||
|
||||
### Endpoint Tests (Browser-like HTTP)
|
||||
|
||||
Tests HTTP endpoints by making requests to the device at 192.168.4.1:
|
||||
|
||||
```bash
|
||||
python tests/test_endpoints.py
|
||||
```
|
||||
|
||||
These tests:
|
||||
- Mimic web browser requests with proper headers
|
||||
- Handle cookies for session management
|
||||
- Test all CRUD operations (GET, POST, PUT, DELETE)
|
||||
- Verify responses and status codes
|
||||
|
||||
**Requirements:**
|
||||
```bash
|
||||
pip install requests
|
||||
```
|
||||
|
||||
### WebSocket Tests
|
||||
|
||||
```bash
|
||||
python tests/test_ws.py
|
||||
```
|
||||
|
||||
**Requirements:**
|
||||
```bash
|
||||
pip install websockets
|
||||
```
|
||||
|
||||
### Model Tests
|
||||
|
||||
```bash
|
||||
python tests/models/run_all.py
|
||||
```
|
||||
|
||||
### Local Development Server
|
||||
|
||||
Run the local development server (port 5000):
|
||||
|
||||
```bash
|
||||
python tests/web.py
|
||||
```
|
||||
105
tests/p2p.py
Normal file
105
tests/p2p.py
Normal file
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
# MicroPython script to test LED bar patterns over ESP-NOW (no WebSocket)
|
||||
|
||||
import json
|
||||
import uasyncio as asyncio
|
||||
|
||||
# Import P2P from src/p2p.py
|
||||
# Note: When running on device, ensure src/p2p.py is in the path
|
||||
try:
|
||||
from p2p import P2P
|
||||
except ImportError:
|
||||
# Fallback: import from src directory
|
||||
import sys
|
||||
sys.path.insert(0, 'src')
|
||||
from p2p import P2P
|
||||
|
||||
async def main():
|
||||
p2p = P2P()
|
||||
|
||||
# Test cases following msg.json format:
|
||||
# {"g": {"df": {...}, "group_name": {...}}, "sv": true, "st": 0}
|
||||
# Note: led-bar device must have matching group in settings["groups"]
|
||||
tests = [
|
||||
# Example 1: Default format with df defaults and dj group (matches msg.json)
|
||||
{
|
||||
"g": {
|
||||
"df": {
|
||||
"pt": "on",
|
||||
"cl": ["#ff0000"],
|
||||
"br": 200,
|
||||
"n1": 10,
|
||||
"n2": 10,
|
||||
"n3": 10,
|
||||
"n4": 10,
|
||||
"n5": 10,
|
||||
"n6": 10,
|
||||
"dl": 100
|
||||
},
|
||||
"dj": {
|
||||
"pt": "blink",
|
||||
"cl": ["#00ff00"],
|
||||
"dl": 500
|
||||
}
|
||||
},
|
||||
"sv": True,
|
||||
"st": 0
|
||||
},
|
||||
# Example 2: Different group with df defaults
|
||||
{
|
||||
"g": {
|
||||
"df": {
|
||||
"pt": "on",
|
||||
"br": 150,
|
||||
"dl": 100
|
||||
},
|
||||
"group1": {
|
||||
"pt": "rainbow",
|
||||
"dl": 50
|
||||
}
|
||||
},
|
||||
"sv": False
|
||||
},
|
||||
# Example 3: Multiple groups
|
||||
{
|
||||
"g": {
|
||||
"df": {
|
||||
"br": 200,
|
||||
"dl": 100
|
||||
},
|
||||
"group1": {
|
||||
"pt": "on",
|
||||
"cl": ["#0000ff"]
|
||||
},
|
||||
"group2": {
|
||||
"pt": "blink",
|
||||
"cl": ["#ff00ff"],
|
||||
"dl": 300
|
||||
}
|
||||
},
|
||||
"sv": True,
|
||||
"st": 1
|
||||
},
|
||||
# Example 4: Single group without df
|
||||
{
|
||||
"g": {
|
||||
"dj": {
|
||||
"pt": "off"
|
||||
}
|
||||
},
|
||||
"sv": False
|
||||
}
|
||||
]
|
||||
|
||||
for i, test in enumerate(tests, 1):
|
||||
print(f"\n{'='*50}")
|
||||
print(f"Test {i}/{len(tests)}")
|
||||
print(f"Sending: {json.dumps(test, indent=2)}")
|
||||
await p2p.send(json.dumps(test))
|
||||
await asyncio.sleep_ms(2000)
|
||||
|
||||
print(f"\n{'='*50}")
|
||||
print("All tests completed")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
548
tests/test_endpoints.py
Normal file
548
tests/test_endpoints.py
Normal file
@@ -0,0 +1,548 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Endpoint tests that mimic web browser requests.
|
||||
Tests run against the device at 192.168.4.1
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
from typing import Dict, Optional
|
||||
|
||||
# Base URL for the device
|
||||
BASE_URL = "http://192.168.4.1"
|
||||
|
||||
class TestClient:
|
||||
"""HTTP client that mimics a web browser with cookie support."""
|
||||
|
||||
def __init__(self, base_url: str = BASE_URL):
|
||||
self.base_url = base_url
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({
|
||||
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
|
||||
'Accept': 'application/json',
|
||||
'Accept-Language': 'en-US,en;q=0.9',
|
||||
})
|
||||
|
||||
def get(self, path: str, **kwargs) -> requests.Response:
|
||||
"""GET request."""
|
||||
url = f"{self.base_url}{path}"
|
||||
return self.session.get(url, **kwargs)
|
||||
|
||||
def post(self, path: str, data: Optional[Dict] = None, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
|
||||
"""POST request."""
|
||||
url = f"{self.base_url}{path}"
|
||||
if json_data:
|
||||
return self.session.post(url, json=json_data, **kwargs)
|
||||
return self.session.post(url, data=data, **kwargs)
|
||||
|
||||
def put(self, path: str, json_data: Optional[Dict] = None, **kwargs) -> requests.Response:
|
||||
"""PUT request."""
|
||||
url = f"{self.base_url}{path}"
|
||||
return self.session.put(url, json=json_data, **kwargs)
|
||||
|
||||
def delete(self, path: str, **kwargs) -> requests.Response:
|
||||
"""DELETE request."""
|
||||
url = f"{self.base_url}{path}"
|
||||
return self.session.delete(url, **kwargs)
|
||||
|
||||
def set_cookie(self, name: str, value: str):
|
||||
"""Set a cookie manually."""
|
||||
self.session.cookies.set(name, value, domain='192.168.4.1', path='/')
|
||||
|
||||
def get_cookie(self, name: str) -> Optional[str]:
|
||||
"""Get a cookie value."""
|
||||
return self.session.cookies.get(name)
|
||||
|
||||
def test_connection(client: TestClient) -> bool:
|
||||
"""Test basic connection to the server."""
|
||||
print("Testing connection...")
|
||||
try:
|
||||
response = client.get('/')
|
||||
if response.status_code == 200:
|
||||
print("✓ Connection successful")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ Connection failed: {response.status_code}")
|
||||
return False
|
||||
except requests.exceptions.ConnectionError:
|
||||
print(f"✗ Cannot connect to {BASE_URL}")
|
||||
print(" Make sure the device is running and accessible at 192.168.4.1")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ Connection error: {e}")
|
||||
return False
|
||||
|
||||
def test_tabs(client: TestClient) -> bool:
|
||||
"""Test tabs endpoints."""
|
||||
print("\n=== Testing Tabs Endpoints ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
# Test 1: List tabs
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/tabs')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"✓ GET /tabs - Found {len(data.get('tabs', {}))} tabs")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /tabs - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /tabs - Error: {e}")
|
||||
|
||||
# Test 2: Create tab
|
||||
total += 1
|
||||
try:
|
||||
tab_data = {
|
||||
"name": "Test Tab",
|
||||
"names": ["1", "2"]
|
||||
}
|
||||
response = client.post('/tabs', json_data=tab_data)
|
||||
if response.status_code == 201:
|
||||
created_tab = response.json()
|
||||
# Response format: {tab_id: {tab_data}}
|
||||
if isinstance(created_tab, dict):
|
||||
# Get the first key which should be the tab ID
|
||||
tab_id = next(iter(created_tab.keys())) if created_tab else None
|
||||
else:
|
||||
tab_id = None
|
||||
print(f"✓ POST /tabs - Created tab: {tab_id}")
|
||||
passed += 1
|
||||
|
||||
# Test 3: Get specific tab
|
||||
if tab_id:
|
||||
total += 1
|
||||
response = client.get(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ GET /tabs/{tab_id} - Retrieved tab")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /tabs/{tab_id} - Status: {response.status_code}")
|
||||
|
||||
# Test 4: Set current tab
|
||||
total += 1
|
||||
response = client.post(f'/tabs/{tab_id}/set-current')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ POST /tabs/{tab_id}/set-current - Set current tab")
|
||||
# Check cookie was set
|
||||
cookie = client.get_cookie('current_tab')
|
||||
if cookie == tab_id:
|
||||
print(f" ✓ Cookie 'current_tab' set to {tab_id}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ POST /tabs/{tab_id}/set-current - Status: {response.status_code}")
|
||||
|
||||
# Test 5: Get current tab
|
||||
total += 1
|
||||
response = client.get('/tabs/current')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
if data.get('tab_id') == tab_id:
|
||||
print(f"✓ GET /tabs/current - Current tab is {tab_id}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /tabs/current - Wrong tab ID")
|
||||
else:
|
||||
print(f"✗ GET /tabs/current - Status: {response.status_code}")
|
||||
|
||||
# Test 6: Update tab (edit functionality)
|
||||
total += 1
|
||||
update_data = {
|
||||
"name": "Updated Test Tab",
|
||||
"names": ["1", "2", "3"] # Update device IDs too
|
||||
}
|
||||
response = client.put(f'/tabs/{tab_id}', json_data=update_data)
|
||||
if response.status_code == 200:
|
||||
updated = response.json()
|
||||
if updated.get('name') == "Updated Test Tab" and updated.get('names') == ["1", "2", "3"]:
|
||||
print(f"✓ PUT /tabs/{tab_id} - Updated tab (name and device IDs)")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ PUT /tabs/{tab_id} - Update didn't work correctly")
|
||||
print(f" Expected name='Updated Test Tab', got '{updated.get('name')}'")
|
||||
print(f" Expected names=['1','2','3'], got {updated.get('names')}")
|
||||
else:
|
||||
print(f"✗ PUT /tabs/{tab_id} - Status: {response.status_code}, Response: {response.text}")
|
||||
|
||||
# Test 6b: Verify update persisted
|
||||
total += 1
|
||||
response = client.get(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
verified = response.json()
|
||||
if verified.get('name') == "Updated Test Tab":
|
||||
print(f"✓ GET /tabs/{tab_id} - Verified update persisted")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /tabs/{tab_id} - Update didn't persist")
|
||||
else:
|
||||
print(f"✗ GET /tabs/{tab_id} - Status: {response.status_code}")
|
||||
|
||||
# Test 7: Delete tab
|
||||
total += 1
|
||||
response = client.delete(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ DELETE /tabs/{tab_id} - Deleted tab")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ DELETE /tabs/{tab_id} - Status: {response.status_code}")
|
||||
else:
|
||||
print(f"✗ POST /tabs - Status: {response.status_code}, Response: {response.text}")
|
||||
except Exception as e:
|
||||
print(f"✗ POST /tabs - Error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print(f"\nTabs tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def test_profiles(client: TestClient) -> bool:
|
||||
"""Test profiles endpoints."""
|
||||
print("\n=== Testing Profiles Endpoints ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
# Test 1: List profiles
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/profiles')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
profiles = data.get('profiles', {})
|
||||
current_id = data.get('current_profile_id')
|
||||
print(f"✓ GET /profiles - Found {len(profiles)} profiles, current: {current_id}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /profiles - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /profiles - Error: {e}")
|
||||
|
||||
# Test 2: Get current profile
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/profiles/current')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"✓ GET /profiles/current - Current profile: {data.get('id')}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /profiles/current - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /profiles/current - Error: {e}")
|
||||
|
||||
# Test 3: Create profile
|
||||
total += 1
|
||||
try:
|
||||
profile_data = {"name": "Test Profile"}
|
||||
response = client.post('/profiles', json_data=profile_data)
|
||||
if response.status_code == 201:
|
||||
created = response.json()
|
||||
# Response format: {profile_id: {profile_data}}
|
||||
if isinstance(created, dict):
|
||||
profile_id = next(iter(created.keys())) if created else None
|
||||
else:
|
||||
profile_id = None
|
||||
print(f"✓ POST /profiles - Created profile: {profile_id}")
|
||||
passed += 1
|
||||
|
||||
# Test 4: Apply profile
|
||||
if profile_id:
|
||||
total += 1
|
||||
response = client.post(f'/profiles/{profile_id}/apply')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ POST /profiles/{profile_id}/apply - Applied profile")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ POST /profiles/{profile_id}/apply - Status: {response.status_code}")
|
||||
|
||||
# Test 5: Delete profile
|
||||
total += 1
|
||||
response = client.delete(f'/profiles/{profile_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ DELETE /profiles/{profile_id} - Deleted profile")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ DELETE /profiles/{profile_id} - Status: {response.status_code}")
|
||||
else:
|
||||
print(f"✗ POST /profiles - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ POST /profiles - Error: {e}")
|
||||
|
||||
print(f"\nProfiles tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def test_presets(client: TestClient) -> bool:
|
||||
"""Test presets endpoints."""
|
||||
print("\n=== Testing Presets Endpoints ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
# Test 1: List presets
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/presets')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
preset_count = len(data) if isinstance(data, dict) else 0
|
||||
print(f"✓ GET /presets - Found {preset_count} presets")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /presets - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /presets - Error: {e}")
|
||||
|
||||
# Test 2: Create preset
|
||||
total += 1
|
||||
try:
|
||||
preset_data = {
|
||||
"name": "Test Preset",
|
||||
"pattern": "on",
|
||||
"colors": ["#ff0000"],
|
||||
"brightness": 200
|
||||
}
|
||||
response = client.post('/presets', json_data=preset_data)
|
||||
if response.status_code == 201:
|
||||
created = response.json()
|
||||
# Response format: {preset_id: {preset_data}}
|
||||
if isinstance(created, dict):
|
||||
preset_id = next(iter(created.keys())) if created else None
|
||||
else:
|
||||
preset_id = None
|
||||
print(f"✓ POST /presets - Created preset: {preset_id}")
|
||||
passed += 1
|
||||
|
||||
# Test 3: Get specific preset
|
||||
if preset_id:
|
||||
total += 1
|
||||
response = client.get(f'/presets/{preset_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ GET /presets/{preset_id} - Retrieved preset")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /presets/{preset_id} - Status: {response.status_code}")
|
||||
|
||||
# Test 4: Update preset
|
||||
total += 1
|
||||
update_data = {"brightness": 150}
|
||||
response = client.put(f'/presets/{preset_id}', json_data=update_data)
|
||||
if response.status_code == 200:
|
||||
print(f"✓ PUT /presets/{preset_id} - Updated preset")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ PUT /presets/{preset_id} - Status: {response.status_code}")
|
||||
|
||||
# Test 5: Delete preset
|
||||
total += 1
|
||||
response = client.delete(f'/presets/{preset_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ DELETE /presets/{preset_id} - Deleted preset")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ DELETE /presets/{preset_id} - Status: {response.status_code}")
|
||||
else:
|
||||
print(f"✗ POST /presets - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ POST /presets - Error: {e}")
|
||||
|
||||
print(f"\nPresets tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def test_patterns(client: TestClient) -> bool:
|
||||
"""Test patterns endpoints."""
|
||||
print("\n=== Testing Patterns Endpoints ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
# Test 1: List patterns
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/patterns')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
pattern_count = len(data) if isinstance(data, dict) else 0
|
||||
print(f"✓ GET /patterns - Found {pattern_count} patterns")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /patterns - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /patterns - Error: {e}")
|
||||
|
||||
# Test 2: Get pattern definitions
|
||||
total += 1
|
||||
try:
|
||||
response = client.get('/patterns/definitions')
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
print(f"✓ GET /patterns/definitions - Retrieved definitions")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /patterns/definitions - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET /patterns/definitions - Error: {e}")
|
||||
|
||||
print(f"\nPatterns tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def test_tab_edit_workflow(client: TestClient) -> bool:
|
||||
"""Test complete tab edit workflow like a browser would."""
|
||||
print("\n=== Testing Tab Edit Workflow ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
# Step 1: Create a tab to edit
|
||||
total += 1
|
||||
try:
|
||||
tab_data = {
|
||||
"name": "Tab to Edit",
|
||||
"names": ["1"]
|
||||
}
|
||||
response = client.post('/tabs', json_data=tab_data)
|
||||
if response.status_code == 201:
|
||||
created = response.json()
|
||||
if isinstance(created, dict):
|
||||
tab_id = next(iter(created.keys())) if created else None
|
||||
else:
|
||||
tab_id = None
|
||||
|
||||
if tab_id:
|
||||
print(f"✓ Created tab {tab_id} for editing")
|
||||
passed += 1
|
||||
|
||||
# Step 2: Get the tab to verify initial state
|
||||
total += 1
|
||||
response = client.get(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
original_tab = response.json()
|
||||
print(f"✓ Retrieved tab - Name: '{original_tab.get('name')}', IDs: {original_tab.get('names')}")
|
||||
passed += 1
|
||||
|
||||
# Step 3: Edit the tab (simulate browser edit form submission)
|
||||
total += 1
|
||||
edit_data = {
|
||||
"name": "Edited Tab Name",
|
||||
"names": ["2", "3", "4"]
|
||||
}
|
||||
response = client.put(f'/tabs/{tab_id}', json_data=edit_data)
|
||||
if response.status_code == 200:
|
||||
edited = response.json()
|
||||
if edited.get('name') == "Edited Tab Name" and edited.get('names') == ["2", "3", "4"]:
|
||||
print(f"✓ PUT /tabs/{tab_id} - Successfully edited tab")
|
||||
print(f" New name: '{edited.get('name')}'")
|
||||
print(f" New device IDs: {edited.get('names')}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ PUT /tabs/{tab_id} - Edit didn't work correctly")
|
||||
print(f" Got: {edited}")
|
||||
else:
|
||||
print(f"✗ PUT /tabs/{tab_id} - Status: {response.status_code}, Response: {response.text}")
|
||||
|
||||
# Step 4: Verify edit persisted by getting the tab again
|
||||
total += 1
|
||||
response = client.get(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
verified = response.json()
|
||||
if verified.get('name') == "Edited Tab Name" and verified.get('names') == ["2", "3", "4"]:
|
||||
print(f"✓ GET /tabs/{tab_id} - Verified edit persisted")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET /tabs/{tab_id} - Edit didn't persist")
|
||||
print(f" Expected name='Edited Tab Name', got '{verified.get('name')}'")
|
||||
print(f" Expected names=['2','3','4'], got {verified.get('names')}")
|
||||
else:
|
||||
print(f"✗ GET /tabs/{tab_id} - Status: {response.status_code}")
|
||||
|
||||
# Step 5: Clean up - delete the test tab
|
||||
total += 1
|
||||
response = client.delete(f'/tabs/{tab_id}')
|
||||
if response.status_code == 200:
|
||||
print(f"✓ DELETE /tabs/{tab_id} - Cleaned up test tab")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ DELETE /tabs/{tab_id} - Status: {response.status_code}")
|
||||
else:
|
||||
print(f"✗ Failed to extract tab ID from create response")
|
||||
else:
|
||||
print(f"✗ POST /tabs - Status: {response.status_code}, Response: {response.text}")
|
||||
except Exception as e:
|
||||
print(f"✗ Tab edit workflow - Error: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
print(f"\nTab edit workflow tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def test_static_files(client: TestClient) -> bool:
|
||||
"""Test static file serving."""
|
||||
print("\n=== Testing Static Files ===")
|
||||
passed = 0
|
||||
total = 0
|
||||
|
||||
static_files = [
|
||||
'/static/style.css',
|
||||
'/static/app.js',
|
||||
'/static/tabs.js',
|
||||
'/static/presets.js',
|
||||
'/static/profiles.js',
|
||||
]
|
||||
|
||||
for file_path in static_files:
|
||||
total += 1
|
||||
try:
|
||||
response = client.get(file_path)
|
||||
if response.status_code == 200:
|
||||
print(f"✓ GET {file_path} - Retrieved")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ GET {file_path} - Status: {response.status_code}")
|
||||
except Exception as e:
|
||||
print(f"✗ GET {file_path} - Error: {e}")
|
||||
|
||||
print(f"\nStatic files tests: {passed}/{total} passed")
|
||||
return passed == total
|
||||
|
||||
def main():
|
||||
"""Run all endpoint tests."""
|
||||
print("=" * 60)
|
||||
print("LED Controller Endpoint Tests")
|
||||
print(f"Testing against: {BASE_URL}")
|
||||
print("=" * 60)
|
||||
|
||||
client = TestClient()
|
||||
|
||||
# Test connection first
|
||||
if not test_connection(client):
|
||||
print("\n✗ Cannot connect to device. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
results = []
|
||||
|
||||
# Run all tests
|
||||
results.append(("Tabs", test_tabs(client)))
|
||||
results.append(("Tab Edit Workflow", test_tab_edit_workflow(client)))
|
||||
results.append(("Profiles", test_profiles(client)))
|
||||
results.append(("Presets", test_presets(client)))
|
||||
results.append(("Patterns", test_patterns(client)))
|
||||
results.append(("Static Files", test_static_files(client)))
|
||||
|
||||
# Summary
|
||||
print("\n" + "=" * 60)
|
||||
print("Test Summary")
|
||||
print("=" * 60)
|
||||
all_passed = True
|
||||
for name, passed in results:
|
||||
status = "✓ PASS" if passed else "✗ FAIL"
|
||||
print(f"{status} - {name}")
|
||||
if not passed:
|
||||
all_passed = False
|
||||
|
||||
print("=" * 60)
|
||||
if all_passed:
|
||||
print("✓ All tests passed!")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print("✗ Some tests failed")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
12
tests/test_main_old.py
Normal file
12
tests/test_main_old.py
Normal file
@@ -0,0 +1,12 @@
|
||||
from microdot import Microdot
|
||||
from src.profile import profile_app
|
||||
|
||||
app = Microdot()
|
||||
|
||||
@app.route('/')
|
||||
async def index(request):
|
||||
return 'Hello, world!'
|
||||
|
||||
app.mount(profile_app, url_prefix="/profile")
|
||||
|
||||
app.run(port=8080, debug=True)
|
||||
193
tests/ws.py
Normal file
193
tests/ws.py
Normal file
@@ -0,0 +1,193 @@
|
||||
import asyncio
|
||||
import websockets
|
||||
import json
|
||||
import sys
|
||||
|
||||
async def test_websocket():
|
||||
uri = "ws://192.168.4.1:8080/ws"
|
||||
tests_passed = 0
|
||||
tests_total = 0
|
||||
|
||||
async def run_test(name, test_func):
|
||||
nonlocal tests_passed, tests_total
|
||||
tests_total += 1
|
||||
try:
|
||||
result = await test_func()
|
||||
if result is not False:
|
||||
print(f"✓ {name}")
|
||||
tests_passed += 1
|
||||
return True
|
||||
else:
|
||||
print(f"✗ {name} (failed)")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ {name} (error: {e})")
|
||||
return False
|
||||
|
||||
try:
|
||||
print(f"Connecting to WebSocket server at {uri}...")
|
||||
async with websockets.connect(uri) as websocket:
|
||||
print(f"✓ Connected to WebSocket server\n")
|
||||
|
||||
# Test 1: Empty JSON
|
||||
print("Test 1: Empty JSON")
|
||||
await run_test("Send empty JSON", lambda: websocket.send(json.dumps({})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 2: Pattern on with single color
|
||||
print("\nTest 2: Pattern 'on'")
|
||||
await run_test("Send on pattern", lambda: websocket.send(json.dumps({
|
||||
"settings": {"pattern": "on", "colors": ["#00ff00"], "brightness": 200}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 3: Pattern blink
|
||||
print("\nTest 3: Pattern 'blink'")
|
||||
await run_test("Send blink pattern", lambda: websocket.send(json.dumps({
|
||||
"settings": {"pattern": "blink", "colors": ["#ff0000"], "delay": 500}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 4: Pattern rainbow
|
||||
print("\nTest 4: Pattern 'rainbow'")
|
||||
await run_test("Send rainbow pattern", lambda: websocket.send(json.dumps({
|
||||
"settings": {"pattern": "rainbow", "delay": 100}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 5: Pattern off
|
||||
print("\nTest 5: Pattern 'off'")
|
||||
await run_test("Send off pattern", lambda: websocket.send(json.dumps({
|
||||
"settings": {"pattern": "off"}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 6: Multiple colors
|
||||
print("\nTest 6: Multiple colors")
|
||||
await run_test("Send multiple colors", lambda: websocket.send(json.dumps({
|
||||
"settings": {
|
||||
"pattern": "color_transition",
|
||||
"colors": ["#ff0000", "#00ff00", "#0000ff"],
|
||||
"delay": 100
|
||||
}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 7: RGB tuple colors (if supported)
|
||||
print("\nTest 7: RGB tuple colors")
|
||||
await run_test("Send RGB tuple colors", lambda: websocket.send(json.dumps({
|
||||
"settings": {
|
||||
"pattern": "on",
|
||||
"colors": [[255, 0, 128], [128, 255, 0]],
|
||||
"brightness": 150
|
||||
}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 8: Pattern with all parameters
|
||||
print("\nTest 8: Pattern with all parameters")
|
||||
await run_test("Send pattern with all params", lambda: websocket.send(json.dumps({
|
||||
"settings": {
|
||||
"pattern": "flicker",
|
||||
"colors": ["#ff8800"],
|
||||
"brightness": 127,
|
||||
"delay": 80,
|
||||
"n1": 10,
|
||||
"n2": 5,
|
||||
"n3": 1,
|
||||
"n4": 1
|
||||
}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 9: Short-key format (df/dj)
|
||||
print("\nTest 9: Short-key format (df/dj)")
|
||||
await run_test("Send df/dj format", lambda: websocket.send(json.dumps({
|
||||
"df": {"pt": "on", "cl": ["#ff0000"], "br": 200},
|
||||
"dj": {"pa": "blink", "cl": ["#00ff00"], "dl": 500},
|
||||
"settings": {"pattern": "blink", "colors": ["#00ff00"], "delay": 500, "brightness": 200}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 10: Rapid message sending
|
||||
print("\nTest 10: Rapid message sending")
|
||||
patterns = ["on", "off", "on", "blink"]
|
||||
for i, pattern in enumerate(patterns):
|
||||
p = pattern # Capture in closure
|
||||
await run_test(f"Rapid send {i+1}/{len(patterns)}", lambda p=p: websocket.send(json.dumps({
|
||||
"settings": {"pattern": p, "colors": ["#ffffff"]}
|
||||
})))
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Test 11: Large message
|
||||
print("\nTest 11: Large message")
|
||||
large_colors = [f"#{i%256:02x}{i*2%256:02x}{i*3%256:02x}" for i in range(50)]
|
||||
await run_test("Send large message", lambda: websocket.send(json.dumps({
|
||||
"settings": {
|
||||
"pattern": "color_transition",
|
||||
"colors": large_colors,
|
||||
"delay": 50
|
||||
}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 12: Invalid JSON (should be handled gracefully)
|
||||
print("\nTest 12: Invalid JSON handling")
|
||||
try:
|
||||
await websocket.send("not valid json")
|
||||
print("⚠ Invalid JSON sent (server should handle gracefully)")
|
||||
tests_total += 1
|
||||
except Exception as e:
|
||||
print(f"✗ Invalid JSON failed to send: {e}")
|
||||
tests_total += 1
|
||||
|
||||
# Test 13: Malformed structure (missing settings)
|
||||
print("\nTest 13: Malformed structure")
|
||||
await run_test("Send message without settings", lambda: websocket.send(json.dumps({
|
||||
"pattern": "on",
|
||||
"colors": ["#ff0000"]
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 14: Just settings key, no pattern
|
||||
print("\nTest 14: Settings without pattern")
|
||||
await run_test("Send settings without pattern", lambda: websocket.send(json.dumps({
|
||||
"settings": {"colors": ["#0000ff"], "brightness": 100}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
# Test 15: Empty settings
|
||||
print("\nTest 15: Empty settings")
|
||||
await run_test("Send empty settings", lambda: websocket.send(json.dumps({
|
||||
"settings": {}
|
||||
})))
|
||||
await asyncio.sleep(0.3)
|
||||
|
||||
print(f"\n{'='*50}")
|
||||
print(f"Tests completed: {tests_passed}/{tests_total} passed")
|
||||
if tests_passed == tests_total:
|
||||
print("✓ All tests passed!")
|
||||
else:
|
||||
print(f"⚠ {tests_total - tests_passed} test(s) failed")
|
||||
print(f"{'='*50}")
|
||||
|
||||
except websockets.exceptions.ConnectionClosedOK:
|
||||
print("✓ WebSocket connection closed gracefully.")
|
||||
except websockets.exceptions.ConnectionClosedError as e:
|
||||
print(f"✗ WebSocket connection closed with error: {e}")
|
||||
sys.exit(1)
|
||||
except ConnectionRefusedError:
|
||||
print(f"✗ Connection refused. Is the server running at {uri}?")
|
||||
print("Make sure:")
|
||||
print(" 1. The device is connected to WiFi")
|
||||
print(" 2. The server is running on the device")
|
||||
print(" 3. You can reach 192.168.4.1")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"✗ An unexpected error occurred: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_websocket())
|
||||
Reference in New Issue
Block a user