hemna
/
aprsd
mirror of https://github.com/craigerl/aprsd.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Added NotifyPlugin unit tests and more 

This patch restructures the unit tests for plugins.
This also adds unit tests for the NotifyPlugin
This commit is contained in:
Hemna 2021-12-07 11:25:14 -05:00
parent 96cc07d15f
commit 0ad791bdd9
10 changed files with 423 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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)

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

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