Private
Public Access
140 lines
4.9 KiB
Python
140 lines
4.9 KiB
Python
"""tests/test_ir_blaster.py"""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from tuya_re import crypto, packet
|
|
from tuya_re.config import Credentials
|
|
from tuya_re.const import Code, DPType, SECURITY_FLAG_LOGIN, SECURITY_FLAG_SESSION
|
|
from tuya_re.ir import IRBlaster
|
|
from tuya_re.protocol import Session
|
|
|
|
CREDS = Credentials(
|
|
device_id="testdevice000000000001",
|
|
uuid="uuidTESTtest0000",
|
|
local_key="abcdef0123456789",
|
|
product_id="xgahj3gjk3ftaejt",
|
|
)
|
|
SRAND = b"\x01\x02\x03\x04\x05\x06"
|
|
LOGIN_KEY = crypto.login_key(CREDS.local_key)
|
|
SESSION_KEY = crypto.session_key(CREDS.local_key, SRAND)
|
|
|
|
|
|
class FakeDevice:
|
|
"""Decrypts what Session writes; answers as the device. Mirrors the
|
|
sibling repo's protocol-test harness."""
|
|
|
|
def __init__(self):
|
|
self.reassembler = packet.Reassembler()
|
|
self.dp_writes: list[bytes] = [] # payloads of SENDER_DPS
|
|
self.session: Session | None = None
|
|
self._device_seq = 100
|
|
|
|
async def send(self, chunk: bytes) -> None:
|
|
blob = self.reassembler.feed(chunk)
|
|
if blob is None:
|
|
return
|
|
key = LOGIN_KEY if blob[0] == SECURITY_FLAG_LOGIN else SESSION_KEY
|
|
frame = crypto.aes_cbc_decrypt(key, blob[1:17], blob[17:])
|
|
seq, _, code, payload = packet.parse_frame(frame)
|
|
if code == Code.SENDER_DEVICE_INFO:
|
|
info = bytes([1, 0, 3, 0, 0, 1]) + SRAND + bytes(2) + bytes(32)
|
|
self._respond(
|
|
seq, Code.SENDER_DEVICE_INFO, info, LOGIN_KEY, SECURITY_FLAG_LOGIN
|
|
)
|
|
elif code == Code.SENDER_PAIR:
|
|
self._respond(
|
|
seq, Code.SENDER_PAIR, b"\x00", SESSION_KEY, SECURITY_FLAG_SESSION
|
|
)
|
|
elif code == Code.SENDER_DPS:
|
|
self.dp_writes.append(payload)
|
|
self._respond(
|
|
seq, Code.SENDER_DPS, b"\x00", SESSION_KEY, SECURITY_FLAG_SESSION
|
|
)
|
|
|
|
def _respond(self, response_to, code, payload, key, flag) -> None:
|
|
self._device_seq += 1
|
|
frame = packet.build_frame(self._device_seq, response_to, code, payload)
|
|
blob = packet.encrypt_frame(frame, key, flag)
|
|
for chunk in packet.split_packets(blob):
|
|
self.session.handle_notification(chunk)
|
|
|
|
def push_dp(self, dp_id: int, dp_type: DPType, value) -> None:
|
|
"""Device-initiated DP report (response_to=0)."""
|
|
from tuya_re.protocol import encode_dp_value
|
|
|
|
enc = encode_dp_value(dp_type, value)
|
|
payload = bytes([dp_id, dp_type, len(enc)]) + enc
|
|
self._device_seq += 1
|
|
frame = packet.build_frame(self._device_seq, 0, Code.RECEIVE_DP, payload)
|
|
blob = packet.encrypt_frame(frame, SESSION_KEY, SECURITY_FLAG_SESSION)
|
|
for chunk in packet.split_packets(blob):
|
|
self.session.handle_notification(chunk)
|
|
|
|
|
|
@pytest.fixture
|
|
def linked():
|
|
device = FakeDevice()
|
|
session = Session(CREDS, device.send, response_timeout=1.0)
|
|
device.session = session
|
|
return device, session
|
|
|
|
|
|
def _decode_dp_write(payload: bytes):
|
|
dp_id, dp_type, length = payload[0], payload[1], payload[2]
|
|
return dp_id, DPType(dp_type), payload[3 : 3 + length]
|
|
|
|
|
|
async def test_send_ir_writes_base64_to_send_dp(linked):
|
|
device, session = linked
|
|
await session.handshake()
|
|
blaster = IRBlaster(session) # defaults: send DP 201, STRING
|
|
await blaster.send_ir("8iOQERoC")
|
|
assert len(device.dp_writes) == 1
|
|
dp_id, dp_type, raw = _decode_dp_write(device.dp_writes[0])
|
|
assert dp_id == 201
|
|
assert dp_type == DPType.STRING
|
|
assert raw == b"8iOQERoC"
|
|
|
|
|
|
async def test_send_ir_honors_custom_dp_and_type(linked):
|
|
device, session = linked
|
|
await session.handshake()
|
|
blaster = IRBlaster(session, send_dp=7, send_dp_type=DPType.RAW)
|
|
await blaster.send_ir("8iOQERoC")
|
|
dp_id, dp_type, _ = _decode_dp_write(device.dp_writes[0])
|
|
assert dp_id == 7
|
|
assert dp_type == DPType.RAW
|
|
|
|
|
|
async def test_learn_enters_waits_reports_and_exits(linked):
|
|
device, session = linked
|
|
await session.handshake()
|
|
blaster = IRBlaster(session)
|
|
|
|
async def press_remote():
|
|
await asyncio.sleep(0.05)
|
|
device.push_dp(202, DPType.RAW, b"\xf2\x23\x90\x11\x1a\x02")
|
|
|
|
learned, _ = await asyncio.gather(
|
|
blaster.learn(timeout=1.0, poll_interval=0.01),
|
|
press_remote(),
|
|
)
|
|
assert learned == "8iOQERoC" # RAW bytes -> base64
|
|
|
|
first = _decode_dp_write(device.dp_writes[0])
|
|
last = _decode_dp_write(device.dp_writes[-1])
|
|
assert first == (202, DPType.ENUM, b"\x01") # STUDY_ON
|
|
assert last == (202, DPType.ENUM, b"\x00") # STUDY_OFF
|
|
|
|
|
|
async def test_learn_times_out_but_still_exits_study(linked):
|
|
device, session = linked
|
|
await session.handshake()
|
|
blaster = IRBlaster(session)
|
|
with pytest.raises(TimeoutError, match="no IR code learned"):
|
|
await blaster.learn(timeout=0.1, poll_interval=0.01)
|
|
last = _decode_dp_write(device.dp_writes[-1])
|
|
assert last == (202, DPType.ENUM, b"\x00") # STUDY_OFF
|