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
|
"""Unit tests for description_cache."""
import asyncio
from unittest.mock import patch
import aiohttp
import defusedxml.ElementTree as DET
import pytest
from async_upnp_client.const import HttpResponse
from async_upnp_client.description_cache import DescriptionCache
from .conftest import UpnpTestRequester
@pytest.mark.asyncio
async def test_fetch_parse_success() -> None:
"""Test properly fetching and parsing a description."""
xml = """<root xmlns="urn:schemas-upnp-org:device-1-0">
<device>
<deviceType>urn:schemas-upnp-org:device:TestDevice:1</deviceType>
<UDN>uuid:test_udn</UDN>
</device>
</root>"""
requester = UpnpTestRequester(
{("GET", "http://192.168.1.1/desc.xml"): HttpResponse(200, {}, xml)}
)
description_cache = DescriptionCache(requester)
descr_xml = await description_cache.async_get_description_xml(
"http://192.168.1.1/desc.xml"
)
assert descr_xml == xml
descr_dict = await description_cache.async_get_description_dict(
"http://192.168.1.1/desc.xml"
)
assert descr_dict == {
"deviceType": "urn:schemas-upnp-org:device:TestDevice:1",
"UDN": "uuid:test_udn",
}
@pytest.mark.asyncio
async def test_fetch_parse_success_invalid_chars() -> None:
"""Test fail parsing a description with invalid characters."""
xml = """<root xmlns="urn:schemas-upnp-org:device-1-0">
<device>
<deviceType>urn:schemas-upnp-org:device:TestDevice:1</deviceType>
<UDN>uuid:test_udn</UDN>
<serialNumber>\xff\xff\xff\xff</serialNumber>
</device>
</root>"""
requester = UpnpTestRequester(
{("GET", "http://192.168.1.1/desc.xml"): HttpResponse(200, {}, xml)}
)
description_cache = DescriptionCache(requester)
descr_xml = await description_cache.async_get_description_xml(
"http://192.168.1.1/desc.xml"
)
assert descr_xml == xml
descr_dict = await description_cache.async_get_description_dict(
"http://192.168.1.1/desc.xml"
)
assert descr_dict == {
"deviceType": "urn:schemas-upnp-org:device:TestDevice:1",
"UDN": "uuid:test_udn",
"serialNumber": "ÿÿÿÿ",
}
@pytest.mark.asyncio
@pytest.mark.parametrize("exc", [asyncio.TimeoutError, aiohttp.ClientError])
async def test_fetch_fail(exc: Exception) -> None:
"""Test fail fetching a description."""
xml = ""
requester = UpnpTestRequester(
{("GET", "http://192.168.1.1/desc.xml"): HttpResponse(200, {}, xml)}
)
requester.exceptions.append(exc)
description_cache = DescriptionCache(requester)
descr_xml = await description_cache.async_get_description_xml(
"http://192.168.1.1/desc.xml"
)
assert descr_xml is None
descr_dict = await description_cache.async_get_description_dict(
"http://192.168.1.1/desc.xml"
)
assert descr_dict is None
@pytest.mark.asyncio
async def test_parsing_fail_invalid_xml() -> None:
"""Test fail parsing a description with invalid XML."""
xml = """<root xmlns="urn:schemas-upnp-org:device-1-0">INVALIDXML"""
requester = UpnpTestRequester(
{("GET", "http://192.168.1.1/desc.xml"): HttpResponse(200, {}, xml)}
)
description_cache = DescriptionCache(requester)
descr_xml = await description_cache.async_get_description_xml(
"http://192.168.1.1/desc.xml"
)
assert descr_xml == xml
descr_dict = await description_cache.async_get_description_dict(
"http://192.168.1.1/desc.xml"
)
assert descr_dict is None
@pytest.mark.asyncio
async def test_parsing_fail_error() -> None:
"""Test fail parsing a description with invalid XML."""
xml = ""
requester = UpnpTestRequester(
{("GET", "http://192.168.1.1/desc.xml"): HttpResponse(200, {}, xml)}
)
description_cache = DescriptionCache(requester)
descr_xml = await description_cache.async_get_description_xml(
"http://192.168.1.1/desc.xml"
)
assert descr_xml == xml
with patch(
"async_upnp_client.description_cache.DET.fromstring",
side_effect=DET.ParseError,
):
descr_dict = await description_cache.async_get_description_dict(
"http://192.168.1.1/desc.xml"
)
assert descr_dict is None
|
