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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
|
import datetime
import json
import os
import shutil
import tempfile
import unittest
from unittest import mock
import pytest
from chirp import chirp_common
from chirp.sources import repeaterbook
class TestRepeaterbook(unittest.TestCase):
def fake_config(self, fn):
return os.path.join(self.tempdir, fn)
def setUp(self):
self.tempdir = tempfile.mkdtemp()
self.patches = []
self.patches.append(mock.patch('chirp.platform.Platform.config_file',
self.fake_config))
for p in self.patches:
p.start()
self.testfile = os.path.join(os.path.dirname(__file__),
'rb-united_states-oregon.json')
def tearDown(self):
for p in self.patches:
p.stop()
shutil.rmtree(self.tempdir)
@pytest.mark.network
def _test_get_state(self, state, coords=(None, None)):
rb = repeaterbook.RepeaterBook()
self.assertRaises(IndexError, rb.get_memory, 0)
rb.do_fetch(mock.MagicMock(), {
'country': 'United States',
'state': state,
'lat': coords[0],
'lon': coords[1],
'dist': 100,
'openonly': False,
'cached': False,
})
m = rb.get_memory(0)
self.assertIsInstance(m, chirp_common.Memory)
f = rb.get_features()
self.assertGreater(sum(f.memory_bounds), 20)
for i in range(*f.memory_bounds):
m = rb.get_memory(i)
if m.mode == 'DV':
self.assertIsInstance(m, chirp_common.DVMemory)
self.assertEqual('CQCQCQ ', m.dv_urcall)
self.assertEqual(8, len(m.dv_rpt1call))
self.assertEqual(8, len(m.dv_rpt2call))
self.assertEqual(m.dv_rpt1call, m.dv_rpt2call)
self.assertNotEqual('', m.dv_rpt1call)
break
else:
raise Exception('Did not find any DV results')
@pytest.mark.network
def test_get_oregon(self):
self._test_get_state('Oregon', (45, -122))
@pytest.mark.network
def test_get_california(self):
self._test_get_state('California')
def _test_get_gmrs(self, state, coords=(None, None)):
rb = repeaterbook.RepeaterBook()
self.assertRaises(IndexError, rb.get_memory, 0)
rb.do_fetch(mock.MagicMock(), {
'country': 'United States',
'state': state,
'lat': coords[0],
'lon': coords[1],
'dist': 100,
'service': 'gmrs',
'openonly': False,
'cached': False,
})
m = rb.get_memory(0)
self.assertIsInstance(m, chirp_common.Memory)
f = rb.get_features()
self.assertGreater(sum(f.memory_bounds), 1)
@pytest.mark.network
def test_get_california_gmrs(self):
self._test_get_gmrs('California')
@pytest.mark.network
def test_get_australia(self):
rb = repeaterbook.RepeaterBook()
self.assertRaises(IndexError, rb.get_memory, 0)
rb.do_fetch(mock.MagicMock(), {
'country': 'Australia',
'state': 'ALL',
'lat': -26,
'lon': 133,
'dist': 20000,
'openonly': False,
'cached': False,
})
m = rb.get_memory(0)
self.assertIsInstance(m, chirp_common.Memory)
f = rb.get_features()
self.assertGreater(sum(f.memory_bounds), 20)
def _test_with_mocked(self, params):
rb = repeaterbook.RepeaterBook()
with mock.patch.object(rb, 'get_data') as gd:
gd.return_value = self.testfile
rb.do_fetch(mock.MagicMock(), params)
return rb
def test_filter(self):
params = {'country': 'United States',
'state': 'Oregon',
'lat': 45,
'lon': -122,
'dist': 0,
'openonly': False,
'cached': False,
'filter': 'tower'}
rb = self._test_with_mocked(params)
self.assertGreater(sum(rb.get_features().memory_bounds), 5)
def test_distance(self):
params = {'country': 'United States',
'state': 'Oregon',
'lat': 45,
'lon': -122,
'dist': 0,
'openonly': False,
'cached': False,
'filter': 'tower'}
rb1 = self._test_with_mocked(dict(params))
self.assertGreater(sum(rb1.get_features().memory_bounds), 5)
# Make sure we got fewer than everything in Oregon
params['dist'] = 100
rb2 = self._test_with_mocked(dict(params))
self.assertLess(sum(rb2.get_features().memory_bounds),
sum(rb1.get_features().memory_bounds))
# There is nothing 1km away from this spot in the ocean
params['dist'] = 1
params['lat'] = 45
params['lon'] = -140
rb3 = self._test_with_mocked(dict(params))
self.assertEqual(sum(rb3.get_features().memory_bounds), -1)
def test_get_data_500(self):
rb = repeaterbook.RepeaterBook()
with mock.patch('requests.get') as mock_get:
mock_get.return_value.status_code = 500
status = mock.MagicMock()
r = rb.get_data(status, 'US', 'OR', '')
self.assertIsNone(r)
status.send_fail.assert_called()
# Make sure we wrote no files
self.assertEqual([], os.listdir(os.path.join(self.tempdir,
'repeaterbook')))
def test_get_data_json_fail(self):
rb = repeaterbook.RepeaterBook()
with mock.patch('requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.iter_content.return_value = [b'foo']
status = mock.MagicMock()
r = rb.get_data(status, 'US', 'OR', '')
self.assertIsNone(r)
files = os.listdir(os.path.join(self.tempdir,
'repeaterbook'))
# Make sure we only wrote one file and that it is a tempfile
# not one we will find as a data file later
self.assertEqual(1, len(files))
self.assertTrue(files[0].endswith('tmp'))
status.send_fail.assert_called()
def test_get_data_no_results(self):
rb = repeaterbook.RepeaterBook()
with mock.patch('requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.iter_content.return_value = [json.dumps(
{'count': 0}).encode()]
status = mock.MagicMock()
r = rb.get_data(status, 'US', 'OR', '')
self.assertIsNone(r)
files = os.listdir(os.path.join(self.tempdir,
'repeaterbook'))
# Make sure we cleaned up and signaled failure
self.assertEqual(0, len(files))
status.send_fail.assert_called()
def test_get_data_got_results(self):
files = os.listdir(self.tempdir)
# Make sure we started with no data files
self.assertEqual(0, len(files))
rb = repeaterbook.RepeaterBook()
with mock.patch('requests.get') as mock_get:
mock_get.return_value.status_code = 200
mock_get.return_value.iter_content.return_value = [json.dumps(
{'count': 1}).encode()]
status = mock.MagicMock()
r = rb.get_data(status, 'US', 'OR', '')
self.assertIsNotNone(r)
files = os.listdir(os.path.join(self.tempdir,
'repeaterbook'))
# Make sure we left only a data file and did not signal failure
self.assertEqual(1, len(files))
self.assertFalse(files[0].endswith('tmp'))
self.assertEqual(os.path.basename(r), files[0])
status.send_fail.assert_not_called()
def test_get_data_honors_cache_rules(self):
os.mkdir(os.path.join(self.tempdir, 'repeaterbook'))
cache_file = os.path.join(self.tempdir,
'repeaterbook',
os.path.basename(self.testfile))
with open(cache_file, 'w') as f:
f.write('foo')
rb = repeaterbook.RepeaterBook()
with mock.patch('requests.get') as mock_get:
r = rb.get_data(mock.MagicMock(), 'United States', 'Oregon', '')
self.assertEqual(cache_file, r)
# Make sure we returned the cached file
mock_get.assert_not_called()
real_timedelta = datetime.timedelta
real_datetime = datetime.datetime
future = datetime.datetime.now() + datetime.timedelta(days=4)
with mock.patch.object(repeaterbook, 'datetime') as mock_dt:
mock_dt.datetime.fromtimestamp = real_datetime.fromtimestamp
mock_dt.timedelta = real_timedelta
mock_dt.datetime.now.return_value = future
r = rb.get_data(mock.MagicMock(),
'United States', 'Oregon', '')
# Cache file is 4 days old, we should use it
mock_get.assert_not_called()
future = datetime.datetime.now() + datetime.timedelta(days=45)
with mock.patch.object(repeaterbook, 'datetime') as mock_dt:
mock_dt.datetime.fromtimestamp = real_datetime.fromtimestamp
mock_dt.timedelta = real_timedelta
mock_dt.datetime.now.return_value = future
r = rb.get_data(mock.MagicMock(),
'United States', 'Oregon', '')
# Cache file is 45 days old, we should re-fetch
mock_get.assert_called()
|
