hemna/aprsd
1
0
Fork 0
mirror of https://github.com/craigerl/aprsd.git synced 2024-09-27 15:46:53 -04:00
Code Issues Projects Releases Wiki Activity

Merge pull request #75 from craigerl/unittests

Browse Source 
Unittests
This commit is contained in:
Walter A. Boring IV 2021-12-07 13:37:02 -05:00 committed by GitHub
commit bcc1b4e309
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 725 additions and 235 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
aprsd/messaging.py
View File

@ -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))

3
aprsd/plugins/notify.py
View File

@ -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

28
aprsd/plugins/weather.py
View File

@ -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"]

2
aprsd/threads.py
View File

@ -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):

10
tests/fake.py
View File

@ -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

0
tests/plugins/__init__.py Normal file
View File

28
tests/plugins/test_fortune.py Normal file
View File

@ -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)

89
tests/plugins/test_location.py Normal file
View File

@ -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)

185
tests/plugins/test_notify.py Normal file
View File

@ -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)

50
tests/plugins/test_ping.py Normal file
View File

@ -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)

37
tests/plugins/test_query.py Normal file
View File

@ -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()

50
tests/plugins/test_time.py Normal file
View File

@ -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)

35
tests/plugins/test_version.py Normal file
View File

@ -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)

176
tests/plugins/test_weather.py Normal file
View File

@ -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)

212
tests/test_plugin.py
View File

@ -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)

23
tox.ini
View File

@ -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