Remove Python cache files from version control
- Remove __pycache__ directories that should not be tracked - These are now ignored via .gitignore
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
12
test/main.py
12
test/main.py
@@ -1,12 +0,0 @@
|
||||
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)
|
||||
105
test/p2p.py
105
test/p2p.py
@@ -1,105 +0,0 @@
|
||||
#!/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())
|
||||
193
test/ws.py
193
test/ws.py
@@ -1,193 +0,0 @@
|
||||
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())
|
||||
Binary file not shown.
Reference in New Issue
Block a user