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
|
"""Define tests for motion detection media fetching."""
from __future__ import annotations
from typing import Any
import aiohttp
import pytest
from aresponses import ResponsesMockServer
from simplipy import API
from simplipy.errors import SimplipyError
from .common import TEST_AUTHORIZATION_CODE, TEST_CODE_VERIFIER
# Used in a testcase that counts requests to simulate a delayed
# media fetch
COUNT = 0 # pylint: disable=global-statement
@pytest.mark.asyncio
async def test_media_file_fetching(
aresponses: ResponsesMockServer,
authenticated_simplisafe_server_v3: ResponsesMockServer,
) -> None:
"""Test the media fetching method."""
my_string = "this is an image"
content = my_string.encode("utf-8")
authenticated_simplisafe_server_v3.add(
"remix.us-east-1.prd.cam.simplisafe.com",
"/v1/preview/normal",
"get",
aresponses.Response(body=content, status=200),
)
authenticated_simplisafe_server_v3.add(
"remix.us-east-1.prd.cam.simplisafe.com",
"/v1/preview/timeout",
"get",
aresponses.Response(status=404),
repeat=5,
)
# pylint: disable-next=unused-argument
def delayed(request: Any) -> aresponses.Response:
"""Return a 404 a few times, then a 200."""
global COUNT # pylint: disable=global-statement
if COUNT >= 3:
return aresponses.Response(body=content, status=200)
COUNT = COUNT + 1
return aresponses.Response(status=404)
authenticated_simplisafe_server_v3.add(
"remix.us-east-1.prd.cam.simplisafe.com",
"/v1/preview/delayed",
"get",
response=delayed,
repeat=5,
)
async with authenticated_simplisafe_server_v3, aiohttp.ClientSession() as session:
simplisafe = await API.async_from_auth(
TEST_AUTHORIZATION_CODE, TEST_CODE_VERIFIER, session=session
)
# simple fetch
res = await simplisafe.async_media(
url="https://remix.us-east-1.prd.cam.simplisafe.com/v1/preview/normal"
)
assert res == content
# timeout with error
with pytest.raises(SimplipyError):
await simplisafe.async_media(
url="https://remix.us-east-1.prd.cam.simplisafe.com/v1/preview/timeout"
)
# test retries
res = await simplisafe.async_media(
url="https://remix.us-east-1.prd.cam.simplisafe.com/v1/preview/delayed"
)
assert res == content
aresponses.assert_plan_strictly_followed()
@pytest.mark.asyncio
async def test_media_file_enabe_disable_retires(
aresponses: ResponsesMockServer,
authenticated_simplisafe_server_v3: ResponsesMockServer,
) -> None:
"""Test the ability to enable/disable retries."""
my_string = "this is an image"
content = my_string.encode("utf-8")
authenticated_simplisafe_server_v3.add(
"remix.us-east-1.prd.cam.simplisafe.com",
"/v1/preview/timeout",
"get",
aresponses.Response(status=404),
repeat=2,
)
authenticated_simplisafe_server_v3.add(
"remix.us-east-1.prd.cam.simplisafe.com",
"/v1/preview/timeout",
"get",
aresponses.Response(body=content, status=200),
repeat=1,
)
async with authenticated_simplisafe_server_v3, aiohttp.ClientSession() as session:
simplisafe = await API.async_from_auth(
TEST_AUTHORIZATION_CODE, TEST_CODE_VERIFIER, session=session
)
# With retries disabled, the first attempt will raise an error
simplisafe.disable_request_retries()
with pytest.raises(SimplipyError):
await simplisafe.async_media(
url="https://remix.us-east-1.prd.cam.simplisafe.com/v1/preview/timeout"
)
# When re-enabled, there will be one 404 followed by a 200, no error
simplisafe.enable_request_retries()
res = await simplisafe.async_media(
url="https://remix.us-east-1.prd.cam.simplisafe.com/v1/preview/timeout"
)
assert res == content
aresponses.assert_plan_strictly_followed()
|
