mirror of
https://github.com/kk7ds/chirp
synced 2026-08-13 17:51:36 -04:00
2cd274dae9
This moves CHIRP to use a proxy for repeaterbook at chirpmyradio.com to reduce load on the repeaterbook API and improve performance for our users. Also, don't run the network tests in CI as they're less important now and also their failing probably has nothing to do with the patch being submitted.
261 lines
10 KiB
Python
261 lines
10 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 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):
|
|
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]
|
|
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]
|
|
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()
|