-
-
Notifications
You must be signed in to change notification settings - Fork 253
Expand file tree
/
Copy pathtest_httpclient.py
More file actions
99 lines (86 loc) · 2.65 KB
/
test_httpclient.py
File metadata and controls
99 lines (86 loc) · 2.65 KB
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
import re
import aiohttp
import pytest
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import (
KasaException,
TimeoutError,
_ConnectionError,
)
from kasa.httpclient import HttpClient
@pytest.mark.parametrize(
("error", "error_raises", "error_message"),
[
(
aiohttp.ServerDisconnectedError(),
_ConnectionError,
"Device connection error: ",
),
(
aiohttp.ClientOSError(),
_ConnectionError,
"Device connection error: ",
),
(
aiohttp.ServerTimeoutError(),
TimeoutError,
"Unable to query the device, timed out: ",
),
(
TimeoutError(),
TimeoutError,
"Unable to query the device, timed out: ",
),
(Exception(), KasaException, "Unable to query the device: "),
(
aiohttp.ServerFingerprintMismatch(b"exp", b"got", "host", 1),
KasaException,
"Unable to query the device: ",
),
],
ids=(
"ServerDisconnectedError",
"ClientOSError",
"ServerTimeoutError",
"TimeoutError",
"Exception",
"ServerFingerprintMismatch",
),
)
@pytest.mark.parametrize("mock_read", [False, True], ids=("post", "read"))
async def test_httpclient_errors(mocker, error, error_raises, error_message, mock_read):
class _mock_response:
def __init__(self, status, error):
self.status = status
self.error = error
self.call_count = 0
async def __aenter__(self):
return self
async def __aexit__(self, exc_t, exc_v, exc_tb):
pass
async def read(self):
self.call_count += 1
raise self.error
mock_response = _mock_response(200, error)
async def _post(url, *_, **__):
nonlocal mock_response
return mock_response
host = "127.0.0.1"
side_effect = _post if mock_read else error
conn = mocker.patch.object(aiohttp.ClientSession, "post", side_effect=side_effect)
client = HttpClient(DeviceConfig(host))
# Exceptions with parameters print with double quotes, without use single quotes
full_msg = (
re.escape("(")
+ "['\"]"
+ re.escape(f"{error_message}{host}: {error}")
+ "['\"]"
+ re.escape(f", {repr(error)})")
)
with pytest.raises(error_raises, match=error_message) as exc_info:
await client.post("http://foobar")
assert re.match(full_msg, str(exc_info.value))
if mock_read:
assert mock_response.call_count == 1
else:
assert conn.call_count == 1