feat(02-02): implement ESP32Transport UDP client with asyncio DatagramProtocol
- ESP32Transport.send_command() injects v:1, fire-and-forget, silent drop if not ready - 512-byte payload limit enforced with warning log - datagram_received() parses STATUS ok → connected=True, stores last_status - on_status callback support for status update notifications - connection_lost() sets connected=False, clears transport ref - create_esp32_transport() async factory sends initial STATUS ping (D-12) - All protocol behaviors match docs/protocol.md exactly (D-10, D-11, D-12, D-14) - Add pytest-asyncio to dev deps + asyncio_mode=auto for test suite
This commit is contained in:
32
tests/test_player_smoke.py
Normal file
32
tests/test_player_smoke.py
Normal file
@@ -0,0 +1,32 @@
|
||||
"""
|
||||
Manual smoke test for AudioPlayer — requires an actual audio file.
|
||||
Not part of the automated test suite.
|
||||
|
||||
Run: uv run python tests/test_player_smoke.py /path/to/test.mp3
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
from led_sync.audio.player import AudioPlayer
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: uv run python tests/test_player_smoke.py <audio_file>")
|
||||
sys.exit(1)
|
||||
|
||||
p = AudioPlayer()
|
||||
p.play(sys.argv[1])
|
||||
time.sleep(3)
|
||||
print(f"Position: {p.position_seconds:.2f}s")
|
||||
p.pause()
|
||||
print(f"Paused at: {p.position_seconds:.2f}s")
|
||||
time.sleep(1)
|
||||
p.resume()
|
||||
time.sleep(2)
|
||||
print(f"After resume: {p.position_seconds:.2f}s")
|
||||
p.stop()
|
||||
print("Done")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
33
tests/test_scaffold.py
Normal file
33
tests/test_scaffold.py
Normal file
@@ -0,0 +1,33 @@
|
||||
"""TDD RED: Tests for project scaffold — these must fail before implementation."""
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def test_led_sync_importable():
|
||||
"""led_sync package must be importable with a __version__ attribute."""
|
||||
result = subprocess.run(
|
||||
["uv", "run", "python", "-c", "import led_sync; print(led_sync.__version__)"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd="/home/claude/keineahnungirgendeintesthalt",
|
||||
)
|
||||
assert result.returncode == 0, f"Import failed: {result.stderr}"
|
||||
assert result.stdout.strip(), "No version printed"
|
||||
|
||||
|
||||
def test_core_dependencies_importable():
|
||||
"""All Phase 2 core deps must be importable."""
|
||||
result = subprocess.run(
|
||||
[
|
||||
"uv",
|
||||
"run",
|
||||
"python",
|
||||
"-c",
|
||||
"import miniaudio, sounddevice, numpy, pydantic; print('OK')",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd="/home/claude/keineahnungirgendeintesthalt",
|
||||
)
|
||||
assert result.returncode == 0, f"Dependency import failed: {result.stderr}"
|
||||
assert "OK" in result.stdout
|
||||
Reference in New Issue
Block a user