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
|
"""Tests for device module."""
from hashlib import sha256
from unittest.mock import AsyncMock, call
import pytest
from jvcprojector import command
from jvcprojector.command.command import CS20191
from jvcprojector.device import (
AUTH_SALT,
HEAD_ACK,
HEAD_OP,
HEAD_REF,
HEAD_RES,
PJACK,
PJNAK,
PJNG,
PJOK,
PJREQ,
Device,
)
from jvcprojector.error import JvcProjectorError
from . import IP, PORT, TIMEOUT, cc
@pytest.mark.asyncio
async def test_send_ref(conn: AsyncMock):
"""Test send reference command succeeds."""
conn.readline.side_effect = [
cc(HEAD_ACK, command.Power.code),
cc(HEAD_RES, command.Power.code + "1"),
]
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
await dev.send(cmd)
await dev.disconnect()
assert cmd.ack
assert cmd.ref_value == command.Power.ON
conn.connect.assert_called_once()
conn.write.assert_has_calls([call(PJREQ), call(cc(HEAD_REF, command.Power.code))])
@pytest.mark.asyncio
async def test_send_op(conn: AsyncMock):
"""Test send operation command succeeds."""
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
await dev.send(cmd)
await dev.disconnect()
assert cmd.ack
assert cmd.ref_value is None
conn.connect.assert_called_once()
conn.write.assert_has_calls(
[call(PJREQ), call(cc(HEAD_OP, f"{command.Power.code}1"))]
)
@pytest.mark.asyncio
async def test_send_with_password(conn: AsyncMock):
"""Test send with 10 character password succeeds."""
dev = Device(IP, PORT, TIMEOUT, "passwd7890")
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
await dev.send(cmd)
await dev.disconnect()
conn.write.assert_has_calls(
[
call(PJREQ + b"_passwd7890\x00\x00\x00\x00\x00\x00"),
call(cc(HEAD_OP, f"{command.Power.code}1")),
]
)
@pytest.mark.asyncio
async def test_send_with_password_sha256(conn: AsyncMock):
"""Test send with a projector requiring sha256 hashing."""
conn.read.side_effect = [PJOK, PJNAK, PJACK]
dev = Device(IP, PORT, TIMEOUT, "passwd78901")
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
await dev.send(cmd)
await dev.disconnect()
auth = sha256(f"passwd78901{AUTH_SALT}".encode()).hexdigest().encode()
conn.write.assert_has_calls(
[call(PJREQ + b"_" + auth), call(cc(HEAD_OP, f"{command.Power.code}1"))]
)
@pytest.mark.asyncio
@pytest.mark.parametrize("conn", [{"raise_on_connect": 1}], indirect=True)
async def test_connection_refused_retry(conn: AsyncMock):
"""Test connection refused results in retry."""
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
await dev.send(cmd)
await dev.disconnect()
assert cmd.ack
assert conn.connect.call_count == 2
conn.write.assert_has_calls(
[call(PJREQ), call(cc(HEAD_OP, f"{command.Power.code}1"))]
)
@pytest.mark.asyncio
async def test_connection_busy_retry(conn: AsyncMock):
"""Test handshake busy results in retry."""
conn.read.side_effect = [PJNG, PJOK, PJACK]
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
await dev.send(cmd)
await dev.disconnect()
assert conn.connect.call_count == 2
conn.write.assert_has_calls(
[call(PJREQ), call(cc(HEAD_OP, f"{command.Power.code}1"))]
)
@pytest.mark.asyncio
async def test_connection_bad_handshake_error(conn: AsyncMock):
"""Test bad handshake results in error."""
conn.read.side_effect = [b"BAD"]
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
with pytest.raises(JvcProjectorError):
await dev.send(cmd)
conn.connect.assert_called_once()
assert not cmd.ack
@pytest.mark.asyncio
async def test_send_op_bad_ack_error(conn: AsyncMock):
"""Test send operation with bad ack results in error."""
conn.readline.side_effect = [cc(HEAD_ACK, "ZZ")]
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
cmd.op_value = command.Power.ON
with pytest.raises(JvcProjectorError):
await dev.send(cmd)
conn.connect.assert_called_once()
assert not cmd.ack
@pytest.mark.asyncio
async def test_send_ref_bad_ack_error(conn: AsyncMock):
"""Test send reference with bad ack results in error."""
conn.readline.side_effect = [cc(HEAD_ACK, command.Power.code), cc(HEAD_RES, "ZZ1")]
dev = Device(IP, PORT, TIMEOUT, None)
cmd = command.Power(CS20191)
with pytest.raises(JvcProjectorError):
await dev.send(cmd)
conn.connect.assert_called_once()
assert not cmd.ack
|
