{{ config_json|safe }}
diff --git a/tests/fake.py b/tests/fake.py
new file mode 100644
index 0000000..a45ead3
--- /dev/null
+++ b/tests/fake.py
@@ -0,0 +1,67 @@
+from aprsd import packets, plugin, threads
+
+
+FAKE_MESSAGE_TEXT = "fake MeSSage"
+FAKE_FROM_CALLSIGN = "KFART"
+FAKE_TO_CALLSIGN = "KMINE"
+
+
+def fake_packet(
+ fromcall=FAKE_FROM_CALLSIGN,
+ tocall=FAKE_TO_CALLSIGN,
+ message=None,
+ msg_number=None,
+ message_format=packets.PACKET_TYPE_MESSAGE,
+):
+ packet = {
+ "from": fromcall,
+ "addresse": tocall,
+ "format": message_format,
+ }
+ if message:
+ packet["message_text"] = message
+
+ if msg_number:
+ packet["msgNo"] = msg_number
+
+ return packet
+
+
+class FakeBaseNoThreadsPlugin(plugin.APRSDPluginBase):
+ version = "1.0"
+
+ def filter(self, packet):
+ return None
+
+ def process(self, packet):
+ return "process"
+
+
+class FakeThread(threads.APRSDThread):
+ def __init__(self):
+ super().__init__("FakeThread")
+
+ def loop(self):
+ return True
+
+
+class FakeBaseThreadsPlugin(plugin.APRSDPluginBase):
+ version = "1.0"
+
+ def filter(self, packet):
+ return None
+
+ def process(self, packet):
+ return "process"
+
+ def create_threads(self):
+ return FakeThread()
+
+
+class FakeRegexCommandPlugin(plugin.APRSDRegexCommandPluginBase):
+ version = "1.0"
+ command_regex = "^[fF]"
+ command_name = "fake"
+
+ def process(self, packet):
+ return FAKE_MESSAGE_TEXT
diff --git a/tests/test_email.py b/tests/test_email.py
index 08a35e2..6a4532f 100644
--- a/tests/test_email.py
+++ b/tests/test_email.py
@@ -1,6 +1,6 @@
import unittest
-from aprsd import email
+from aprsd.plugins import email
class TestEmail(unittest.TestCase):
diff --git a/tests/test_main.py b/tests/test_main.py
index a140131..bd62b55 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -1,7 +1,7 @@
import sys
import unittest
-from aprsd import email
+from aprsd.plugins import email
if sys.version_info >= (3, 2):
@@ -11,8 +11,8 @@ else:
class TestMain(unittest.TestCase):
- @mock.patch("aprsd.email._imap_connect")
- @mock.patch("aprsd.email._smtp_connect")
+ @mock.patch("aprsd.plugins.email._imap_connect")
+ @mock.patch("aprsd.plugins.email._smtp_connect")
def test_validate_email(self, imap_mock, smtp_mock):
"""Test to make sure we fail."""
imap_mock.return_value = None
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 8ee2cd7..7a5d575 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -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,33 +12,123 @@ 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}
- 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")
- actual = fortune.run(packet)
+ packet = fake.fake_packet(message="fortune")
+ actual = fortune.filter(packet)
self.assertEqual(expected, actual)
@mock.patch("subprocess.check_output")
@@ -50,17 +140,19 @@ class TestPlugin(unittest.TestCase):
mock_output.return_value = "Funny fortune"
expected = "Funny fortune"
- packet = self.fake_packet(message="fortune")
- actual = fortune.run(packet)
+ 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."
- actual = query.run(packet)
+ actual = query.filter(packet)
mock_flush.assert_called_once()
self.assertEqual(expected, actual)
@@ -68,11 +160,11 @@ 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"
- actual = query.run(packet)
+ actual = query.filter(packet)
mock_restart.assert_not_called()
self.assertEqual(expected, actual)
mock_restart.reset_mock()
@@ -80,9 +172,11 @@ class TestPlugin(unittest.TestCase):
# add a message
msg = messaging.TextMessage(self.fromcall, "testing", self.ack)
track.add(msg)
- actual = query.run(packet)
+ 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):
@@ -100,19 +194,17 @@ 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,
)
- actual = time.run(packet)
+ actual = time.filter(packet)
self.assertEqual(None, actual)
cur_time = fuzzy(h, m, 1)
- packet = self.fake_packet(
- fromcall="KFART",
+ packet = fake.fake_packet(
message="time",
msg_number=1,
)
@@ -121,9 +213,11 @@ class TestPlugin(unittest.TestCase):
cur_time,
local_short_str,
)
- actual = time.run(packet)
+ 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()
@@ -134,13 +228,12 @@ 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,
)
- result = ping.run(packet)
+ result = ping.filter(packet)
self.assertEqual(None, result)
def ping_str(h, m, s):
@@ -153,49 +246,46 @@ class TestPlugin(unittest.TestCase):
+ str(s).zfill(2)
)
- packet = self.fake_packet(
- fromcall="KFART",
+ packet = fake.fake_packet(
message="Ping",
msg_number=1,
)
- actual = ping.run(packet)
+ actual = ping.filter(packet)
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.run(packet)
+ actual = ping.filter(packet)
self.assertEqual(expected, actual)
- @mock.patch("aprsd.plugin.PluginManager.get_msg_plugins")
+
+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"
+ expected = f"APRSD ver:{aprsd.__version__} uptime:00:00:00"
version = version_plugin.VersionPlugin(self.config)
- packet = self.fake_packet(
- fromcall="KFART",
+ packet = fake.fake_packet(
message="No",
msg_number=1,
)
- actual = version.run(packet)
+ 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.run(packet)
+ actual = version.filter(packet)
self.assertEqual(expected, actual)
- packet = self.fake_packet(
- fromcall="KFART",
+ packet = fake.fake_packet(
message="Version",
msg_number=1,
)
- actual = version.run(packet)
+ actual = version.filter(packet)
self.assertEqual(expected, actual)
diff --git a/tox.ini b/tox.ini
index 0a415d1..95e1349 100644
--- a/tox.ini
+++ b/tox.ini
@@ -2,7 +2,7 @@
minversion = 2.9.0
skipdist = True
skip_missing_interpreters = true
-envlist = pre-commit,pep8,fmt-check,py{36,37,38}
+envlist = pre-commit,pep8,py{36,37,38}
# Activate isolated build environment. tox will use a virtual environment
# to build a source distribution from the source tree. For build tools and