diff --git a/aprsd/messaging.py b/aprsd/messaging.py index 23a6c8f..0834d31 100644 --- a/aprsd/messaging.py +++ b/aprsd/messaging.py @@ -542,37 +542,27 @@ def log_message( log_list = [""] if retry_number: - # LOG.info(" {} _______________(TX:{})".format(header, retry_number)) - log_list.append(f" {header} _______________(TX:{retry_number})") + log_list.append(f"{header} _______________(TX:{retry_number})") else: - # LOG.info(" {} _______________".format(header)) - log_list.append(f" {header} _______________") + log_list.append(f"{header} _______________") - # LOG.info(" Raw : {}".format(raw)) - log_list.append(f" Raw : {raw}") + log_list.append(f" Raw : {raw}") if packet_type: - # LOG.info(" Packet : {}".format(packet_type)) - log_list.append(f" Packet : {packet_type}") + log_list.append(f" Packet : {packet_type}") if tocall: - # LOG.info(" To : {}".format(tocall)) - log_list.append(f" To : {tocall}") + log_list.append(f" To : {tocall}") if fromcall: - # LOG.info(" From : {}".format(fromcall)) - log_list.append(f" From : {fromcall}") + log_list.append(f" From : {fromcall}") if ack: - # LOG.info(" Ack : {}".format(ack)) - log_list.append(f" Ack : {ack}") + log_list.append(f" Ack : {ack}") else: - # LOG.info(" Message : {}".format(message)) - log_list.append(f" Message : {message}") + log_list.append(f" Message : {message}") if msg_num: - # LOG.info(" Msg number : {}".format(msg_num)) - log_list.append(f" Msg number : {msg_num}") + log_list.append(f" Msg number : {msg_num}") if uuid: - log_list.append(f" UUID : {uuid}") - # LOG.info(" {} _______________ Complete".format(header)) - log_list.append(f" {header} _______________ Complete") + log_list.append(f" UUID : {uuid}") + log_list.append(f"{header} _______________ Complete") LOG.info("\n".join(log_list)) diff --git a/aprsd/plugins/notify.py b/aprsd/plugins/notify.py index 187bdd3..bb01084 100644 --- a/aprsd/plugins/notify.py +++ b/aprsd/plugins/notify.py @@ -45,6 +45,8 @@ class NotifySeenPlugin(plugin.APRSDWatchListPluginBase): allow_delay=False, ) return msg + else: + return messaging.NULL_MESSAGE else: LOG.debug( "Not old enough to notify callsign '{}' : {} < {}".format( @@ -53,3 +55,4 @@ class NotifySeenPlugin(plugin.APRSDWatchListPluginBase): wl.max_delta(), ), ) + return messaging.NULL_MESSAGE diff --git a/aprsd/plugins/weather.py b/aprsd/plugins/weather.py index c656073..1880fef 100644 --- a/aprsd/plugins/weather.py +++ b/aprsd/plugins/weather.py @@ -10,7 +10,7 @@ from aprsd import plugin, plugin_utils, trace LOG = logging.getLogger("APRSD") -class USWeatherPlugin(plugin.APRSDRegexCommandPluginBase): +class USWeatherPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin): """USWeather Command Returns a weather report for the calling weather station @@ -27,26 +27,27 @@ class USWeatherPlugin(plugin.APRSDRegexCommandPluginBase): command_name = "USWeather" short_description = "Provide USA only weather of GPS Beacon location" + def setup(self): + self.ensure_aprs_fi_key() + @trace.trace def process(self, packet): LOG.info("Weather Plugin") fromcall = packet.get("from") # message = packet.get("message_text", None) # ack = packet.get("msgNo", "0") - try: - self.config.exists(["services", "aprs.fi", "apiKey"]) - except Exception as ex: - LOG.error(f"Failed to find config aprs.fi:apikey {ex}") - return "No aprs.fi apikey found" - api_key = self.config["services"]["aprs.fi"]["apiKey"] try: aprs_data = plugin_utils.get_aprs_fi(api_key, fromcall) except Exception as ex: LOG.error(f"Failed to fetch aprs.fi data {ex}") - return "Failed to fetch location" + return "Failed to fetch aprs.fi location" + + LOG.debug(f"LocationPlugin: aprs_data = {aprs_data}") + if not len(aprs_data["entries"]): + LOG.error("Didn't get any entries from aprs.fi") + return "Failed to fetch aprs.fi location" - # LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data)) lat = aprs_data["entries"][0]["lat"] lon = aprs_data["entries"][0]["lng"] @@ -71,7 +72,7 @@ class USWeatherPlugin(plugin.APRSDRegexCommandPluginBase): return reply -class USMetarPlugin(plugin.APRSDRegexCommandPluginBase): +class USMetarPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin): """METAR Command This provides a METAR weather report from a station near the caller @@ -90,6 +91,9 @@ class USMetarPlugin(plugin.APRSDRegexCommandPluginBase): command_name = "USMetar" short_description = "USA only METAR of GPS Beacon location" + def setup(self): + self.ensure_aprs_fi_key() + @trace.trace def process(self, packet): fromcall = packet.get("from") @@ -126,12 +130,12 @@ class USMetarPlugin(plugin.APRSDRegexCommandPluginBase): aprs_data = plugin_utils.get_aprs_fi(api_key, fromcall) except Exception as ex: LOG.error(f"Failed to fetch aprs.fi data {ex}") - return "Failed to fetch location" + return "Failed to fetch aprs.fi location" # LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data)) if not len(aprs_data["entries"]): LOG.error("Found no entries from aprs.fi!") - return "Failed to fetch location" + return "Failed to fetch aprs.fi location" lat = aprs_data["entries"][0]["lat"] lon = aprs_data["entries"][0]["lng"] diff --git a/aprsd/threads.py b/aprsd/threads.py index 3d5350b..e8422f3 100644 --- a/aprsd/threads.py +++ b/aprsd/threads.py @@ -88,7 +88,7 @@ class KeepAliveThread(APRSDThread): tracemalloc.start() super().__init__("KeepAlive") self.config = config - max_timeout = {"hours": 0.0, "minutes": 5, "seconds": 0} + max_timeout = {"hours": 0.0, "minutes": 2, "seconds": 0} self.max_delta = datetime.timedelta(**max_timeout) def loop(self): diff --git a/tests/fake.py b/tests/fake.py index 71a2ec1..42a34dd 100644 --- a/tests/fake.py +++ b/tests/fake.py @@ -2,7 +2,7 @@ from aprsd import packets, plugin, threads FAKE_MESSAGE_TEXT = "fake MeSSage" -FAKE_FROM_CALLSIGN = "KFART" +FAKE_FROM_CALLSIGN = "KFAKE" FAKE_TO_CALLSIGN = "KMINE" @@ -45,7 +45,7 @@ class FakeThread(threads.APRSDThread): super().__init__("FakeThread") def loop(self): - return True + return False class FakeBaseThreadsPlugin(plugin.APRSDPluginBase): @@ -71,3 +71,9 @@ class FakeRegexCommandPlugin(plugin.APRSDRegexCommandPluginBase): def process(self, packet): return FAKE_MESSAGE_TEXT + + +class FakeWatchListPlugin(plugin.APRSDWatchListPluginBase): + + def process(self, packet): + return FAKE_MESSAGE_TEXT diff --git a/tests/plugins/__init__.py b/tests/plugins/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugins/test_fortune.py b/tests/plugins/test_fortune.py new file mode 100644 index 0000000..d0622fd --- /dev/null +++ b/tests/plugins/test_fortune.py @@ -0,0 +1,28 @@ +from unittest import mock + +from aprsd.plugins import fortune as fortune_plugin + +from .. import fake, test_plugin + + +class TestFortunePlugin(test_plugin.TestPlugin): + @mock.patch("shutil.which") + def test_fortune_fail(self, mock_which): + mock_which.return_value = None + fortune = fortune_plugin.FortunePlugin(self.config) + expected = "FortunePlugin isn't enabled" + packet = fake.fake_packet(message="fortune") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("subprocess.check_output") + @mock.patch("shutil.which") + def test_fortune_success(self, mock_which, mock_output): + mock_which.return_value = "/usr/bin/games/fortune" + mock_output.return_value = "Funny fortune" + fortune = fortune_plugin.FortunePlugin(self.config) + + expected = "Funny fortune" + packet = fake.fake_packet(message="fortune") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/plugins/test_location.py b/tests/plugins/test_location.py new file mode 100644 index 0000000..ae5893b --- /dev/null +++ b/tests/plugins/test_location.py @@ -0,0 +1,89 @@ +from unittest import mock + +from aprsd.plugins import location as location_plugin + +from .. import fake, test_plugin + + +class TestLocationPlugin(test_plugin.TestPlugin): + + @mock.patch("aprsd.config.Config.check_option") + def test_location_not_enabled_missing_aprs_fi_key(self, mock_check): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check.side_effect = Exception + fortune = location_plugin.LocationPlugin(self.config) + expected = "LocationPlugin isn't enabled" + packet = fake.fake_packet(message="location") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_location_failed_aprs_fi_location(self, mock_check): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check.side_effect = Exception + fortune = location_plugin.LocationPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="location") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_location_failed_aprs_fi_location_no_entries(self, mock_check): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check.return_value = {"entries": []} + fortune = location_plugin.LocationPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="location") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + @mock.patch("aprsd.plugin_utils.get_weather_gov_for_gps") + @mock.patch("time.time") + def test_location_unknown_gps(self, mock_time, mock_weather, mock_check_aprs): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check_aprs.return_value = { + "entries": [ + { + "lat": 10, + "lng": 11, + "lasttime": 10, + }, + ], + } + mock_weather.side_effect = Exception + mock_time.return_value = 10 + fortune = location_plugin.LocationPlugin(self.config) + expected = "KFAKE: Unknown Location 0' 10,11 0.0h ago" + packet = fake.fake_packet(message="location") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + @mock.patch("aprsd.plugin_utils.get_weather_gov_for_gps") + @mock.patch("time.time") + def test_location_works(self, mock_time, mock_weather, mock_check_aprs): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check_aprs.return_value = { + "entries": [ + { + "lat": 10, + "lng": 11, + "lasttime": 10, + }, + ], + } + expected_town = "Appomattox, VA" + wx_data = {"location": {"areaDescription": expected_town}} + mock_weather.return_value = wx_data + mock_time.return_value = 10 + fortune = location_plugin.LocationPlugin(self.config) + expected = f"KFAKE: {expected_town} 0' 10,11 0.0h ago" + packet = fake.fake_packet(message="location") + actual = fortune.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/plugins/test_notify.py b/tests/plugins/test_notify.py new file mode 100644 index 0000000..8f0254f --- /dev/null +++ b/tests/plugins/test_notify.py @@ -0,0 +1,185 @@ +from unittest import mock + +from aprsd import client +from aprsd import config as aprsd_config +from aprsd import messaging, packets +from aprsd.plugins import notify as notify_plugin + +from .. import fake, test_plugin + + +DEFAULT_WATCHLIST_CALLSIGNS = [fake.FAKE_FROM_CALLSIGN] + + +class TestWatchListPlugin(test_plugin.TestPlugin): + def setUp(self): + self.fromcall = fake.FAKE_FROM_CALLSIGN + self.ack = 1 + + def _config( + self, + watchlist_enabled=True, + watchlist_alert_callsign=None, + watchlist_alert_time_seconds=None, + watchlist_packet_keep_count=None, + watchlist_callsigns=DEFAULT_WATCHLIST_CALLSIGNS, + ): + _config = aprsd_config.Config(aprsd_config.DEFAULT_CONFIG_DICT) + default_wl = aprsd_config.DEFAULT_CONFIG_DICT["aprsd"]["watch_list"] + + _config["ham"]["callsign"] = self.fromcall + _config["aprs"]["login"] = fake.FAKE_TO_CALLSIGN + _config["services"]["aprs.fi"]["apiKey"] = "something" + + # Set the watchlist specific config options + + _config["aprsd"]["watch_list"]["enabled"] = watchlist_enabled + if not watchlist_alert_callsign: + watchlist_alert_callsign = fake.FAKE_TO_CALLSIGN + _config["aprsd"]["watch_list"]["alert_callsign"] = watchlist_alert_callsign + + if not watchlist_alert_time_seconds: + watchlist_alert_time_seconds = default_wl["alert_time_seconds"] + _config["aprsd"]["watch_list"]["alert_time_seconds"] = watchlist_alert_time_seconds + + if not watchlist_packet_keep_count: + watchlist_packet_keep_count = default_wl["packet_keep_count"] + _config["aprsd"]["watch_list"]["packet_keep_count"] = watchlist_packet_keep_count + + _config["aprsd"]["watch_list"]["callsigns"] = watchlist_callsigns + return _config + + +class TestAPRSDWatchListPluginBase(TestWatchListPlugin): + + def test_watchlist_not_enabled(self): + config = self._config(watchlist_enabled=False) + self.config_and_init(config=config) + plugin = fake.FakeWatchListPlugin(self.config) + + packet = fake.fake_packet( + message="version", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + @mock.patch("aprsd.client.ClientFactory", autospec=True) + def test_watchlist_not_in_watchlist(self, mock_factory): + client.factory = mock_factory + config = self._config() + self.config_and_init(config=config) + plugin = fake.FakeWatchListPlugin(self.config) + + packet = fake.fake_packet( + fromcall="FAKE", + message="version", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + +class TestNotifySeenPlugin(TestWatchListPlugin): + + def test_disabled(self): + config = self._config(watchlist_enabled=False) + self.config_and_init(config=config) + plugin = notify_plugin.NotifySeenPlugin(self.config) + + packet = fake.fake_packet( + message="version", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + @mock.patch("aprsd.client.ClientFactory", autospec=True) + def test_callsign_not_in_watchlist(self, mock_factory): + client.factory = mock_factory + config = self._config(watchlist_enabled=False) + self.config_and_init(config=config) + plugin = notify_plugin.NotifySeenPlugin(self.config) + + packet = fake.fake_packet( + message="version", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + @mock.patch("aprsd.client.ClientFactory", autospec=True) + @mock.patch("aprsd.packets.WatchList.is_old") + def test_callsign_in_watchlist_not_old(self, mock_is_old, mock_factory): + client.factory = mock_factory + mock_is_old.return_value = False + config = self._config( + watchlist_enabled=True, + watchlist_callsigns=["WB4BOR"], + ) + self.config_and_init(config=config) + plugin = notify_plugin.NotifySeenPlugin(self.config) + + packet = fake.fake_packet( + fromcall="WB4BOR", + message="ping", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + @mock.patch("aprsd.client.ClientFactory", autospec=True) + @mock.patch("aprsd.packets.WatchList.is_old") + def test_callsign_in_watchlist_old_same_alert_callsign(self, mock_is_old, mock_factory): + client.factory = mock_factory + mock_is_old.return_value = True + config = self._config( + watchlist_enabled=True, + watchlist_alert_callsign="WB4BOR", + watchlist_callsigns=["WB4BOR"], + ) + self.config_and_init(config=config) + plugin = notify_plugin.NotifySeenPlugin(self.config) + + packet = fake.fake_packet( + fromcall="WB4BOR", + message="ping", + msg_number=1, + ) + actual = plugin.filter(packet) + expected = messaging.NULL_MESSAGE + self.assertEqual(expected, actual) + + @mock.patch("aprsd.client.ClientFactory", autospec=True) + @mock.patch("aprsd.packets.WatchList.is_old") + def test_callsign_in_watchlist_old_send_alert(self, mock_is_old, mock_factory): + client.factory = mock_factory + mock_is_old.return_value = True + notify_callsign = "KFAKE" + fromcall = "WB4BOR" + config = self._config( + watchlist_enabled=True, + watchlist_alert_callsign=notify_callsign, + watchlist_callsigns=["WB4BOR"], + ) + self.config_and_init(config=config) + plugin = notify_plugin.NotifySeenPlugin(self.config) + + packet = fake.fake_packet( + fromcall=fromcall, + message="ping", + msg_number=1, + ) + packet_type = packets.get_packet_type(packet) + actual = plugin.filter(packet) + msg = f"{fromcall} was just seen by type:'{packet_type}'" + + self.assertIsInstance(actual, messaging.TextMessage) + self.assertEqual(fake.FAKE_TO_CALLSIGN, actual.fromcall) + self.assertEqual(notify_callsign, actual.tocall) + self.assertEqual(msg, actual.message) diff --git a/tests/plugins/test_ping.py b/tests/plugins/test_ping.py new file mode 100644 index 0000000..22b69eb --- /dev/null +++ b/tests/plugins/test_ping.py @@ -0,0 +1,50 @@ +from unittest import mock + +from aprsd.plugins import ping as ping_plugin + +from .. import fake, test_plugin + + +class TestPingPlugin(test_plugin.TestPlugin): + @mock.patch("time.localtime") + def test_ping(self, mock_time): + fake_time = mock.MagicMock() + h = fake_time.tm_hour = 16 + m = fake_time.tm_min = 12 + s = fake_time.tm_sec = 55 + mock_time.return_value = fake_time + + ping = ping_plugin.PingPlugin(self.config) + + packet = fake.fake_packet( + message="location", + msg_number=1, + ) + + result = ping.filter(packet) + self.assertEqual(None, result) + + def ping_str(h, m, s): + return ( + "Pong! " + + str(h).zfill(2) + + ":" + + str(m).zfill(2) + + ":" + + str(s).zfill(2) + ) + + packet = fake.fake_packet( + message="Ping", + msg_number=1, + ) + actual = ping.filter(packet) + expected = ping_str(h, m, s) + self.assertEqual(expected, actual) + + packet = fake.fake_packet( + message="ping", + msg_number=1, + ) + actual = ping.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/plugins/test_query.py b/tests/plugins/test_query.py new file mode 100644 index 0000000..0a8d5db --- /dev/null +++ b/tests/plugins/test_query.py @@ -0,0 +1,37 @@ +from unittest import mock + +from aprsd import messaging +from aprsd.plugins import query as query_plugin + +from .. import fake, test_plugin + + +class TestQueryPlugin(test_plugin.TestPlugin): + @mock.patch("aprsd.messaging.MsgTrack.flush") + def test_query_flush(self, mock_flush): + packet = fake.fake_packet(message="!delete") + query = query_plugin.QueryPlugin(self.config) + + expected = "Deleted ALL pending msgs." + actual = query.filter(packet) + mock_flush.assert_called_once() + self.assertEqual(expected, actual) + + @mock.patch("aprsd.messaging.MsgTrack.restart_delayed") + def test_query_restart_delayed(self, mock_restart): + track = messaging.MsgTrack() + track.data = {} + packet = fake.fake_packet(message="!4") + query = query_plugin.QueryPlugin(self.config) + + expected = "No pending msgs to resend" + actual = query.filter(packet) + mock_restart.assert_not_called() + self.assertEqual(expected, actual) + mock_restart.reset_mock() + + # add a message + msg = messaging.TextMessage(self.fromcall, "testing", self.ack) + track.add(msg) + actual = query.filter(packet) + mock_restart.assert_called_once() diff --git a/tests/plugins/test_time.py b/tests/plugins/test_time.py new file mode 100644 index 0000000..52616a4 --- /dev/null +++ b/tests/plugins/test_time.py @@ -0,0 +1,50 @@ +from unittest import mock + +import pytz + +from aprsd.fuzzyclock import fuzzy +from aprsd.plugins import time as time_plugin + +from .. import fake, test_plugin + + +class TestTimePlugins(test_plugin.TestPlugin): + + @mock.patch("aprsd.plugins.time.TimePlugin._get_local_tz") + @mock.patch("aprsd.plugins.time.TimePlugin._get_utcnow") + def test_time(self, mock_utcnow, mock_localtz): + utcnow = pytz.datetime.datetime.utcnow() + mock_utcnow.return_value = utcnow + tz = pytz.timezone("US/Pacific") + mock_localtz.return_value = tz + + gmt_t = pytz.utc.localize(utcnow) + local_t = gmt_t.astimezone(tz) + + fake_time = mock.MagicMock() + h = int(local_t.strftime("%H")) + m = int(local_t.strftime("%M")) + fake_time.tm_sec = 13 + time = time_plugin.TimePlugin(self.config) + + packet = fake.fake_packet( + message="location", + msg_number=1, + ) + + actual = time.filter(packet) + self.assertEqual(None, actual) + + cur_time = fuzzy(h, m, 1) + + packet = fake.fake_packet( + message="time", + msg_number=1, + ) + local_short_str = local_t.strftime("%H:%M %Z") + expected = "{} ({})".format( + cur_time, + local_short_str, + ) + actual = time.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/plugins/test_version.py b/tests/plugins/test_version.py new file mode 100644 index 0000000..c6e9055 --- /dev/null +++ b/tests/plugins/test_version.py @@ -0,0 +1,35 @@ +from unittest import mock + +import aprsd +from aprsd.plugins import version as version_plugin + +from .. import fake, test_plugin + + +class TestVersionPlugin(test_plugin.TestPlugin): + @mock.patch("aprsd.plugin.PluginManager.get_plugins") + def test_version(self, mock_get_plugins): + expected = f"APRSD ver:{aprsd.__version__} uptime:00:00:00" + version = version_plugin.VersionPlugin(self.config) + + packet = fake.fake_packet( + message="No", + msg_number=1, + ) + + actual = version.filter(packet) + self.assertEqual(None, actual) + + packet = fake.fake_packet( + message="version", + msg_number=1, + ) + actual = version.filter(packet) + self.assertEqual(expected, actual) + + packet = fake.fake_packet( + message="Version", + msg_number=1, + ) + actual = version.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/plugins/test_weather.py b/tests/plugins/test_weather.py new file mode 100644 index 0000000..d607e0b --- /dev/null +++ b/tests/plugins/test_weather.py @@ -0,0 +1,176 @@ +from unittest import mock + +from aprsd.plugins import weather as weather_plugin + +from .. import fake, test_plugin + + +class TestUSWeatherPluginPlugin(test_plugin.TestPlugin): + + @mock.patch("aprsd.config.Config.check_option") + def test_not_enabled_missing_aprs_fi_key(self, mock_check): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check.side_effect = Exception + wx = weather_plugin.USWeatherPlugin(self.config) + expected = "USWeatherPlugin isn't enabled" + packet = fake.fake_packet(message="weather") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_failed_aprs_fi_location(self, mock_check): + # When the aprs.fi api key isn't set, then + # the Plugin will be disabled. + mock_check.side_effect = Exception + wx = weather_plugin.USWeatherPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="weather") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_failed_aprs_fi_location_no_entries(self, mock_check): + # When the aprs.fi api key isn't set, then + # the Plugin will be disabled. + mock_check.return_value = {"entries": []} + wx = weather_plugin.USWeatherPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="weather") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + @mock.patch("aprsd.plugin_utils.get_weather_gov_for_gps") + def test_unknown_gps(self, mock_weather, mock_check_aprs): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check_aprs.return_value = { + "entries": [ + { + "lat": 10, + "lng": 11, + "lasttime": 10, + }, + ], + } + mock_weather.side_effect = Exception + wx = weather_plugin.USWeatherPlugin(self.config) + expected = "Unable to get weather" + packet = fake.fake_packet(message="weather") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + @mock.patch("aprsd.plugin_utils.get_weather_gov_for_gps") + def test_working(self, mock_weather, mock_check_aprs): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check_aprs.return_value = { + "entries": [ + { + "lat": 10, + "lng": 11, + "lasttime": 10, + }, + ], + } + mock_weather.return_value = { + "currentobservation": {"Temp": "400"}, + "data": { + "temperature": ["10", "11"], + "weather": ["test", "another"], + }, + "time": {"startPeriodName": ["ignored", "sometime"]}, + } + wx = weather_plugin.USWeatherPlugin(self.config) + expected = "400F(10F/11F) test. sometime, another." + packet = fake.fake_packet(message="weather") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + +class TestUSMetarPlugin(test_plugin.TestPlugin): + + @mock.patch("aprsd.config.Config.check_option") + def test_not_enabled_missing_aprs_fi_key(self, mock_check): + # When the aprs.fi api key isn't set, then + # the LocationPlugin will be disabled. + mock_check.side_effect = Exception + wx = weather_plugin.USMetarPlugin(self.config) + expected = "USMetarPlugin isn't enabled" + packet = fake.fake_packet(message="metar") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_failed_aprs_fi_location(self, mock_check): + # When the aprs.fi api key isn't set, then + # the Plugin will be disabled. + mock_check.side_effect = Exception + wx = weather_plugin.USMetarPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="metar") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + def test_failed_aprs_fi_location_no_entries(self, mock_check): + # When the aprs.fi api key isn't set, then + # the Plugin will be disabled. + mock_check.return_value = {"entries": []} + wx = weather_plugin.USMetarPlugin(self.config) + expected = "Failed to fetch aprs.fi location" + packet = fake.fake_packet(message="metar") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_weather_gov_metar") + def test_gov_metar_fetch_fails(self, mock_metar): + mock_metar.side_effect = Exception + wx = weather_plugin.USMetarPlugin(self.config) + expected = "Unable to find station METAR" + packet = fake.fake_packet(message="metar KPAO") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_weather_gov_metar") + def test_airport_works(self, mock_metar): + + class Response: + text = '{"properties": {"rawMessage": "BOGUSMETAR"}}' + mock_metar.return_value = Response() + + wx = weather_plugin.USMetarPlugin(self.config) + expected = "BOGUSMETAR" + packet = fake.fake_packet(message="metar KPAO") + actual = wx.filter(packet) + self.assertEqual(expected, actual) + + @mock.patch("aprsd.plugin_utils.get_weather_gov_metar") + @mock.patch("aprsd.plugin_utils.get_aprs_fi") + @mock.patch("aprsd.plugin_utils.get_weather_gov_for_gps") + def test_metar_works(self, mock_wx_for_gps, mock_check_aprs, mock_metar): + mock_wx_for_gps.return_value = { + "location": {"metar": "BOGUSMETAR"}, + } + + class Response: + text = '{"properties": {"rawMessage": "BOGUSMETAR"}}' + + mock_check_aprs.return_value = { + "entries": [ + { + "lat": 10, + "lng": 11, + "lasttime": 10, + }, + ], + } + mock_metar.return_value = Response() + + wx = weather_plugin.USMetarPlugin(self.config) + expected = "BOGUSMETAR" + packet = fake.fake_packet(message="metar") + actual = wx.filter(packet) + self.assertEqual(expected, actual) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 2575e6b..9e59eee 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,34 +1,44 @@ import unittest from unittest import mock -import pytz - -import aprsd -from aprsd import config, messaging, packets, stats -from aprsd.fuzzyclock import fuzzy -from aprsd.plugins import fortune as fortune_plugin -from aprsd.plugins import ping as ping_plugin -from aprsd.plugins import query as query_plugin -from aprsd.plugins import time as time_plugin -from aprsd.plugins import version as version_plugin +from aprsd import config as aprsd_config +from aprsd import messaging, packets, stats from . import fake class TestPlugin(unittest.TestCase): - def setUp(self): + + def setUp(self) -> None: self.fromcall = fake.FAKE_FROM_CALLSIGN self.ack = 1 - self.config = config.DEFAULT_CONFIG_DICT - self.config["ham"]["callsign"] = self.fromcall - self.config["aprs"]["login"] = fake.FAKE_TO_CALLSIGN - self.config["services"]["aprs.fi"]["apiKey"] = "something" + self.config_and_init() + + def tearDown(self) -> None: + stats.APRSDStats._instance = None + packets.WatchList._instance = None + packets.SeenList._instance = None + messaging.MsgTrack._instance = None + self.config = None + + def config_and_init(self, config=None): + if not config: + self.config = aprsd_config.Config(aprsd_config.DEFAULT_CONFIG_DICT) + self.config["ham"]["callsign"] = self.fromcall + self.config["aprs"]["login"] = fake.FAKE_TO_CALLSIGN + self.config["services"]["aprs.fi"]["apiKey"] = "something" + else: + self.config = config + # Inintialize the stats object with the config stats.APRSDStats(self.config) packets.WatchList(config=self.config) packets.SeenList(config=self.config) messaging.MsgTrack(config=self.config) + +class TestPluginBase(TestPlugin): + @mock.patch.object(fake.FakeBaseNoThreadsPlugin, "process") def test_base_plugin_no_threads(self, mock_process): p = fake.FakeBaseNoThreadsPlugin(self.config) @@ -52,8 +62,9 @@ class TestPlugin(unittest.TestCase): @mock.patch.object(fake.FakeBaseThreadsPlugin, "create_threads") def test_base_plugin_threads_created(self, mock_create): - fake.FakeBaseThreadsPlugin(self.config) + p = fake.FakeBaseThreadsPlugin(self.config) mock_create.assert_called_once() + p.stop_threads() def test_base_plugin_threads(self): p = fake.FakeBaseThreadsPlugin(self.config) @@ -123,172 +134,3 @@ class TestPlugin(unittest.TestCase): expected = fake.FAKE_MESSAGE_TEXT actual = p.filter(packet) self.assertEqual(expected, actual) - - -class TestFortunePlugin(TestPlugin): - @mock.patch("shutil.which") - def test_fortune_fail(self, mock_which): - mock_which.return_value = None - fortune = fortune_plugin.FortunePlugin(self.config) - expected = "FortunePlugin isn't enabled" - packet = fake.fake_packet(message="fortune") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - @mock.patch("subprocess.check_output") - @mock.patch("shutil.which") - def test_fortune_success(self, mock_which, mock_output): - mock_which.return_value = "/usr/bin/games/fortune" - mock_output.return_value = "Funny fortune" - fortune = fortune_plugin.FortunePlugin(self.config) - - expected = "Funny fortune" - packet = fake.fake_packet(message="fortune") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - -class TestQueryPlugin(TestPlugin): - @mock.patch("aprsd.messaging.MsgTrack.flush") - def test_query_flush(self, mock_flush): - packet = fake.fake_packet(message="!delete") - query = query_plugin.QueryPlugin(self.config) - - expected = "Deleted ALL pending msgs." - actual = query.filter(packet) - mock_flush.assert_called_once() - self.assertEqual(expected, actual) - - @mock.patch("aprsd.messaging.MsgTrack.restart_delayed") - def test_query_restart_delayed(self, mock_restart): - track = messaging.MsgTrack() - track.data = {} - packet = fake.fake_packet(message="!4") - query = query_plugin.QueryPlugin(self.config) - - expected = "No pending msgs to resend" - actual = query.filter(packet) - mock_restart.assert_not_called() - self.assertEqual(expected, actual) - mock_restart.reset_mock() - - # add a message - msg = messaging.TextMessage(self.fromcall, "testing", self.ack) - track.add(msg) - actual = query.filter(packet) - mock_restart.assert_called_once() - - -class TestTimePlugins(TestPlugin): - @mock.patch("aprsd.plugins.time.TimePlugin._get_local_tz") - @mock.patch("aprsd.plugins.time.TimePlugin._get_utcnow") - def test_time(self, mock_utcnow, mock_localtz): - utcnow = pytz.datetime.datetime.utcnow() - mock_utcnow.return_value = utcnow - tz = pytz.timezone("US/Pacific") - mock_localtz.return_value = tz - - gmt_t = pytz.utc.localize(utcnow) - local_t = gmt_t.astimezone(tz) - - fake_time = mock.MagicMock() - h = int(local_t.strftime("%H")) - m = int(local_t.strftime("%M")) - fake_time.tm_sec = 13 - time = time_plugin.TimePlugin(self.config) - - packet = fake.fake_packet( - message="location", - msg_number=1, - ) - - actual = time.filter(packet) - self.assertEqual(None, actual) - - cur_time = fuzzy(h, m, 1) - - packet = fake.fake_packet( - message="time", - msg_number=1, - ) - local_short_str = local_t.strftime("%H:%M %Z") - expected = "{} ({})".format( - cur_time, - local_short_str, - ) - actual = time.filter(packet) - self.assertEqual(expected, actual) - - -class TestPingPlugin(TestPlugin): - @mock.patch("time.localtime") - def test_ping(self, mock_time): - fake_time = mock.MagicMock() - h = fake_time.tm_hour = 16 - m = fake_time.tm_min = 12 - s = fake_time.tm_sec = 55 - mock_time.return_value = fake_time - - ping = ping_plugin.PingPlugin(self.config) - - packet = fake.fake_packet( - message="location", - msg_number=1, - ) - - result = ping.filter(packet) - self.assertEqual(None, result) - - def ping_str(h, m, s): - return ( - "Pong! " - + str(h).zfill(2) - + ":" - + str(m).zfill(2) - + ":" - + str(s).zfill(2) - ) - - packet = fake.fake_packet( - message="Ping", - msg_number=1, - ) - actual = ping.filter(packet) - expected = ping_str(h, m, s) - self.assertEqual(expected, actual) - - packet = fake.fake_packet( - message="ping", - msg_number=1, - ) - actual = ping.filter(packet) - self.assertEqual(expected, actual) - - -class TestVersionPlugin(TestPlugin): - @mock.patch("aprsd.plugin.PluginManager.get_plugins") - def test_version(self, mock_get_plugins): - expected = f"APRSD ver:{aprsd.__version__} uptime:00:00:00" - version = version_plugin.VersionPlugin(self.config) - - packet = fake.fake_packet( - message="No", - msg_number=1, - ) - - actual = version.filter(packet) - self.assertEqual(None, actual) - - packet = fake.fake_packet( - message="version", - msg_number=1, - ) - actual = version.filter(packet) - self.assertEqual(expected, actual) - - packet = fake.fake_packet( - message="Version", - msg_number=1, - ) - actual = version.filter(packet) - self.assertEqual(expected, actual) diff --git a/tox.ini b/tox.ini index 95e1349..0fb906b 100644 --- a/tox.ini +++ b/tox.ini @@ -2,7 +2,7 @@ minversion = 2.9.0 skipdist = True skip_missing_interpreters = true -envlist = pre-commit,pep8,py{36,37,38} +envlist = pre-commit,pep8,py{36,37,38,39} # Activate isolated build environment. tox will use a virtual environment # to build a source distribution from the source tree. For build tools and @@ -10,8 +10,11 @@ envlist = pre-commit,pep8,py{36,37,38} isolated_build = true [testenv] -setenv = _PYTEST_SETUP_SKIP_APRSD_DEP=1 - coverage: _APRSD_TOX_CMD=coverage run -m pytest +description = Run unit-testing +setenv = + _PYTEST_SETUP_SKIP_APRSD_DEP=1 + PYTHONDONTWRITEBYTECODE=1 + PYTHONUNBUFFERED=1 usedevelop = True install_command = pip install {opts} {packages} extras = tests @@ -20,18 +23,10 @@ deps = coverage: coverage -r{toxinidir}/dev-requirements.txt pytestmain: git+https://github.com/pytest-dev/pytest.git@main commands = - {env:_APRSD_TOX_CMD:pytest} {posargs} + pytest -v --cov-report term-missing --cov=aprsd {posargs} coverage: coverage report -m coverage: coverage xml -[pytest] -minversion=2.0 -testpaths = tests -#--pyargs --doctest-modules --ignore=.tox -addopts=-r a -filterwarnings = - error - [testenv:docs] skip_install = true deps = @@ -78,8 +73,8 @@ exclude = .venv,.git,.tox,dist,doc,.ropeproject python = 3.6: py36, pep8 3.7: py38, pep8 - 3.8: py38, pep8, type-check, docs - 3.9: py39 + 3.8: py38, pep8 + 3.9: py39, pep8, type-check, docs [testenv:fmt] # This will reformat your code to comply with pep8