From 0ad791bdd917495d8acd43f1895bbd16535af5c6 Mon Sep 17 00:00:00 2001 From: Hemna Date: Tue, 7 Dec 2021 11:25:14 -0500 Subject: [PATCH] Added NotifyPlugin unit tests and more This patch restructures the unit tests for plugins. This also adds unit tests for the NotifyPlugin --- aprsd/plugins/notify.py | 3 + tests/fake.py | 10 +- tests/plugins/__init__.py | 0 tests/plugins/test_fortune.py | 28 +++++ tests/plugins/test_notify.py | 185 +++++++++++++++++++++++++++++ tests/plugins/test_ping.py | 50 ++++++++ tests/plugins/test_query.py | 37 ++++++ tests/plugins/test_time.py | 50 ++++++++ tests/plugins/test_version.py | 35 ++++++ tests/test_plugin.py | 212 +++++----------------------------- 10 files changed, 423 insertions(+), 187 deletions(-) create mode 100644 tests/plugins/__init__.py create mode 100644 tests/plugins/test_fortune.py create mode 100644 tests/plugins/test_notify.py create mode 100644 tests/plugins/test_ping.py create mode 100644 tests/plugins/test_query.py create mode 100644 tests/plugins/test_time.py create mode 100644 tests/plugins/test_version.py 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/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_notify.py b/tests/plugins/test_notify.py new file mode 100644 index 0000000..76ea58c --- /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.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/test_plugin.py b/tests/test_plugin.py index 2575e6b..ad9be56 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.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)