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
|
"""Test pydeCONZ utilities.
pytest --cov-report term-missing --cov=pydeconz.utils tests/test_utils.py
"""
from __future__ import annotations
from unittest.mock import AsyncMock, Mock, patch
import aiohttp
import pytest
from pydeconz import errors, utils
API_KEY = "1234567890"
IP = "127.0.0.1"
PORT = "80"
async def test_delete_api_key() -> None:
"""Test a successful call of delete_api_key."""
session = Mock()
with patch("pydeconz.utils.request", new=AsyncMock(return_value=True)):
await utils.delete_api_key(session, IP, PORT, API_KEY)
async def test_delete_all_keys() -> None:
"""Test a successful call of delete_all_keys.
Delete all keys doesn't care what happens with delete_api_key.
"""
session = Mock()
with patch(
"pydeconz.utils.request",
new=AsyncMock(return_value={"whitelist": {1: "123", 2: "456"}}),
):
await utils.delete_all_keys(session, IP, PORT, API_KEY)
async def test_get_bridge_id() -> None:
"""Test a successful call of get_bridgeid."""
session = Mock()
with patch(
"pydeconz.utils.request",
new=AsyncMock(return_value={"bridgeid": "12345"}),
):
response = await utils.get_bridge_id(session, IP, PORT, API_KEY)
assert response == "12345"
async def test_discovery() -> None:
"""Test a successful call to discovery."""
session = Mock()
with patch(
"pydeconz.utils.request",
new=AsyncMock(
return_value=[
{
"id": "123456FFFFABCDEF",
"internalipaddress": "host1",
"internalport": "port1",
"macaddress": "a:b:c",
"name": "gateway",
},
{
"id": "234567BCDEFG",
"internalipaddress": "host2",
"internalport": "port2",
},
]
),
):
response = await utils.discovery(session)
assert response == [
{
"id": "123456ABCDEF",
"host": "host1",
"port": "port1",
"mac": "a:b:c",
"name": "gateway",
},
{
"id": "234567BCDEFG",
"host": "host2",
"port": "port2",
"mac": "",
"name": "",
},
]
async def test_discovery_response_empty() -> None:
"""Test an empty discovery returns an empty list."""
session = Mock()
with patch("pydeconz.utils.request", new=AsyncMock(return_value={})):
response = await utils.discovery(session)
assert not response
async def test_request() -> None:
"""Test a successful call of request."""
response = Mock()
response.content_type = "application/json"
response.json = AsyncMock(return_value={"json": "response"})
session = AsyncMock(return_value=response)
result = await utils.request(session, "url")
assert result == {"json": "response"}
async def test_request_fails_client_error() -> None:
"""Test a successful call of request."""
session = AsyncMock(side_effect=aiohttp.ClientError)
with pytest.raises(errors.RequestError) as e_info:
await utils.request(session, "url")
assert str(e_info.value) == "Error requesting data from url: "
async def test_request_fails_invalid_content() -> None:
"""Test a successful call of request."""
response = Mock()
response.content_type = "application/binary"
session = AsyncMock(return_value=response)
with pytest.raises(errors.ResponseError) as e_info:
await utils.request(session, "url")
assert str(e_info.value) == "Invalid content type: application/binary"
async def test_request_fails_raise_error() -> None:
"""Test a successful call of request."""
response = Mock()
response.content_type = "application/json"
response.json = AsyncMock(
return_value=[
{"error": {"type": 1, "address": "address", "description": "description"}}
]
)
session = AsyncMock(return_value=response)
with pytest.raises(errors.Unauthorized) as e_info:
await utils.request(session, "url")
assert str(e_info.value) == "1 address description"
|
