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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
"""Test api functions."""
from unittest import mock
from unittest import IsolatedAsyncioTestCase
from blinkpy import api
from blinkpy.blinkpy import Blink, util
from blinkpy.auth import Auth
import tests.mock_responses as mresp
COMMAND_RESPONSE = {"network_id": "12345", "id": "54321"}
COMMAND_COMPLETE = {"complete": True, "status_code": 908}
COMMAND_COMPLETE_BAD = {"complete": True, "status_code": 999}
COMMAND_NOT_COMPLETE = {"complete": False, "status_code": 908}
COMMAND_DONE = {"message": "Command 1234567890 set to :done", "code": 902}
@mock.patch("blinkpy.auth.Auth.query")
class TestAPI(IsolatedAsyncioTestCase):
"""Test the API class in blinkpy."""
async def asyncSetUp(self):
"""Set up Login Handler."""
self.blink = Blink(session=mock.AsyncMock())
self.auth = Auth()
self.blink.available = True
self.blink.urls = util.BlinkURLHandler("region_id")
self.auth.account_id = 1234
self.auth.client_id = 5678
def tearDown(self):
"""Clean up after test."""
self.blink = None
self.auth = None
async def test_request_logout(self, mock_resp):
"""Test request_logout."""
mock_resp.return_value = mresp.MockResponse({}, 200)
response = await api.request_logout(self.blink)
self.assertEqual(response.status, 200)
async def test_request_networks(self, mock_resp):
"""Test request networks."""
mock_resp.return_value = {"networks": "1234"}
self.assertEqual(await api.request_networks(self.blink), {"networks": "1234"})
async def test_request_user(self, mock_resp):
"""Test request_user."""
mock_resp.return_value = {"user": "userid"}
self.assertEqual(await api.request_user(self.blink), {"user": "userid"})
async def test_request_network_status(self, mock_resp):
"""Test request network status."""
mock_resp.return_value = {"user": "userid"}
self.assertEqual(
await api.request_network_status(self.blink, "network"), {"user": "userid"}
)
async def test_request_command_status(self, mock_resp):
"""Test command_status."""
mock_resp.side_effect = ({"command": "done"}, COMMAND_COMPLETE)
self.assertEqual(
await api.request_command_status(self.blink, "network", "command"),
{"command": "done"},
)
async def test_request_command_done(self, mock_resp):
"""Test command_done."""
mock_resp.return_value = mresp.MockResponse(COMMAND_DONE, 200)
response = await api.request_command_done(self.blink, "network", "1234567890")
self.assertEqual(response.status, 200)
self.assertEqual(await response.json(), COMMAND_DONE)
async def test_request_new_image(self, mock_resp):
"""Test api request new image."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_new_image(self.blink, "network", "camera")
self.assertEqual(response.status, 200)
async def test_request_new_video(self, mock_resp):
"""Test api request new Video."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_new_video(self.blink, "network", "camera")
self.assertEqual(response.status, 200)
async def test_request_video_count(self, mock_resp):
"""Test api request video count."""
mock_resp.return_value = {"count": "10"}
self.assertEqual(await api.request_video_count(self.blink), {"count": "10"})
async def test_request_cameras(self, mock_resp):
"""Test api request cameras."""
mock_resp.return_value = {"cameras": {"camera_id": 1}}
self.assertEqual(
await api.request_cameras(self.blink, "network"),
{"cameras": {"camera_id": 1}},
)
async def test_request_camera_usage(self, mock_resp):
"""Test api request cameras."""
mock_resp.return_value = {"cameras": "1111"}
self.assertEqual(
await api.request_camera_usage(self.blink), {"cameras": "1111"}
)
async def test_request_notification_flags(self, mock_resp):
"""Test notification flag request."""
mock_resp.return_value = {"notifications": {"some_key": False}}
self.assertEqual(
await api.request_notification_flags(self.blink),
{"notifications": {"some_key": False}},
)
async def test_request_set_notification_flag(self, mock_resp):
"""Test set of notification flags."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_set_notification_flag(self.blink, {})
self.assertEqual(response.status, 200)
async def test_request_motion_detection_enable(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_motion_detection_enable(
self.blink, "network", "camera"
)
self.assertEqual(response.status, 200)
async def test_request_motion_detection_disable(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_motion_detection_disable(
self.blink, "network", "camera"
)
self.assertEqual(response.status, 200)
async def test_request_local_storage_clip(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.side_effect = (
mresp.MockResponse(COMMAND_RESPONSE, 200),
COMMAND_COMPLETE,
)
response = await api.request_local_storage_clip(
self.blink, "network", "sync_id", "manifest_id", "clip_id"
)
self.assertEqual(response.status, 200)
async def test_request_get_config(self, mock_resp):
"""Test request get config."""
mock_resp.return_value = {"config": "values"}
self.assertEqual(
await api.request_get_config(self.blink, "network", "camera_id", "owl"),
{"config": "values"},
)
self.assertEqual(
await api.request_get_config(
self.blink, "network", "camera_id", "catalina"
),
{"config": "values"},
)
async def test_request_update_config(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.return_value = mresp.MockResponse(COMMAND_RESPONSE, 200)
response = await api.request_update_config(
self.blink, "network", "camera_id", "owl"
)
self.assertEqual(response.status, 200)
response = await api.request_update_config(
self.blink, "network", "camera_id", "catalina"
)
self.assertEqual(response.status, 200)
self.assertIsNone(
await api.request_update_config(
self.blink, "network", "camera_id", "other_camera"
)
)
async def test_wait_for_command(self, mock_resp):
"""Test Motion detect enable."""
mock_resp.side_effect = (COMMAND_NOT_COMPLETE, COMMAND_COMPLETE)
response = await api.wait_for_command(self.blink, COMMAND_RESPONSE)
assert response
# mock_resp.side_effect = (COMMAND_NOT_COMPLETE, COMMAND_NOT_COMPLETE, None)
# response = await api.wait_for_command(self.blink, COMMAND_RESPONSE)
# self.assertFalse(response)
mock_resp.side_effect = (COMMAND_COMPLETE_BAD, {})
response = await api.wait_for_command(self.blink, COMMAND_RESPONSE)
self.assertFalse(response)
response = await api.wait_for_command(self.blink, None)
self.assertFalse(response)
|
