strawmelonjuice.com
Forking what is left of ZeroNet and hopefully adding an AT Proto Frontend/Proxy
1import socket
2from urllib.parse import urlparse
3
4import pytest
5import mock
6
7from util import UpnpPunch as upnp
8
9
10@pytest.fixture
11def mock_socket():
12 mock_socket = mock.MagicMock()
13 mock_socket.recv = mock.MagicMock(return_value=b'Hello')
14 mock_socket.bind = mock.MagicMock()
15 mock_socket.send_to = mock.MagicMock()
16
17 return mock_socket
18
19
20@pytest.fixture
21def url_obj():
22 return urlparse('http://192.168.1.1/ctrlPoint.xml')
23
24
25@pytest.fixture(params=['WANPPPConnection', 'WANIPConnection'])
26def igd_profile(request):
27 return """<root><serviceList><service>
28 <serviceType>urn:schemas-upnp-org:service:{}:1</serviceType>
29 <serviceId>urn:upnp-org:serviceId:wanpppc:pppoa</serviceId>
30 <controlURL>/upnp/control/wanpppcpppoa</controlURL>
31 <eventSubURL>/upnp/event/wanpppcpppoa</eventSubURL>
32 <SCPDURL>/WANPPPConnection.xml</SCPDURL>
33</service></serviceList></root>""".format(request.param)
34
35
36@pytest.fixture
37def httplib_response():
38 class FakeResponse(object):
39 def __init__(self, status=200, body='OK'):
40 self.status = status
41 self.body = body
42
43 def read(self):
44 return self.body
45 return FakeResponse
46
47
48class TestUpnpPunch(object):
49 def test_perform_m_search(self, mock_socket):
50 local_ip = '127.0.0.1'
51
52 with mock.patch('util.UpnpPunch.socket.socket',
53 return_value=mock_socket):
54 result = upnp.perform_m_search(local_ip)
55 assert result == 'Hello'
56 assert local_ip == mock_socket.bind.call_args_list[0][0][0][0]
57 assert ('239.255.255.250',
58 1900) == mock_socket.sendto.call_args_list[0][0][1]
59
60 def test_perform_m_search_socket_error(self, mock_socket):
61 mock_socket.recv.side_effect = socket.error('Timeout error')
62
63 with mock.patch('util.UpnpPunch.socket.socket',
64 return_value=mock_socket):
65 with pytest.raises(upnp.UpnpError):
66 upnp.perform_m_search('127.0.0.1')
67
68 def test_retrieve_location_from_ssdp(self, url_obj):
69 ctrl_location = url_obj.geturl()
70 parsed_location = urlparse(ctrl_location)
71 rsp = ('auth: gibberish\r\nlocation: {0}\r\n'
72 'Content-Type: text/html\r\n\r\n').format(ctrl_location)
73 result = upnp._retrieve_location_from_ssdp(rsp)
74 assert result == parsed_location
75
76 def test_retrieve_location_from_ssdp_no_header(self):
77 rsp = 'auth: gibberish\r\nContent-Type: application/json\r\n\r\n'
78 with pytest.raises(upnp.IGDError):
79 upnp._retrieve_location_from_ssdp(rsp)
80
81 def test_retrieve_igd_profile(self, url_obj):
82 with mock.patch('urllib.request.urlopen') as mock_urlopen:
83 upnp._retrieve_igd_profile(url_obj)
84 mock_urlopen.assert_called_with(url_obj.geturl(), timeout=5)
85
86 def test_retrieve_igd_profile_timeout(self, url_obj):
87 with mock.patch('urllib.request.urlopen') as mock_urlopen:
88 mock_urlopen.side_effect = socket.error('Timeout error')
89 with pytest.raises(upnp.IGDError):
90 upnp._retrieve_igd_profile(url_obj)
91
92 def test_parse_igd_profile_service_type(self, igd_profile):
93 control_path, upnp_schema = upnp._parse_igd_profile(igd_profile)
94 assert control_path == '/upnp/control/wanpppcpppoa'
95 assert upnp_schema in ('WANPPPConnection', 'WANIPConnection',)
96
97 def test_parse_igd_profile_no_ctrlurl(self, igd_profile):
98 igd_profile = igd_profile.replace('controlURL', 'nope')
99 with pytest.raises(upnp.IGDError):
100 control_path, upnp_schema = upnp._parse_igd_profile(igd_profile)
101
102 def test_parse_igd_profile_no_schema(self, igd_profile):
103 igd_profile = igd_profile.replace('Connection', 'nope')
104 with pytest.raises(upnp.IGDError):
105 control_path, upnp_schema = upnp._parse_igd_profile(igd_profile)
106
107 def test_create_open_message_parsable(self):
108 from xml.parsers.expat import ExpatError
109 msg, _ = upnp._create_open_message('127.0.0.1', 8888)
110 try:
111 upnp.parseString(msg)
112 except ExpatError as e:
113 pytest.fail('Incorrect XML message: {}'.format(e))
114
115 def test_create_open_message_contains_right_stuff(self):
116 settings = {'description': 'test desc',
117 'protocol': 'test proto',
118 'upnp_schema': 'test schema'}
119 msg, fn_name = upnp._create_open_message('127.0.0.1', 8888, **settings)
120 assert fn_name == 'AddPortMapping'
121 assert '127.0.0.1' in msg
122 assert '8888' in msg
123 assert settings['description'] in msg
124 assert settings['protocol'] in msg
125 assert settings['upnp_schema'] in msg
126
127 def test_parse_for_errors_bad_rsp(self, httplib_response):
128 rsp = httplib_response(status=500)
129 with pytest.raises(upnp.IGDError) as err:
130 upnp._parse_for_errors(rsp)
131 assert 'Unable to parse' in str(err.value)
132
133 def test_parse_for_errors_error(self, httplib_response):
134 soap_error = ('<document>'
135 '<errorCode>500</errorCode>'
136 '<errorDescription>Bad request</errorDescription>'
137 '</document>')
138 rsp = httplib_response(status=500, body=soap_error)
139 with pytest.raises(upnp.IGDError) as err:
140 upnp._parse_for_errors(rsp)
141 assert 'SOAP request error' in str(err.value)
142
143 def test_parse_for_errors_good_rsp(self, httplib_response):
144 rsp = httplib_response(status=200)
145 assert rsp == upnp._parse_for_errors(rsp)
146
147 def test_send_requests_success(self):
148 with mock.patch(
149 'util.UpnpPunch._send_soap_request') as mock_send_request:
150 mock_send_request.return_value = mock.MagicMock(status=200)
151 upnp._send_requests(['msg'], None, None, None)
152
153 assert mock_send_request.called
154
155 def test_send_requests_failed(self):
156 with mock.patch(
157 'util.UpnpPunch._send_soap_request') as mock_send_request:
158 mock_send_request.return_value = mock.MagicMock(status=500)
159 with pytest.raises(upnp.UpnpError):
160 upnp._send_requests(['msg'], None, None, None)
161
162 assert mock_send_request.called
163
164 def test_collect_idg_data(self):
165 pass
166
167 @mock.patch('util.UpnpPunch._get_local_ips')
168 @mock.patch('util.UpnpPunch._collect_idg_data')
169 @mock.patch('util.UpnpPunch._send_requests')
170 def test_ask_to_open_port_success(self, mock_send_requests,
171 mock_collect_idg, mock_local_ips):
172 mock_collect_idg.return_value = {'upnp_schema': 'schema-yo'}
173 mock_local_ips.return_value = ['192.168.0.12']
174
175 result = upnp.ask_to_open_port(retries=5)
176
177 soap_msg = mock_send_requests.call_args[0][0][0][0]
178
179 assert result is True
180
181 assert mock_collect_idg.called
182 assert '192.168.0.12' in soap_msg
183 assert '15441' in soap_msg
184 assert 'schema-yo' in soap_msg
185
186 @mock.patch('util.UpnpPunch._get_local_ips')
187 @mock.patch('util.UpnpPunch._collect_idg_data')
188 @mock.patch('util.UpnpPunch._send_requests')
189 def test_ask_to_open_port_failure(self, mock_send_requests,
190 mock_collect_idg, mock_local_ips):
191 mock_local_ips.return_value = ['192.168.0.12']
192 mock_collect_idg.return_value = {'upnp_schema': 'schema-yo'}
193 mock_send_requests.side_effect = upnp.UpnpError()
194
195 with pytest.raises(upnp.UpnpError):
196 upnp.ask_to_open_port()
197
198 @mock.patch('util.UpnpPunch._collect_idg_data')
199 @mock.patch('util.UpnpPunch._send_requests')
200 def test_orchestrate_soap_request(self, mock_send_requests,
201 mock_collect_idg):
202 soap_mock = mock.MagicMock()
203 args = ['127.0.0.1', 31337, soap_mock, 'upnp-test', {'upnp_schema':
204 'schema-yo'}]
205 mock_collect_idg.return_value = args[-1]
206
207 upnp._orchestrate_soap_request(*args[:-1])
208
209 assert mock_collect_idg.called
210 soap_mock.assert_called_with(
211 *args[:2] + ['upnp-test', 'UDP', 'schema-yo'])
212 assert mock_send_requests.called
213
214 @mock.patch('util.UpnpPunch._collect_idg_data')
215 @mock.patch('util.UpnpPunch._send_requests')
216 def test_orchestrate_soap_request_without_desc(self, mock_send_requests,
217 mock_collect_idg):
218 soap_mock = mock.MagicMock()
219 args = ['127.0.0.1', 31337, soap_mock, {'upnp_schema': 'schema-yo'}]
220 mock_collect_idg.return_value = args[-1]
221
222 upnp._orchestrate_soap_request(*args[:-1])
223
224 assert mock_collect_idg.called
225 soap_mock.assert_called_with(*args[:2] + [None, 'UDP', 'schema-yo'])
226 assert mock_send_requests.called
227
228 def test_create_close_message_parsable(self):
229 from xml.parsers.expat import ExpatError
230 msg, _ = upnp._create_close_message('127.0.0.1', 8888)
231 try:
232 upnp.parseString(msg)
233 except ExpatError as e:
234 pytest.fail('Incorrect XML message: {}'.format(e))
235
236 def test_create_close_message_contains_right_stuff(self):
237 settings = {'protocol': 'test proto',
238 'upnp_schema': 'test schema'}
239 msg, fn_name = upnp._create_close_message('127.0.0.1', 8888, **
240 settings)
241 assert fn_name == 'DeletePortMapping'
242 assert '8888' in msg
243 assert settings['protocol'] in msg
244 assert settings['upnp_schema'] in msg
245
246 @mock.patch('util.UpnpPunch._get_local_ips')
247 @mock.patch('util.UpnpPunch._orchestrate_soap_request')
248 def test_communicate_with_igd_success(self, mock_orchestrate,
249 mock_get_local_ips):
250 mock_get_local_ips.return_value = ['192.168.0.12']
251 upnp._communicate_with_igd()
252 assert mock_get_local_ips.called
253 assert mock_orchestrate.called
254
255 @mock.patch('util.UpnpPunch._get_local_ips')
256 @mock.patch('util.UpnpPunch._orchestrate_soap_request')
257 def test_communicate_with_igd_succeed_despite_single_failure(
258 self, mock_orchestrate, mock_get_local_ips):
259 mock_get_local_ips.return_value = ['192.168.0.12']
260 mock_orchestrate.side_effect = [upnp.UpnpError, None]
261 upnp._communicate_with_igd(retries=2)
262 assert mock_get_local_ips.called
263 assert mock_orchestrate.called
264
265 @mock.patch('util.UpnpPunch._get_local_ips')
266 @mock.patch('util.UpnpPunch._orchestrate_soap_request')
267 def test_communicate_with_igd_total_failure(self, mock_orchestrate,
268 mock_get_local_ips):
269 mock_get_local_ips.return_value = ['192.168.0.12']
270 mock_orchestrate.side_effect = [upnp.UpnpError, upnp.IGDError]
271 with pytest.raises(upnp.UpnpError):
272 upnp._communicate_with_igd(retries=2)
273 assert mock_get_local_ips.called
274 assert mock_orchestrate.called