1 | #!/usr/bin/env python |
||
2 | # -*- coding: utf-8 -*- |
||
3 | |||
4 | """Tests for Python APRS-IS Bindings.""" |
||
5 | |||
6 | # These imports are for python3 compatibility inside python2 |
||
7 | from __future__ import absolute_import |
||
8 | from __future__ import division |
||
9 | from __future__ import print_function |
||
10 | |||
11 | import unittest |
||
12 | |||
13 | import six |
||
14 | |||
15 | import apex.aprs.constants |
||
16 | import apex.aprs.igate |
||
17 | |||
18 | from .kiss_mock import KissMock |
||
19 | |||
20 | if six.PY2: |
||
21 | import httpretty |
||
22 | |||
23 | __author__ = 'Jeffrey Phillips Freeman (WI2ARD)' |
||
24 | __maintainer__ = 'Jeffrey Phillips Freeman (WI2ARD)' |
||
25 | __email__ = '[email protected]' |
||
26 | __license__ = 'Apache License, Version 2.0' |
||
27 | __copyright__ = 'Copyright 2016, Syncleus, Inc. and contributors' |
||
28 | __credits__ = [] |
||
29 | __version__ = '0.0.2' |
||
30 | |||
31 | ALPHABET = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ' |
||
32 | NUMBERS = '0123456789' |
||
33 | POSITIVE_NUMBERS = NUMBERS[1:] |
||
34 | ALPHANUM = ''.join([ALPHABET, NUMBERS]) |
||
35 | |||
36 | DECODED_CALLSIGN = callsign = {'callsign': 'W2GMD', 'ssid': 1} |
||
37 | ENCODED_CALLSIGN = [174, 100, 142, 154, 136, 64, 98] |
||
38 | |||
39 | DECODED_CALLSIGN_DIGIPEATED = {'callsign': 'W2GMD*', 'ssid': 1} |
||
40 | ENCODED_CALLSIGN_DIGIPEATED = [174, 100, 142, 154, 136, 64, 226] |
||
41 | |||
42 | DECODED_FRAME = { |
||
43 | 'source': 'W2GMD-1', |
||
44 | 'destination': 'OMG', |
||
45 | 'path': ['WIDE1-1'], |
||
46 | 'text': 'test_encode_frame' |
||
47 | } |
||
48 | ENCODED_FRAME = [158, 154, 142, 64, 64, 64, 96, 174, 100, 142, 154, 136, 64, 98, 174, 146, 136, 138, 98, 64, 99, 3, 240, |
||
49 | 116, 101, 115, 116, 95, 101, 110, 99, 111, 100, 101, 95, 102, 114, 97, 109, 101] |
||
50 | |||
51 | DECODED_FRAME_RECORDED = { |
||
52 | 'source': 'W2GMD-6', |
||
53 | 'destination': 'APRX24', |
||
54 | 'path': ['WIDE1-1'], |
||
55 | 'text': ('!3745.75NI12228.05W#W2GMD-6 Inner Sunset, ' |
||
56 | 'SF iGate/Digipeater http://w2gmd.org') |
||
57 | } |
||
58 | ENCODED_FRAME_RECORDED = [130, 160, 164, 176, 100, 104, 96, 174, 100, 142, 154, 136, 64, 108, 174, 146, 136, 138, 98, |
||
59 | 64, 99, 3, 240, 33, 51, 55, 52, 53, 46, 55, 53, 78, 73, 49, 50, 50, 50, 56, 46, 48, 53, 87, |
||
60 | 35, 87, 50, 71, 77, 68, 45, 54, 32, 73, 110, 110, 101, 114, 32, 83, 117, 110, 115, 101, 116, |
||
61 | 44, 32, 83, 70, 32, 105, 71, 97, 116, 101, 47, 68, 105, 103, 105, 112, 101, 97, 116, 101, 114, |
||
62 | 32, 104, 116, 116, 112, 58, 47, 47, 119, 50, 103, 109, 100, 46, 111, 114, 103] |
||
63 | |||
64 | DECODED_FRAME_MULTIPATH = { |
||
65 | 'source': 'W2GMD-1', |
||
66 | 'destination': 'OMG', |
||
67 | 'path': ['WIDE1-1', 'WIDE2-2'], |
||
68 | 'text': 'test_encode_frame' |
||
69 | } |
||
70 | ENCODED_FRAME_MULTIPATH = [158, 154, 142, 64, 64, 64, 96, 174, 100, 142, 154, 136, 64, 98, 174, 146, 136, 138, 98, 64, |
||
71 | 98, 174, 146, 136, 138, 100, 64, 101, 3, 240, 116, 101, 115, 116, 95, 101, 110, 99, 111, 100, |
||
72 | 101, 95, 102, 114, 97, 109, 101] |
||
73 | |||
74 | DECODED_FRAME_KISS = { |
||
75 | 'source': 'W2GMD-1', |
||
76 | 'destination': 'OMG', |
||
77 | 'path': ['WIDE1-1', 'WIDE2-2'], |
||
78 | 'text': 'test_encode_frame' |
||
79 | } |
||
80 | ENCODED_FRAME_KISS = [192, 0, 158, 154, 142, 64, 64, 64, 96, 174, 100, 142, 154, 136, 64, 98, 174, 146, 136, 138, 98, |
||
81 | 64, 98, 174, 146, 136, 138, 100, 64, 101, 3, 240, 116, 101, 115, 116, 95, 101, 110, 99, 111, 100, |
||
82 | 101, 95, 102, 114, 97, 109, 101, 192] |
||
83 | |||
84 | DECODED_FRAME_KISS_INVALID = { |
||
85 | 'source': 'KG6WTF', |
||
86 | 'destination': 'S7TSUV', |
||
87 | 'path': ['MTOSO-2', 'WIDE2*' 'qAR', 'KF6FIR-10'], |
||
88 | 'text': '`17El#X-/[email protected]' |
||
89 | } |
||
90 | ENCODED_FRAME_KISS_INVALID = [192, 0, 166, 110, 168, 166, 170, 172, 96, 150, 142, 108, 174, 168, 140, 96, 154, 168, 158, |
||
91 | 166, 158, 64, 100, 174, 146, 136, 138, 100, 226, 130, 164, 224, 150, 140, 108, 140, 146, 164, |
||
92 | 117, 3, 240, 96, 49, 55, 69, 108, 35, 88, 45, 47, 107, 103, 54, 119, 116, 102, 64, 103, 111, |
||
93 | 115, 115, 101, 108, 105, 110, 102, 97, 109, 105, 108, 121, 46, 99, 111, 109, 192] |
||
94 | |||
95 | |||
96 | |||
97 | class AprsTest(unittest.TestCase): # pylint: disable=R0904 |
||
98 | """Tests for Python APRS-IS Bindings.""" |
||
99 | |||
100 | def setUp(self): # pylint: disable=C0103 |
||
101 | self.fake_server = 'http://localhost:5567/' |
||
102 | self.fake_callsign = 'KWN4YGH-5' |
||
103 | |||
104 | def test_read_frame_kiss(self): |
||
105 | kiss_mock = KissMock() |
||
106 | aprs_kiss = apex.aprs.AprsKiss(kiss_mock) |
||
107 | |||
108 | kiss_mock.clear_interface() |
||
109 | kiss_mock.add_read_from_interface(ENCODED_FRAME_KISS) |
||
110 | translated_frame = None |
||
111 | iter_left = 1000 |
||
112 | while not translated_frame and iter_left > 0: |
||
113 | translated_frame = aprs_kiss.read() |
||
114 | iter_left -= 1 |
||
115 | |||
116 | self.assertEqual(DECODED_FRAME_KISS, translated_frame) |
||
117 | |||
118 | def test_write_frame_kiss(self): |
||
119 | kiss_mock = KissMock() |
||
120 | aprs_kiss = apex.aprs.AprsKiss(kiss_mock) |
||
121 | |||
122 | kiss_mock.clear_interface() |
||
123 | aprs_kiss.write(DECODED_FRAME_KISS) |
||
124 | all_raw_frames = kiss_mock.get_sent_to_interface() |
||
125 | |||
126 | self.assertEqual(ENCODED_FRAME_KISS, all_raw_frames[0]) |
||
127 | |||
128 | def test_read_frame_kiss_invalid(self): |
||
129 | kiss_mock = KissMock() |
||
130 | aprs_kiss = apex.aprs.AprsKiss(kiss_mock) |
||
131 | |||
132 | kiss_mock.clear_interface() |
||
133 | kiss_mock.add_read_from_interface(ENCODED_FRAME_KISS_INVALID) |
||
134 | translated_frame = None |
||
135 | iter_left = 1000 |
||
136 | while not translated_frame and iter_left > 0: |
||
137 | translated_frame = aprs_kiss.read() |
||
138 | iter_left -= 1 |
||
139 | |||
140 | self.assertEqual(None, translated_frame) |
||
141 | |||
142 | def test_write_frame_kiss_invalid(self): |
||
143 | kiss_mock = KissMock() |
||
144 | aprs_kiss = apex.aprs.AprsKiss(kiss_mock) |
||
145 | |||
146 | kiss_mock.clear_interface() |
||
147 | aprs_kiss.write(DECODED_FRAME_KISS_INVALID) |
||
148 | all_raw_frames = kiss_mock.get_sent_to_interface() |
||
149 | |||
150 | self.assertEqual(0, len(all_raw_frames)) |
||
151 | |||
152 | def test_encode_callsign(self): |
||
153 | """ |
||
154 | Tests encoding a callsign. |
||
155 | """ |
||
156 | encoded_callsign = apex.aprs.AprsKiss._AprsKiss__encode_callsign(DECODED_CALLSIGN) |
||
157 | self.assertEqual(ENCODED_CALLSIGN, encoded_callsign) |
||
158 | |||
159 | def test_encode_callsign_digipeated(self): |
||
160 | """ |
||
161 | Tests encoding a digipeated callsign with |
||
162 | `aprs.util.encode_callsign()`. |
||
163 | """ |
||
164 | encoded_callsign = apex.aprs.AprsKiss._AprsKiss__encode_callsign(DECODED_CALLSIGN_DIGIPEATED) |
||
165 | self.assertEqual(ENCODED_CALLSIGN_DIGIPEATED, encoded_callsign) |
||
166 | |||
167 | def test_decode_callsign(self): |
||
168 | """ |
||
169 | Tests extracting the callsign from a KISS-encoded APRS frame. |
||
170 | """ |
||
171 | decoded_callsign = apex.aprs.AprsKiss._AprsKiss__extract_callsign(ENCODED_CALLSIGN) |
||
172 | self.assertEqual(DECODED_CALLSIGN, decoded_callsign) |
||
173 | |||
174 | def test_decode_callsign_digipeated(self): |
||
175 | """ |
||
176 | Tests extracting the callsign from a KISS-encoded APRS frame. |
||
177 | """ |
||
178 | decoded_callsign = apex.aprs.AprsKiss._AprsKiss__extract_callsign(ENCODED_CALLSIGN_DIGIPEATED) |
||
179 | self.assertEqual(DECODED_CALLSIGN, decoded_callsign) |
||
180 | |||
181 | def test_encode_frame(self): |
||
182 | View Code Duplication | """ |
|
0 ignored issues
–
show
Duplication
introduced
by
![]() |
|||
183 | Tests KISS-encoding an APRS. |
||
184 | """ |
||
185 | encoded_frame = apex.aprs.AprsKiss._AprsKiss__encode_frame(DECODED_FRAME) |
||
186 | self.assertEqual(ENCODED_FRAME, encoded_frame) |
||
187 | |||
188 | def test_encode_frame_recorded(self): |
||
189 | """ |
||
190 | Tests encoding a KISS-encoded APRS. |
||
191 | """ |
||
192 | encoded_frame = apex.aprs.AprsKiss._AprsKiss__encode_frame(DECODED_FRAME_RECORDED) |
||
193 | self.assertEqual(ENCODED_FRAME_RECORDED, encoded_frame) |
||
194 | |||
195 | def test_encode_frame_multipath(self): |
||
196 | """ |
||
197 | Tests encoding a KISS-encoded APRS. |
||
198 | """ |
||
199 | encoded_frame = apex.aprs.AprsKiss._AprsKiss__encode_frame(DECODED_FRAME_MULTIPATH) |
||
200 | self.assertEqual(ENCODED_FRAME_MULTIPATH, encoded_frame) |
||
201 | |||
202 | def test_decode_frame(self): |
||
203 | """ |
||
204 | Tests KISS-encoding an APRS |
||
205 | """ |
||
206 | decoded_frame = apex.aprs.AprsKiss._AprsKiss__decode_frame(ENCODED_FRAME) |
||
207 | self.assertEqual(DECODED_FRAME, decoded_frame) |
||
208 | |||
209 | def test_decode_frame_recorded(self): |
||
210 | View Code Duplication | """ |
|
0 ignored issues
–
show
|
|||
211 | Tests decoding a KISS-encoded APRS frame |
||
212 | """ |
||
213 | decoded_frame = apex.aprs.AprsKiss._AprsKiss__decode_frame(ENCODED_FRAME_RECORDED) |
||
214 | self.assertEqual(DECODED_FRAME_RECORDED, decoded_frame) |
||
215 | |||
216 | def test_decode_frame_multipath(self): |
||
217 | """ |
||
218 | Tests decoding a KISS-encoded APRS frame |
||
219 | """ |
||
220 | decoded_frame = apex.aprs.AprsKiss._AprsKiss__decode_frame(ENCODED_FRAME_MULTIPATH) |
||
221 | self.assertEqual(DECODED_FRAME_MULTIPATH, decoded_frame) |
||
222 | |||
223 | def test_extract_path(self): |
||
224 | """ |
||
225 | Tests extracting the APRS path from a KISS-encoded. |
||
226 | """ |
||
227 | extracted_path = apex.aprs.AprsKiss._AprsKiss__extract_path(3, ENCODED_FRAME) |
||
228 | self.assertEqual(DECODED_FRAME['path'][0], extracted_path[0]) |
||
229 | |||
230 | def test_idwentity_as_string_with_ssid(self): |
||
231 | """ |
||
232 | Tests creating a full callsign string from a callsign+ssid dict using |
||
233 | `aprs.util.full_callsign()`. |
||
234 | """ |
||
235 | callsign = { |
||
236 | 'callsign': 'W2GMD', |
||
237 | 'ssid': 1 |
||
238 | } |
||
239 | full_callsign = apex.aprs.AprsKiss._AprsKiss__identity_as_string(callsign) |
||
240 | self.assertEqual(full_callsign, 'W2GMD-1') |
||
241 | |||
242 | def test_identity_as_string_sans_ssid(self): |
||
243 | """ |
||
244 | Tests creating a full callsign string from a callsign dict using |
||
245 | `aprs.util.full_callsign()`. |
||
246 | """ |
||
247 | callsign = { |
||
248 | 'callsign': 'W2GMD', |
||
249 | 'ssid': 0 |
||
250 | } |
||
251 | full_callsign = apex.aprs.AprsKiss._AprsKiss__identity_as_string(callsign) |
||
252 | self.assertEqual(full_callsign, 'W2GMD') |
||
253 | |||
254 | if six.PY2: |
||
255 | @httpretty.httprettified |
||
256 | def test_fake_good_auth(self): |
||
257 | """ |
||
258 | Tests authenticating against APRS-IS using a valid call+pass. |
||
259 | """ |
||
260 | httpretty.HTTPretty.register_uri( |
||
261 | httpretty.HTTPretty.POST, |
||
262 | self.fake_server, |
||
263 | status=204 |
||
264 | ) |
||
265 | |||
266 | aprs_conn = apex.aprs.igate.IGate( |
||
267 | user=self.fake_callsign, |
||
268 | input_url=self.fake_server |
||
269 | ) |
||
270 | aprs_conn.connect() |
||
271 | |||
272 | msg = { |
||
273 | 'source': self.fake_callsign, |
||
274 | 'destination': 'APRS', |
||
275 | 'path': ['TCPIP*'], |
||
276 | 'text': '=3745.00N/12227.00W-Simulated Location' |
||
277 | } |
||
278 | |||
279 | result = aprs_conn.write(msg) |
||
280 | |||
281 | self.assertTrue(result) |
||
282 | |||
283 | @httpretty.httprettified |
||
284 | def test_fake_bad_auth_http(self): |
||
285 | """ |
||
286 | Tests authenticating against APRS-IS using an invalid call+pass. |
||
287 | """ |
||
288 | httpretty.HTTPretty.register_uri( |
||
289 | httpretty.HTTPretty.POST, |
||
290 | self.fake_server, |
||
291 | status=401 |
||
292 | ) |
||
293 | |||
294 | aprs_conn = apex.aprs.igate.IGate( |
||
295 | user=self.fake_callsign, |
||
296 | input_url=self.fake_server |
||
297 | ) |
||
298 | aprs_conn.connect() |
||
299 | |||
300 | msg = { |
||
301 | 'source': self.fake_callsign, |
||
302 | 'destination': 'APRS', |
||
303 | 'path': ['TCPIP*'], |
||
304 | 'text': '=3745.00N/12227.00W-Simulated Location' |
||
305 | } |
||
306 | |||
307 | result = aprs_conn.write(msg, protocol='HTTP') |
||
308 | |||
309 | self.assertFalse(result) |
||
310 | |||
311 | @unittest.skip('Test only works with real server.') |
||
312 | def test_more(self): |
||
313 | """ |
||
314 | Tests APRS-IS binding against a real APRS-IS server. |
||
315 | """ |
||
316 | aprs_conn = apex.aprs.igate.IGate( |
||
317 | user=self.real_callsign, |
||
318 | input_url=self.real_server |
||
319 | ) |
||
320 | aprs_conn.connect() |
||
321 | |||
322 | msg = { |
||
323 | 'source': self.fake_callsign, |
||
324 | 'destination': 'APRS', |
||
325 | 'path': ['TCPIP*'], |
||
326 | 'text': '=3745.00N/12227.00W-Simulated Location' |
||
327 | } |
||
328 | self.logger.debug(locals()) |
||
329 | |||
330 | result = aprs_conn.write(msg) |
||
331 | |||
332 | self.assertFalse(result) |
||
333 |