hemna/aprsd
1
0
Fork 0
mirror of https://github.com/craigerl/aprsd.git synced 2024-11-17 13:51:54 -05:00
Code Issues Projects Releases Wiki Activity

More messaging -> packets cleanup

Browse Source 
Fixed the unit tests and the notify plugin
This commit is contained in:
Hemna 2022-12-16 15:28:31 -05:00
parent bfc0a5a1e9
commit 6030cb394b
8 changed files with 49 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
aprsd/plugins/notify.py
View File

@ -43,8 +43,8 @@ class NotifySeenPlugin(plugin.APRSDWatchListPluginBase):
message_text=(
f"{fromcall} was just seen by type:'{packet_type}'"
),
_allow_delay=False,
)
pkt._allow_delay = False
return pkt
else:
LOG.debug("fromcall and notify_callsign are the same, not notifying")

17
tests/cmds/test_webchat.py
View File

@ -7,8 +7,9 @@ import flask
import flask_socketio
from aprsd import config as aprsd_config
from aprsd import messaging, packets
from aprsd import packets
from aprsd.cmds import webchat # noqa
from aprsd.packets import core
from .. import fake
@ -63,12 +64,11 @@ class TestSendMessageCommand(unittest.TestCase):
self.assertIsInstance(socketio, flask_socketio.SocketIO)
self.assertIsInstance(flask_app, flask.Flask)
@mock.patch("aprsd.messaging.log_message")
@mock.patch("aprsd.config.parse_config")
@mock.patch("aprsd.messaging.MsgTrack.remove")
@mock.patch("aprsd.packets.tracker.PacketTrack.remove")
@mock.patch("aprsd.cmds.webchat.socketio.emit")
def test_process_ack_packet(
self, mock_parse_config, mock_log_message,
self, mock_parse_config,
mock_remove, mock_emit,
):
config = self._build_config()
@ -76,17 +76,16 @@ class TestSendMessageCommand(unittest.TestCase):
packet = fake.fake_packet(
message="blah",
msg_number=1,
message_format=packets.PACKET_TYPE_ACK,
message_format=core.PACKET_TYPE_ACK,
)
socketio = mock.MagicMock()
packets.PacketList(config=config)
messaging.MsgTrack(config=config)
packets.PacketTrack(config=config)
packets.WatchList(config=config)
packets.SeenList(config=config)
wcp = webchat.WebChatProcessPacketThread(config, packet, socketio)
wcp.process_ack_packet(packet)
mock_log_message.called_once()
mock_remove.called_once()
mock_emit.called_once()
@ -103,11 +102,11 @@ class TestSendMessageCommand(unittest.TestCase):
packet = fake.fake_packet(
message="blah",
msg_number=1,
message_format=packets.PACKET_TYPE_MESSAGE,
message_format=core.PACKET_TYPE_MESSAGE,
)
socketio = mock.MagicMock()
packets.PacketList(config=config)
messaging.MsgTrack(config=config)
packets.PacketTrack(config=config)
packets.WatchList(config=config)
packets.SeenList(config=config)
wcp = webchat.WebChatProcessPacketThread(config, packet, socketio)

3
tests/fake.py
View File

@ -1,4 +1,5 @@
from aprsd import packets, plugin, threads
from aprsd.packets import core
FAKE_MESSAGE_TEXT = "fake MeSSage"
@ -11,7 +12,7 @@ def fake_packet(
tocall=FAKE_TO_CALLSIGN,
message=None,
msg_number=None,
message_format=packets.PACKET_TYPE_MESSAGE,
message_format=core.PACKET_TYPE_MESSAGE,
):
packet_dict = {
"from": fromcall,

29
tests/plugins/test_notify.py
View File

@ -2,7 +2,7 @@ from unittest import mock
from aprsd import client
from aprsd import config as aprsd_config
from aprsd import messaging, packets
from aprsd import packets
from aprsd.plugins import notify as notify_plugin
from .. import fake, test_plugin
@ -28,7 +28,8 @@ class TestWatchListPlugin(test_plugin.TestPlugin):
default_wl = aprsd_config.DEFAULT_CONFIG_DICT["aprsd"]["watch_list"]
_config["ham"]["callsign"] = self.fromcall
_config["aprs"]["login"] = fake.FAKE_TO_CALLSIGN
_config["aprsd"]["callsign"] = self.fromcall
_config["aprs"]["login"] = self.fromcall
_config["services"]["aprs.fi"]["apiKey"] = "something"
# Set the watchlist specific config options
@ -62,7 +63,7 @@ class TestAPRSDWatchListPluginBase(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@mock.patch("aprsd.client.ClientFactory", autospec=True)
@ -78,7 +79,7 @@ class TestAPRSDWatchListPluginBase(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@ -94,7 +95,7 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@mock.patch("aprsd.client.ClientFactory", autospec=True)
@ -109,7 +110,7 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@mock.patch("aprsd.client.ClientFactory", autospec=True)
@ -130,7 +131,7 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@mock.patch("aprsd.client.ClientFactory", autospec=True)
@ -152,7 +153,7 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
msg_number=1,
)
actual = plugin.filter(packet)
expected = messaging.NULL_MESSAGE
expected = packets.NULL_MESSAGE
self.assertEqual(expected, actual)
@mock.patch("aprsd.client.ClientFactory", autospec=True)
@ -160,7 +161,7 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
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"
notify_callsign = fake.FAKE_TO_CALLSIGN
fromcall = "WB4BOR"
config = self._config(
watchlist_enabled=True,
@ -175,11 +176,11 @@ class TestNotifySeenPlugin(TestWatchListPlugin):
message="ping",
msg_number=1,
)
packet_type = packets.get_packet_type(packet)
packet_type = packet.__class__.__name__
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)
self.assertIsInstance(actual, packets.MessagePacket)
self.assertEqual(fake.FAKE_FROM_CALLSIGN, actual.from_call)
self.assertEqual(notify_callsign, actual.to_call)
self.assertEqual(msg, actual.message_text)

17
tests/plugins/test_query.py
View File

@ -1,13 +1,14 @@
from unittest import mock
from aprsd import messaging
from aprsd import packets
from aprsd.packets import tracker
from aprsd.plugins import query as query_plugin
from .. import fake, test_plugin
class TestQueryPlugin(test_plugin.TestPlugin):
@mock.patch("aprsd.messaging.MsgTrack.flush")
@mock.patch("aprsd.packets.tracker.PacketTrack.flush")
def test_query_flush(self, mock_flush):
packet = fake.fake_packet(message="!delete")
query = query_plugin.QueryPlugin(self.config)
@ -17,9 +18,9 @@ class TestQueryPlugin(test_plugin.TestPlugin):
mock_flush.assert_called_once()
self.assertEqual(expected, actual)
@mock.patch("aprsd.messaging.MsgTrack.restart_delayed")
@mock.patch("aprsd.packets.tracker.PacketTrack.restart_delayed")
def test_query_restart_delayed(self, mock_restart):
track = messaging.MsgTrack()
track = tracker.PacketTrack()
track.data = {}
packet = fake.fake_packet(message="!4")
query = query_plugin.QueryPlugin(self.config)
@ -31,7 +32,11 @@ class TestQueryPlugin(test_plugin.TestPlugin):
mock_restart.reset_mock()
# add a message
msg = messaging.TextMessage(self.fromcall, "testing", self.ack)
track.add(msg)
pkt = packets.MessagePacket(
from_call=self.fromcall,
to_call="testing",
msgNo=self.ack,
)
track.add(pkt)
actual = query.filter(packet)
mock_restart.assert_called_once()

155
tests/test_messaging.py
View File

@ -1,155 +0,0 @@
import datetime
import unittest
from unittest import mock
from aprsd import messaging
class TestMessageTrack(unittest.TestCase):
def setUp(self) -> None:
config = {}
messaging.MsgTrack(config=config)
def _clean_track(self):
track = messaging.MsgTrack()
track.data = {}
track.total_messages_tracked = 0
return track
def test_create(self):
track1 = messaging.MsgTrack()
track2 = messaging.MsgTrack()
self.assertEqual(track1, track2)
def test_add(self):
track = self._clean_track()
fromcall = "KFART"
tocall = "KHELP"
message = "somthing"
msg = messaging.TextMessage(fromcall, tocall, message)
track.add(msg)
self.assertEqual(msg, track.get(msg.id))
def test_remove(self):
track = self._clean_track()
fromcall = "KFART"
tocall = "KHELP"
message = "somthing"
msg = messaging.TextMessage(fromcall, tocall, message)
track.add(msg)
track.remove(msg.id)
self.assertEqual(None, track.get(msg.id))
def test_len(self):
"""Test getting length of tracked messages."""
track = self._clean_track()
fromcall = "KFART"
tocall = "KHELP"
message = "somthing"
msg = messaging.TextMessage(fromcall, tocall, message)
track.add(msg)
self.assertEqual(1, len(track))
msg2 = messaging.TextMessage(tocall, fromcall, message)
track.add(msg2)
self.assertEqual(2, len(track))
track.remove(msg.id)
self.assertEqual(1, len(track))
@mock.patch("aprsd.messaging.TextMessage.send")
def test__resend(self, mock_send):
"""Test the _resend method."""
track = self._clean_track()
fromcall = "KFART"
tocall = "KHELP"
message = "somthing"
msg = messaging.TextMessage(fromcall, tocall, message)
msg.last_send_attempt = 3
track.add(msg)
track._resend(msg)
msg.send.assert_called_with()
self.assertEqual(0, msg.last_send_attempt)
@mock.patch("aprsd.messaging.TextMessage.send")
def test_restart_delayed(self, mock_send):
"""Test the _resend method."""
track = self._clean_track()
fromcall = "KFART"
tocall = "KHELP"
message1 = "something"
message2 = "something another"
message3 = "something another again"
mock1_send = mock.MagicMock()
mock2_send = mock.MagicMock()
mock3_send = mock.MagicMock()
msg1 = messaging.TextMessage(fromcall, tocall, message1)
msg1.last_send_attempt = 3
msg1.last_send_time = datetime.datetime.now()
msg1.send = mock1_send
track.add(msg1)
msg2 = messaging.TextMessage(tocall, fromcall, message2)
msg2.last_send_attempt = 3
msg2.last_send_time = datetime.datetime.now()
msg2.send = mock2_send
track.add(msg2)
track.restart_delayed(count=None)
msg1.send.assert_called_once()
self.assertEqual(0, msg1.last_send_attempt)
msg2.send.assert_called_once()
self.assertEqual(0, msg2.last_send_attempt)
msg1.last_send_attempt = 3
msg1.send.reset_mock()
msg2.last_send_attempt = 3
msg2.send.reset_mock()
track.restart_delayed(count=1)
msg1.send.assert_not_called()
msg2.send.assert_called_once()
self.assertEqual(3, msg1.last_send_attempt)
self.assertEqual(0, msg2.last_send_attempt)
msg3 = messaging.TextMessage(tocall, fromcall, message3)
msg3.last_send_attempt = 3
msg3.last_send_time = datetime.datetime.now()
msg3.send = mock3_send
track.add(msg3)
msg1.last_send_attempt = 3
msg1.send.reset_mock()
msg2.last_send_attempt = 3
msg2.send.reset_mock()
msg3.last_send_attempt = 3
msg3.send.reset_mock()
track.restart_delayed(count=2)
msg1.send.assert_not_called()
msg2.send.assert_called_once()
msg3.send.assert_called_once()
self.assertEqual(3, msg1.last_send_attempt)
self.assertEqual(0, msg2.last_send_attempt)
self.assertEqual(0, msg3.last_send_attempt)
msg1.last_send_attempt = 3
msg1.send.reset_mock()
msg2.last_send_attempt = 3
msg2.send.reset_mock()
msg3.last_send_attempt = 3
msg3.send.reset_mock()
track.restart_delayed(count=2, most_recent=False)
msg1.send.assert_called_once()
msg2.send.assert_called_once()
msg3.send.assert_not_called()
self.assertEqual(0, msg1.last_send_attempt)
self.assertEqual(0, msg2.last_send_attempt)
self.assertEqual(3, msg3.last_send_attempt)

11
tests/test_packets.py
View File

@ -1,6 +1,7 @@
import unittest
from aprsd import packets
from aprsd.packets import core
from . import fake
@ -13,7 +14,7 @@ class TestPluginBase(unittest.TestCase):
to_call=fake.FAKE_TO_CALLSIGN,
message=None,
msg_number=None,
message_format=packets.PACKET_TYPE_MESSAGE,
message_format=core.PACKET_TYPE_MESSAGE,
):
packet_dict = {
"from": from_call,
@ -56,9 +57,9 @@ class TestPluginBase(unittest.TestCase):
pkt = packets.Packet.factory(pkt_dict)
self.assertIsInstance(pkt, packets.MessagePacket)
self.assertEqual(pkt_dict["from"], pkt.from_call)
self.assertEqual(pkt_dict["to"], pkt.to_call)
self.assertEqual(pkt_dict["addresse"], pkt.addresse)
self.assertEqual(fake.FAKE_FROM_CALLSIGN, pkt.from_call)
self.assertEqual(fake.FAKE_TO_CALLSIGN, pkt.to_call)
self.assertEqual(fake.FAKE_TO_CALLSIGN, pkt.addresse)
pkt_dict["symbol"] = "_"
pkt_dict["weather"] = {
@ -68,6 +69,6 @@ class TestPluginBase(unittest.TestCase):
"pressure": 1095.12,
"comment": "Home!",
}
pkt_dict["format"] = packets.PACKET_TYPE_UNCOMPRESSED
pkt_dict["format"] = core.PACKET_TYPE_UNCOMPRESSED
pkt = packets.Packet.factory(pkt_dict)
self.assertIsInstance(pkt, packets.WeatherPacket)

11
tests/test_plugin.py
View File

@ -2,7 +2,8 @@ import unittest
from unittest import mock
from aprsd import config as aprsd_config
from aprsd import messaging, packets, stats
from aprsd import packets, stats
from aprsd.packets import core
from . import fake
@ -18,7 +19,7 @@ class TestPlugin(unittest.TestCase):
stats.APRSDStats._instance = None
packets.WatchList._instance = None
packets.SeenList._instance = None
messaging.MsgTrack._instance = None
packets.PacketTrack._instance = None
self.config = None
def config_and_init(self, config=None):
@ -34,7 +35,7 @@ class TestPlugin(unittest.TestCase):
stats.APRSDStats(self.config)
packets.WatchList(config=self.config)
packets.SeenList(config=self.config)
messaging.MsgTrack(config=self.config)
packets.PacketTrack(config=self.config)
class TestPluginBase(TestPlugin):
@ -89,7 +90,7 @@ class TestPluginBase(TestPlugin):
packet = fake.fake_packet(
message="F",
message_format=packets.PACKET_TYPE_MICE,
message_format=core.PACKET_TYPE_MICE,
)
expected = None
actual = p.filter(packet)
@ -98,7 +99,7 @@ class TestPluginBase(TestPlugin):
packet = fake.fake_packet(
message="f",
message_format=packets.PACKET_TYPE_ACK,
message_format=core.PACKET_TYPE_ACK,
)
expected = None
actual = p.filter(packet)