From 8ea00e98886db552a83a469441ee4ec8ef644aa4 Mon Sep 17 00:00:00 2001 From: Hemna Date: Tue, 7 Dec 2021 13:22:43 -0500 Subject: [PATCH] Added unit tests for USWeatherPlugin, USMetarPlugin --- aprsd/plugins/weather.py | 28 +++--- tests/plugins/test_weather.py | 176 ++++++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 12 deletions(-) create mode 100644 tests/plugins/test_weather.py 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/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)