From fd52e40d17e9bf2f060d7bb0680ea0e85e744027 Mon Sep 17 00:00:00 2001 From: jimmy Date: Tue, 27 Jan 2026 13:04:56 +1300 Subject: [PATCH] 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 --- tests/README.md | 79 ++++++ tests/p2p.py | 105 ++++++++ tests/test_endpoints.py | 548 ++++++++++++++++++++++++++++++++++++++++ tests/test_main_old.py | 12 + tests/ws.py | 193 ++++++++++++++ 5 files changed, 937 insertions(+) create mode 100644 tests/README.md create mode 100644 tests/p2p.py create mode 100644 tests/test_endpoints.py create mode 100644 tests/test_main_old.py create mode 100644 tests/ws.py diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..e6d677a --- /dev/null +++ b/tests/README.md @@ -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 +``` diff --git a/tests/p2p.py b/tests/p2p.py new file mode 100644 index 0000000..4b9b16e --- /dev/null +++ b/tests/p2p.py @@ -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()) \ No newline at end of file diff --git a/tests/test_endpoints.py b/tests/test_endpoints.py new file mode 100644 index 0000000..95fc5b5 --- /dev/null +++ b/tests/test_endpoints.py @@ -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() diff --git a/tests/test_main_old.py b/tests/test_main_old.py new file mode 100644 index 0000000..6ad2202 --- /dev/null +++ b/tests/test_main_old.py @@ -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) \ No newline at end of file diff --git a/tests/ws.py b/tests/ws.py new file mode 100644 index 0000000..8b2a6a3 --- /dev/null +++ b/tests/ws.py @@ -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())