1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
"""Simple integration tests for basic MadVR functionality.
These tests require a real MadVR device.
Set MADVR_HOST and MADVR_PORT environment variables to run these tests.
For power on/off tests, see test_power_integration.py
"""
import asyncio
import os
import socket
import sys
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
import pytest
from pymadvr.madvr import Madvr
def is_device_available() -> bool:
"""Check if MadVR device is reachable."""
host = os.getenv("MADVR_HOST", "192.168.1.100")
port = int(os.getenv("MADVR_PORT", "44077"))
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex((host, port))
sock.close()
return result == 0
except Exception:
return False
# Skip all tests in this module if device is not available
pytestmark = pytest.mark.skipif(
not is_device_available(),
reason="MadVR device not available at MADVR_HOST:MADVR_PORT"
)
@pytest.mark.asyncio
async def test_basic_connection():
"""Test basic connection and data retrieval."""
host = os.getenv("MADVR_HOST", "192.168.1.100")
port = int(os.getenv("MADVR_PORT", "44077"))
madvr = Madvr(host, port=port)
try:
await madvr.open_connection()
# Wait a bit for data
await asyncio.sleep(2)
# Check basic properties
assert madvr.connected is True
assert isinstance(madvr.is_on, bool)
# Check msg_dict
assert "is_on" in madvr.msg_dict
assert madvr.msg_dict["is_on"] is True
# If MAC is available, check it
if madvr.mac_address:
assert len(madvr.mac_address) > 0
assert ":" in madvr.mac_address or "-" in madvr.mac_address
# Send a command
response = await madvr.send_command(["GetMacAddress"])
assert response is not None
# Test menu commands
await madvr.add_command_to_queue(["OpenMenu", "Info"])
await asyncio.sleep(1)
await madvr.add_command_to_queue(["CloseMenu"])
await asyncio.sleep(1)
print(f"✓ Test passed! Device info: {madvr.msg_dict}")
finally:
await madvr.close_connection()
@pytest.mark.asyncio
async def test_display_message():
"""Test display message functionality."""
host = os.getenv("MADVR_HOST", "192.168.1.100")
port = int(os.getenv("MADVR_PORT", "44077"))
madvr = Madvr(host, port=port)
try:
await madvr.open_connection()
await asyncio.sleep(1)
# Send display message
await madvr.display_message(3, "Hello from Python!")
await asyncio.sleep(4)
print("✓ Display message test passed!")
finally:
await madvr.close_connection()
@pytest.mark.asyncio
async def test_ha_command_formats():
"""Test Home Assistant command formats that were failing."""
host = os.getenv("MADVR_HOST", "192.168.1.100")
port = int(os.getenv("MADVR_PORT", "44077"))
madvr = Madvr(host, port=port)
try:
await madvr.open_connection()
await asyncio.sleep(1)
# Test the exact commands that were failing in HA logs
print("Testing HA command formats...")
# Test KeyPress commands (comma-separated format from HA)
await madvr.add_command_to_queue(["KeyPress, MENU"])
await asyncio.sleep(2)
await madvr.add_command_to_queue(["KeyPress, MENU"]) # Close menu
await asyncio.sleep(1)
# Test OpenMenu command
await madvr.add_command_to_queue(["OpenMenu, Info"])
await asyncio.sleep(2)
await madvr.add_command_to_queue(["CloseMenu"])
await asyncio.sleep(1)
print("✓ HA command format test passed!")
finally:
await madvr.close_connection()
if __name__ == "__main__":
# Run directly without pytest
async def main():
host = os.getenv("MADVR_HOST", "192.168.1.100")
port = int(os.getenv("MADVR_PORT", "44077"))
print(f"Testing MadVR at {host}:{port}")
if not is_device_available():
print(f"Device not available at {host}:{port}")
return
try:
await test_basic_connection()
await test_display_message()
await test_ha_command_formats()
print("\n✓ All basic tests passed!")
except Exception as e:
print(f"✗ Test failed: {e}")
import traceback
traceback.print_exc()
asyncio.run(main())
|
