mirror of https://github.com/craigerl/aprsd.git
Added unit tests for USWeatherPlugin, USMetarPlugin
This commit is contained in:
parent
5d6ac5cf31
commit
8ea00e9888
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue