mirror of
https://github.com/kk7ds/chirp
synced 2026-08-13 17:51:36 -04:00
f5d57bd578
Fixes: - Store and use ETag to prevent pulling unchanged data sets - Return our cache if we get a failure from the server, if present - Return our cache if we are unable to contact the server at all
294 lines
12 KiB
Python
294 lines
12 KiB
Python
import datetime
|
|
import json
|
|
import lzma
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from chirp import chirp_common
|
|
from chirp.sources import base
|
|
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']
|
|
mock_get.return_value.headers = {'ETag': '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 never wrote the bad data to file
|
|
self.assertEqual(0, len(files))
|
|
status.send_fail.assert_called()
|
|
|
|
def test_get_data_no_results(self):
|
|
comp = lzma.LZMACompressor(format=lzma.FORMAT_XZ)
|
|
fake_data = comp.compress(json.dumps({'count': 0}).encode())
|
|
fake_data += comp.flush()
|
|
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 = [fake_data]
|
|
mock_get.return_value.headers = {'ETag': '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 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))
|
|
comp = lzma.LZMACompressor(format=lzma.FORMAT_XZ)
|
|
fake_data = comp.compress(json.dumps({'count': 1}).encode())
|
|
fake_data += comp.flush()
|
|
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 = [fake_data]
|
|
mock_get.return_value.headers = {'ETag': 'foo'}
|
|
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()
|
|
|
|
def test_get_data_honors_etag(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(json.dumps({'ETag': 'foo'}))
|
|
rb = repeaterbook.RepeaterBook()
|
|
with mock.patch('requests.get') as mock_get:
|
|
mock_get.return_value.status_code = 304
|
|
real_timedelta = datetime.timedelta
|
|
real_datetime = datetime.datetime
|
|
|
|
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, but the server
|
|
# said it was unchanged, so we should use the cache
|
|
mock_get.assert_called()
|
|
self.assertEqual(cache_file, r)
|
|
expected_headers = dict(base.HEADERS)
|
|
expected_headers['If-None-Match'] = 'foo'
|
|
mock_get.assert_called_once_with(
|
|
mock.ANY,
|
|
headers=expected_headers,
|
|
stream=True)
|