hemna/aprsd
1
0
Fork 0
mirror of https://github.com/craigerl/aprsd.git synced 2026-06-13 19:38:46 -04:00
Code Issues Projects Releases Wiki Activity

Added threads functions to APRSDPluginBase

Browse Source 
This patch updates the APRSDPluginBase class to include
standard methods for allowing plugins to create, start, stop
threads that the plugin might need/use.  Also update the aprsd-dev
to correctly start the threads and stop them for testing plugin
functionality.
Also added more unit tests and fake objects for unit tests.
This commit is contained in:
Hemna
2021-08-20 15:21:47 -04:00
parent 5f4cf89733
commit 86777d838c
5 changed files with 242 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_plugin.py
+119 -33
View File
@@ -4,7 +4,7 @@ from unittest import mock
import pytz
import aprsd
from aprsd import messaging, stats, utils
from aprsd import messaging, packets, stats, utils
from aprsd.fuzzyclock import fuzzy
from aprsd.plugins import fortune as fortune_plugin
from aprsd.plugins import ping as ping_plugin
@@ -12,36 +12,122 @@ 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 . import fake
class TestPlugin(unittest.TestCase):
def setUp(self):
self.fromcall = "KFART"
self.fromcall = fake.FAKE_FROM_CALLSIGN
self.ack = 1
self.config = utils.DEFAULT_CONFIG_DICT
self.config["ham"]["callsign"] = self.fromcall
self.config["aprs"]["login"] = fake.FAKE_TO_CALLSIGN
# Inintialize the stats object with the config
stats.APRSDStats(self.config)
def fake_packet(self, fromcall="KFART", message=None, msg_number=None):
packet = {
"from": fromcall,
"addresse": self.config["aprs"]["login"],
"format": "message",
}
if message:
packet["message_text"] = message
@mock.patch.object(fake.FakeBaseNoThreadsPlugin, "process")
def test_base_plugin_no_threads(self, mock_process):
p = fake.FakeBaseNoThreadsPlugin(self.config)
if msg_number:
packet["msgNo"] = msg_number
expected = []
actual = p.create_threads()
self.assertEqual(expected, actual)
return packet
expected = "1.0"
actual = p.version
self.assertEqual(expected, actual)
expected = 0
actual = p.message_counter
self.assertEqual(expected, actual)
expected = None
actual = p.filter(fake.fake_packet())
self.assertEqual(expected, actual)
mock_process.assert_not_called()
@mock.patch.object(fake.FakeBaseThreadsPlugin, "create_threads")
def test_base_plugin_threads_created(self, mock_create):
fake.FakeBaseThreadsPlugin(self.config)
mock_create.assert_called_once()
def test_base_plugin_threads(self):
p = fake.FakeBaseThreadsPlugin(self.config)
actual = p.create_threads()
self.assertTrue(isinstance(actual, fake.FakeThread))
p.stop_threads()
@mock.patch.object(fake.FakeRegexCommandPlugin, "process")
def test_regex_base_not_called(self, mock_process):
p = fake.FakeRegexCommandPlugin(self.config)
packet = fake.fake_packet(message="a")
expected = None
actual = p.filter(packet)
self.assertEqual(expected, actual)
mock_process.assert_not_called()
packet = fake.fake_packet(tocall="notMe", message="f")
expected = None
actual = p.filter(packet)
self.assertEqual(expected, actual)
mock_process.assert_not_called()
packet = fake.fake_packet(
message="F",
message_format=packets.PACKET_TYPE_MICE,
)
expected = None
actual = p.filter(packet)
self.assertEqual(expected, actual)
mock_process.assert_not_called()
packet = fake.fake_packet(
message="f",
message_format=packets.PACKET_TYPE_ACK,
)
expected = None
actual = p.filter(packet)
self.assertEqual(expected, actual)
mock_process.assert_not_called()
@mock.patch.object(fake.FakeRegexCommandPlugin, "process")
def test_regex_base_assert_called(self, mock_process):
p = fake.FakeRegexCommandPlugin(self.config)
packet = fake.fake_packet(message="f")
p.filter(packet)
mock_process.assert_called_once()
def test_regex_base_process_called(self):
p = fake.FakeRegexCommandPlugin(self.config)
packet = fake.fake_packet(message="f")
expected = fake.FAKE_MESSAGE_TEXT
actual = p.filter(packet)
self.assertEqual(expected, actual)
packet = fake.fake_packet(message="F")
expected = fake.FAKE_MESSAGE_TEXT
actual = p.filter(packet)
self.assertEqual(expected, actual)
packet = fake.fake_packet(message="fake")
expected = fake.FAKE_MESSAGE_TEXT
actual = p.filter(packet)
self.assertEqual(expected, actual)
packet = fake.fake_packet(message="FAKE")
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):
fortune = fortune_plugin.FortunePlugin(self.config)
mock_which.return_value = None
expected = "Fortune command not installed"
packet = self.fake_packet(message="fortune")
packet = fake.fake_packet(message="fortune")
actual = fortune.filter(packet)
self.assertEqual(expected, actual)
@@ -54,13 +140,15 @@ class TestPlugin(unittest.TestCase):
mock_output.return_value = "Funny fortune"
expected = "Funny fortune"
packet = self.fake_packet(message="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 = self.fake_packet(message="!delete")
packet = fake.fake_packet(message="!delete")
query = query_plugin.QueryPlugin(self.config)
expected = "Deleted ALL pending msgs."
@@ -72,7 +160,7 @@ class TestPlugin(unittest.TestCase):
def test_query_restart_delayed(self, mock_restart):
track = messaging.MsgTrack()
track.track = {}
packet = self.fake_packet(message="!4")
packet = fake.fake_packet(message="!4")
query = query_plugin.QueryPlugin(self.config)
expected = "No pending msgs to resend"
@@ -87,6 +175,8 @@ class TestPlugin(unittest.TestCase):
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):
@@ -104,8 +194,7 @@ class TestPlugin(unittest.TestCase):
fake_time.tm_sec = 13
time = time_plugin.TimePlugin(self.config)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="location",
msg_number=1,
)
@@ -115,8 +204,7 @@ class TestPlugin(unittest.TestCase):
cur_time = fuzzy(h, m, 1)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="time",
msg_number=1,
)
@@ -128,6 +216,8 @@ class TestPlugin(unittest.TestCase):
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()
@@ -138,8 +228,7 @@ class TestPlugin(unittest.TestCase):
ping = ping_plugin.PingPlugin(self.config)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="location",
msg_number=1,
)
@@ -157,8 +246,7 @@ class TestPlugin(unittest.TestCase):
+ str(s).zfill(2)
)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="Ping",
msg_number=1,
)
@@ -166,21 +254,21 @@ class TestPlugin(unittest.TestCase):
expected = ping_str(h, m, s)
self.assertEqual(expected, actual)
packet = self.fake_packet(
fromcall="KFART",
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:0:0:0"
version = version_plugin.VersionPlugin(self.config)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="No",
msg_number=1,
)
@@ -188,16 +276,14 @@ class TestPlugin(unittest.TestCase):
actual = version.filter(packet)
self.assertEqual(None, actual)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="version",
msg_number=1,
)
actual = version.filter(packet)
self.assertEqual(expected, actual)
packet = self.fake_packet(
fromcall="KFART",
packet = fake.fake_packet(
message="Version",
msg_number=1,
)