mirror of
https://github.com/craigerl/aprsd.git
synced 2024-11-21 07:41:49 -05:00
Compare commits
11 Commits
dd3155cea2
...
47d29b059d
Author | SHA1 | Date | |
---|---|---|---|
|
47d29b059d | ||
d0018a8cd3 | |||
2fdc7b111d | |||
229155d0ee | |||
7d22148b0f | |||
563b06876c | |||
579d0c95a0 | |||
224686cac5 | |||
ab2de86726 | |||
f1d066b8a9 | |||
|
1334eded62 |
70
README.rst
70
README.rst
@ -11,6 +11,37 @@ ____________________
|
||||
`APRSD <http://github.com/craigerl/aprsd>`_ is a Ham radio `APRS <http://aprs.org>`_ message command gateway built on python.
|
||||
|
||||
|
||||
Table of Contents
|
||||
=================
|
||||
|
||||
1. `What is APRSD <#what-is-aprsd>`_
|
||||
2. `APRSD Overview Diagram <#aprsd-overview-diagram>`_
|
||||
3. `Typical Use Case <#typical-use-case>`_
|
||||
4. `Installation <#installation>`_
|
||||
5. `Example Usage <#example-usage>`_
|
||||
6. `Help <#help>`_
|
||||
7. `Commands <#commands>`_
|
||||
- `Configuration <#configuration>`_
|
||||
- `Server <#server>`_
|
||||
- `Current List of Built-in Plugins <#current-list-of-built-in-plugins>`_
|
||||
- `Pypi.org APRSD Installable Plugin Packages <#pypiorg-aprsd-installable-plugin-packages>`_
|
||||
- `🐍 APRSD Installed 3rd Party Plugins <#aprsd-installed-3rd-party-plugins>`_
|
||||
- `Send Message <#send-message>`_
|
||||
- `Send Email (Radio to SMTP Server) <#send-email-radio-to-smtp-server>`_
|
||||
- `Receive Email (IMAP Server to Radio) <#receive-email-imap-server-to-radio>`_
|
||||
- `Location <#location>`_
|
||||
- `Web Admin Interface <#web-admin-interface>`_
|
||||
8. `Development <#development>`_
|
||||
- `Building Your Own APRSD Plugins <#building-your-own-aprsd-plugins>`_
|
||||
9. `Workflow <#workflow>`_
|
||||
10. `Release <#release>`_
|
||||
11. `Docker Container <#docker-container>`_
|
||||
- `Building <#building-1>`_
|
||||
- `Official Build <#official-build>`_
|
||||
- `Development Build <#development-build>`_
|
||||
- `Running the Container <#running-the-container>`_
|
||||
|
||||
|
||||
What is APRSD
|
||||
=============
|
||||
APRSD is a python application for interacting with the APRS network and providing
|
||||
@ -147,8 +178,7 @@ look for incomming commands to the callsign configured in the config file
|
||||
|
||||
|
||||
Current list of built-in plugins
|
||||
======================================
|
||||
|
||||
--------------------------------
|
||||
::
|
||||
|
||||
└─> aprsd list-plugins
|
||||
@ -300,18 +330,21 @@ AND... ping, fortune, time.....
|
||||
|
||||
Web Admin Interface
|
||||
===================
|
||||
APRSD has a web admin interface that allows you to view the status of the running APRSD server instance.
|
||||
The web admin interface shows graphs of packet counts, packet types, number of threads running, the latest
|
||||
packets sent and received, and the status of each of the plugins that are loaded. You can also view the logfile
|
||||
and view the raw APRSD configuration file.
|
||||
|
||||
To start the web admin interface, You have to install gunicorn in your virtualenv that already has aprsd installed.
|
||||
|
||||
::
|
||||
|
||||
source <path to APRSD's virtualenv>/bin/activate
|
||||
pip install gunicorn
|
||||
gunicorn --bind 0.0.0.0:8080 "aprsd.wsgi:app"
|
||||
aprsd admin --loglevel INFO
|
||||
|
||||
The web admin interface will be running on port 8080 on the local machine. http://localhost:8080
|
||||
|
||||
|
||||
|
||||
Development
|
||||
===========
|
||||
|
||||
@ -320,7 +353,7 @@ Development
|
||||
* ``make``
|
||||
|
||||
Workflow
|
||||
========
|
||||
--------
|
||||
|
||||
While working aprsd, The workflow is as follows:
|
||||
|
||||
@ -349,7 +382,7 @@ While working aprsd, The workflow is as follows:
|
||||
|
||||
|
||||
Release
|
||||
=======
|
||||
-------
|
||||
|
||||
To do release to pypi:
|
||||
|
||||
@ -370,6 +403,29 @@ To do release to pypi:
|
||||
``make upload``
|
||||
|
||||
|
||||
Building your own APRSD plugins
|
||||
-------------------------------
|
||||
|
||||
APRSD plugins are the mechanism by which APRSD can respond to APRS Messages. The plugins are loaded at server startup
|
||||
and can also be loaded at listen startup. When a packet is received by APRSD, it is passed to each of the plugins
|
||||
in the order they were registered in the config file. The plugins can then decide what to do with the packet.
|
||||
When a plugin is called, it is passed a APRSD Packet object. The plugin can then do something with the packet and
|
||||
return a reply message if desired. If a plugin does not want to reply to the packet, it can just return None.
|
||||
When a plugin does return a reply message, APRSD will send the reply message to the appropriate destination.
|
||||
|
||||
For example, when a 'ping' message is received, the PingPlugin will return a reply message of 'pong'. When APRSD
|
||||
receives the 'pong' message, it will be sent back to the original caller of the ping message.
|
||||
|
||||
APRSD plugins are simply python packages that can be installed from pypi.org. They are installed into the
|
||||
aprsd virtualenv and can be imported by APRSD at runtime. The plugins are registered in the config file and loaded
|
||||
at startup of the aprsd server command or the aprsd listen command.
|
||||
|
||||
Overview
|
||||
--------
|
||||
You can build your own plugins by following the instructions in the `Building your own APRSD plugins`_ section.
|
||||
|
||||
Plugins are called by APRSD when packe
|
||||
|
||||
Docker Container
|
||||
================
|
||||
|
||||
|
@ -159,14 +159,153 @@ def fetch_stats(ctx, host, port):
|
||||
|
||||
@cli.command()
|
||||
@cli_helper.add_options(cli_helper.common_options)
|
||||
@click.option(
|
||||
"--raw",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Dump raw stats instead of formatted output.",
|
||||
)
|
||||
@click.option(
|
||||
"--show-section",
|
||||
default=["All"],
|
||||
help="Show specific sections of the stats. "
|
||||
" Choices: All, APRSDStats, APRSDThreadList, APRSClientStats,"
|
||||
" PacketList, SeenList, WatchList",
|
||||
multiple=True,
|
||||
type=click.Choice(
|
||||
[
|
||||
"All",
|
||||
"APRSDStats",
|
||||
"APRSDThreadList",
|
||||
"APRSClientStats",
|
||||
"PacketList",
|
||||
"SeenList",
|
||||
"WatchList",
|
||||
],
|
||||
case_sensitive=False,
|
||||
),
|
||||
)
|
||||
@click.pass_context
|
||||
@cli_helper.process_standard_options
|
||||
def dump_stats(ctx):
|
||||
def dump_stats(ctx, raw, show_section):
|
||||
"""Dump the current stats from the running APRSD instance."""
|
||||
console = Console()
|
||||
console.print(f"APRSD Fetch-Stats started version: {aprsd.__version__}")
|
||||
console.print(f"APRSD Dump-Stats started version: {aprsd.__version__}")
|
||||
|
||||
with console.status("Dumping stats"):
|
||||
ss = StatsStore()
|
||||
ss.load()
|
||||
stats = ss.data
|
||||
if raw:
|
||||
if "All" in show_section:
|
||||
console.print(stats)
|
||||
return
|
||||
else:
|
||||
for section in show_section:
|
||||
console.print(f"Dumping {section} section:")
|
||||
console.print(stats[section])
|
||||
return
|
||||
|
||||
t = Table(title="APRSD Stats")
|
||||
t.add_column("Key")
|
||||
t.add_column("Value")
|
||||
for key, value in stats["APRSDStats"].items():
|
||||
t.add_row(key, str(value))
|
||||
|
||||
if "All" in show_section or "APRSDStats" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# Show the thread list
|
||||
t = Table(title="Thread List")
|
||||
t.add_column("Name")
|
||||
t.add_column("Class")
|
||||
t.add_column("Alive?")
|
||||
t.add_column("Loop Count")
|
||||
t.add_column("Age")
|
||||
for name, value in stats["APRSDThreadList"].items():
|
||||
t.add_row(
|
||||
name,
|
||||
value["class"],
|
||||
str(value["alive"]),
|
||||
str(value["loop_count"]),
|
||||
str(value["age"]),
|
||||
)
|
||||
|
||||
if "All" in show_section or "APRSDThreadList" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# Show the plugins
|
||||
t = Table(title="Plugin List")
|
||||
t.add_column("Name")
|
||||
t.add_column("Enabled")
|
||||
t.add_column("Version")
|
||||
t.add_column("TX")
|
||||
t.add_column("RX")
|
||||
for name, value in stats["PluginManager"].items():
|
||||
t.add_row(
|
||||
name,
|
||||
str(value["enabled"]),
|
||||
value["version"],
|
||||
str(value["tx"]),
|
||||
str(value["rx"]),
|
||||
)
|
||||
|
||||
if "All" in show_section or "PluginManager" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# Now show the client stats
|
||||
t = Table(title="Client Stats")
|
||||
t.add_column("Key")
|
||||
t.add_column("Value")
|
||||
for key, value in stats["APRSClientStats"].items():
|
||||
t.add_row(key, str(value))
|
||||
|
||||
if "All" in show_section or "APRSClientStats" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# now show the packet list
|
||||
packet_list = stats.get("PacketList")
|
||||
t = Table(title="Packet List")
|
||||
t.add_column("Key")
|
||||
t.add_column("Value")
|
||||
t.add_row("Total Received", str(packet_list["rx"]))
|
||||
t.add_row("Total Sent", str(packet_list["tx"]))
|
||||
|
||||
if "All" in show_section or "PacketList" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# now show the seen list
|
||||
seen_list = stats.get("SeenList")
|
||||
sorted_seen_list = sorted(
|
||||
seen_list.items(),
|
||||
)
|
||||
t = Table(title="Seen List")
|
||||
t.add_column("Callsign")
|
||||
t.add_column("Message Count")
|
||||
t.add_column("Last Heard")
|
||||
for key, value in sorted_seen_list:
|
||||
t.add_row(
|
||||
key,
|
||||
str(value["count"]),
|
||||
str(value["last"]),
|
||||
)
|
||||
|
||||
if "All" in show_section or "SeenList" in show_section:
|
||||
console.print(t)
|
||||
|
||||
# now show the watch list
|
||||
watch_list = stats.get("WatchList")
|
||||
sorted_watch_list = sorted(
|
||||
watch_list.items(),
|
||||
)
|
||||
t = Table(title="Watch List")
|
||||
t.add_column("Callsign")
|
||||
t.add_column("Last Heard")
|
||||
for key, value in sorted_watch_list:
|
||||
t.add_row(
|
||||
key,
|
||||
str(value["last"]),
|
||||
)
|
||||
|
||||
if "All" in show_section or "WatchList" in show_section:
|
||||
console.print(t)
|
||||
|
@ -136,6 +136,11 @@ aprsd_opts = [
|
||||
default=True,
|
||||
help="Set this to False, to disable logging of packets to the log file.",
|
||||
),
|
||||
cfg.BoolOpt(
|
||||
"load_help_plugin",
|
||||
default=True,
|
||||
help="Set this to False to disable the help plugin.",
|
||||
),
|
||||
]
|
||||
|
||||
watch_list_opts = [
|
||||
|
@ -63,15 +63,11 @@ def _init_msgNo(): # noqa: N802
|
||||
|
||||
|
||||
def _translate_fields(raw: dict) -> dict:
|
||||
translate_fields = {
|
||||
"from": "from_call",
|
||||
"to": "to_call",
|
||||
}
|
||||
# First translate some fields
|
||||
for key in translate_fields:
|
||||
if key in raw:
|
||||
raw[translate_fields[key]] = raw[key]
|
||||
del raw[key]
|
||||
# Direct key checks instead of iteration
|
||||
if "from" in raw:
|
||||
raw["from_call"] = raw.pop("from")
|
||||
if "to" in raw:
|
||||
raw["to_call"] = raw.pop("to")
|
||||
|
||||
# addresse overrides to_call
|
||||
if "addresse" in raw:
|
||||
@ -110,11 +106,7 @@ class Packet:
|
||||
via: Optional[str] = field(default=None, compare=False, hash=False)
|
||||
|
||||
def get(self, key: str, default: Optional[str] = None):
|
||||
"""Emulate a getter on a dict."""
|
||||
if hasattr(self, key):
|
||||
return getattr(self, key)
|
||||
else:
|
||||
return default
|
||||
return getattr(self, key, default)
|
||||
|
||||
@property
|
||||
def key(self) -> str:
|
||||
|
@ -472,9 +472,13 @@ class PluginManager:
|
||||
del self._pluggy_pm
|
||||
self.setup_plugins()
|
||||
|
||||
def setup_plugins(self, load_help_plugin=True):
|
||||
def setup_plugins(self, load_help_plugin=None):
|
||||
"""Create the plugin manager and register plugins."""
|
||||
|
||||
# If load_help_plugin is not specified, load it from the config
|
||||
if load_help_plugin is None:
|
||||
load_help_plugin = CONF.load_help_plugin
|
||||
|
||||
LOG.info("Loading APRSD Plugins")
|
||||
# Help plugin is always enabled.
|
||||
if load_help_plugin:
|
||||
|
@ -12,6 +12,7 @@ import imapclient
|
||||
from oslo_config import cfg
|
||||
|
||||
from aprsd import packets, plugin, threads, utils
|
||||
from aprsd.stats import collector
|
||||
from aprsd.threads import tx
|
||||
from aprsd.utils import trace
|
||||
|
||||
@ -126,6 +127,11 @@ class EmailPlugin(plugin.APRSDRegexCommandPluginBase):
|
||||
|
||||
shortcuts = _build_shortcuts_dict()
|
||||
LOG.info(f"Email shortcuts {shortcuts}")
|
||||
|
||||
# Register the EmailStats producer with the stats collector
|
||||
# We do this here to prevent EmailStats from being registered
|
||||
# when email is not enabled in the config file.
|
||||
collector.Collector().register_producer(EmailStats)
|
||||
else:
|
||||
LOG.info("Email services not enabled.")
|
||||
self.enabled = False
|
||||
|
@ -8,7 +8,7 @@ from aprsd.utils import trace
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
DEFAULT_FORTUNE_PATH = '/usr/games/fortune'
|
||||
DEFAULT_FORTUNE_PATH = "/usr/games/fortune"
|
||||
|
||||
|
||||
class FortunePlugin(plugin.APRSDRegexCommandPluginBase):
|
||||
@ -45,7 +45,7 @@ class FortunePlugin(plugin.APRSDRegexCommandPluginBase):
|
||||
command,
|
||||
shell=True,
|
||||
timeout=3,
|
||||
universal_newlines=True,
|
||||
text=True,
|
||||
)
|
||||
output = (
|
||||
output.replace("\r", "")
|
||||
|
@ -2,8 +2,10 @@ import logging
|
||||
import re
|
||||
import time
|
||||
|
||||
from geopy.geocoders import ArcGIS, AzureMaps, Baidu, Bing, GoogleV3
|
||||
from geopy.geocoders import HereV7, Nominatim, OpenCage, TomTom, What3WordsV3, Woosmap
|
||||
from geopy.geocoders import (
|
||||
ArcGIS, AzureMaps, Baidu, Bing, GoogleV3, HereV7, Nominatim, OpenCage,
|
||||
TomTom, What3WordsV3, Woosmap,
|
||||
)
|
||||
from oslo_config import cfg
|
||||
|
||||
from aprsd import packets, plugin, plugin_utils
|
||||
@ -39,8 +41,8 @@ class USGov:
|
||||
result = plugin_utils.get_weather_gov_for_gps(lat, lon)
|
||||
# LOG.info(f"WEATHER: {result}")
|
||||
# LOG.info(f"area description {result['location']['areaDescription']}")
|
||||
if 'location' in result:
|
||||
loc = UsLocation(result['location']['areaDescription'])
|
||||
if "location" in result:
|
||||
loc = UsLocation(result["location"]["areaDescription"])
|
||||
else:
|
||||
loc = UsLocation("Unknown Location")
|
||||
|
||||
|
@ -1,7 +1,6 @@
|
||||
from aprsd import plugin
|
||||
from aprsd.client import stats as client_stats
|
||||
from aprsd.packets import packet_list, seen_list, tracker, watch_list
|
||||
from aprsd.plugins import email
|
||||
from aprsd.stats import app, collector
|
||||
from aprsd.threads import aprsd
|
||||
|
||||
@ -15,6 +14,5 @@ stats_collector.register_producer(watch_list.WatchList)
|
||||
stats_collector.register_producer(tracker.PacketTrack)
|
||||
stats_collector.register_producer(plugin.PluginManager)
|
||||
stats_collector.register_producer(aprsd.APRSDThreadList)
|
||||
stats_collector.register_producer(email.EmailStats)
|
||||
stats_collector.register_producer(client_stats.APRSClientStats)
|
||||
stats_collector.register_producer(seen_list.SeenList)
|
||||
|
@ -328,8 +328,22 @@ class APRSDPluginProcessPacketThread(APRSDProcessPacketThread):
|
||||
# If the message was for us and we didn't have a
|
||||
# response, then we send a usage statement.
|
||||
if to_call == CONF.callsign and not replied:
|
||||
|
||||
# Is the help plugin installed?
|
||||
help_available = False
|
||||
for p in pm.get_message_plugins():
|
||||
if isinstance(p, plugin.HelpPlugin):
|
||||
help_available = True
|
||||
break
|
||||
|
||||
# Tailor the messages accordingly
|
||||
if help_available:
|
||||
LOG.warning("Sending help!")
|
||||
message_text = "Unknown command! Send 'help' message for help"
|
||||
else:
|
||||
LOG.warning("Unknown command!")
|
||||
message_text = "Unknown command!"
|
||||
|
||||
tx.send(
|
||||
packets.MessagePacket(
|
||||
from_call=CONF.callsign,
|
||||
|
81
tests/client/test_aprsis.py
Normal file
81
tests/client/test_aprsis.py
Normal file
@ -0,0 +1,81 @@
|
||||
import datetime
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from aprsd import exception
|
||||
from aprsd.client.aprsis import APRSISClient
|
||||
|
||||
|
||||
class TestAPRSISClient(unittest.TestCase):
|
||||
"""Test cases for APRSISClient."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
super().setUp()
|
||||
|
||||
# Mock the config
|
||||
self.mock_conf = mock.MagicMock()
|
||||
self.mock_conf.aprs_network.enabled = True
|
||||
self.mock_conf.aprs_network.login = "TEST"
|
||||
self.mock_conf.aprs_network.password = "12345"
|
||||
self.mock_conf.aprs_network.host = "localhost"
|
||||
self.mock_conf.aprs_network.port = 14580
|
||||
|
||||
@mock.patch("aprsd.client.base.APRSClient")
|
||||
@mock.patch("aprsd.client.drivers.aprsis.Aprsdis")
|
||||
def test_stats_not_configured(self, mock_aprsdis, mock_base):
|
||||
"""Test stats when client is not configured."""
|
||||
mock_client = mock.MagicMock()
|
||||
mock_aprsdis.return_value = mock_client
|
||||
|
||||
with mock.patch("aprsd.client.aprsis.cfg.CONF", self.mock_conf):
|
||||
self.client = APRSISClient()
|
||||
|
||||
with mock.patch.object(APRSISClient, "is_configured", return_value=False):
|
||||
stats = self.client.stats()
|
||||
self.assertEqual({}, stats)
|
||||
|
||||
@mock.patch("aprsd.client.base.APRSClient")
|
||||
@mock.patch("aprsd.client.drivers.aprsis.Aprsdis")
|
||||
def test_stats_configured(self, mock_aprsdis, mock_base):
|
||||
"""Test stats when client is configured."""
|
||||
mock_client = mock.MagicMock()
|
||||
mock_aprsdis.return_value = mock_client
|
||||
|
||||
with mock.patch("aprsd.client.aprsis.cfg.CONF", self.mock_conf):
|
||||
self.client = APRSISClient()
|
||||
|
||||
mock_client = mock.MagicMock()
|
||||
mock_client.server_string = "test.server:14580"
|
||||
mock_client.aprsd_keepalive = datetime.datetime.now()
|
||||
self.client._client = mock_client
|
||||
self.client.filter = "m/50"
|
||||
|
||||
with mock.patch.object(APRSISClient, "is_configured", return_value=True):
|
||||
stats = self.client.stats()
|
||||
self.assertEqual(
|
||||
{
|
||||
"server_string": mock_client.server_string,
|
||||
"sever_keepalive": mock_client.aprsd_keepalive,
|
||||
"filter": "m/50",
|
||||
}, stats,
|
||||
)
|
||||
|
||||
def test_is_configured_missing_login(self):
|
||||
"""Test is_configured with missing login."""
|
||||
self.mock_conf.aprs_network.login = None
|
||||
with self.assertRaises(exception.MissingConfigOptionException):
|
||||
APRSISClient.is_configured()
|
||||
|
||||
def test_is_configured_missing_password(self):
|
||||
"""Test is_configured with missing password."""
|
||||
self.mock_conf.aprs_network.password = None
|
||||
with self.assertRaises(exception.MissingConfigOptionException):
|
||||
APRSISClient.is_configured()
|
||||
|
||||
def test_is_configured_missing_host(self):
|
||||
"""Test is_configured with missing host."""
|
||||
self.mock_conf.aprs_network.host = None
|
||||
with mock.patch("aprsd.client.aprsis.cfg.CONF", self.mock_conf):
|
||||
with self.assertRaises(exception.MissingConfigOptionException):
|
||||
APRSISClient.is_configured()
|
140
tests/client/test_client_base.py
Normal file
140
tests/client/test_client_base.py
Normal file
@ -0,0 +1,140 @@
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from aprsd.client.base import APRSClient
|
||||
from aprsd.packets import core
|
||||
|
||||
|
||||
class MockAPRSClient(APRSClient):
|
||||
"""Concrete implementation of APRSClient for testing."""
|
||||
|
||||
def stats(self):
|
||||
return {"packets_received": 0, "packets_sent": 0}
|
||||
|
||||
def setup_connection(self):
|
||||
mock_connection = mock.MagicMock()
|
||||
# Configure the mock with required methods
|
||||
mock_connection.close = mock.MagicMock()
|
||||
mock_connection.stop = mock.MagicMock()
|
||||
mock_connection.set_filter = mock.MagicMock()
|
||||
mock_connection.send = mock.MagicMock()
|
||||
self._client = mock_connection
|
||||
return mock_connection
|
||||
|
||||
def decode_packet(self, *args, **kwargs):
|
||||
return mock.MagicMock()
|
||||
|
||||
def consumer(self, callback, blocking=False, immortal=False, raw=False):
|
||||
pass
|
||||
|
||||
def is_alive(self):
|
||||
return True
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def is_enabled():
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def transport():
|
||||
return "mock"
|
||||
|
||||
def reset(self):
|
||||
"""Mock implementation of reset."""
|
||||
if self._client:
|
||||
self._client.close()
|
||||
self._client = self.setup_connection()
|
||||
if self.filter:
|
||||
self._client.set_filter(self.filter)
|
||||
|
||||
|
||||
class TestAPRSClient(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# Reset the singleton instance before each test
|
||||
APRSClient._instance = None
|
||||
APRSClient._client = None
|
||||
self.client = MockAPRSClient()
|
||||
|
||||
def test_singleton_pattern(self):
|
||||
"""Test that multiple instantiations return the same instance."""
|
||||
client1 = MockAPRSClient()
|
||||
client2 = MockAPRSClient()
|
||||
self.assertIs(client1, client2)
|
||||
|
||||
def test_set_filter(self):
|
||||
"""Test setting APRS filter."""
|
||||
# Get the existing mock client that was created in __init__
|
||||
mock_client = self.client._client
|
||||
|
||||
test_filter = "m/50"
|
||||
self.client.set_filter(test_filter)
|
||||
self.assertEqual(self.client.filter, test_filter)
|
||||
# The filter is set once during set_filter() and once during reset()
|
||||
mock_client.set_filter.assert_called_with(test_filter)
|
||||
|
||||
@mock.patch("aprsd.client.base.LOG")
|
||||
def test_reset(self, mock_log):
|
||||
"""Test client reset functionality."""
|
||||
# Create a new mock client with the necessary methods
|
||||
old_client = mock.MagicMock()
|
||||
self.client._client = old_client
|
||||
|
||||
self.client.reset()
|
||||
|
||||
# Verify the old client was closed
|
||||
old_client.close.assert_called_once()
|
||||
|
||||
# Verify a new client was created
|
||||
self.assertIsNotNone(self.client._client)
|
||||
self.assertNotEqual(old_client, self.client._client)
|
||||
|
||||
def test_send_packet(self):
|
||||
"""Test sending an APRS packet."""
|
||||
mock_packet = mock.Mock(spec=core.Packet)
|
||||
self.client.send(mock_packet)
|
||||
self.client._client.send.assert_called_once_with(mock_packet)
|
||||
|
||||
def test_stop(self):
|
||||
"""Test stopping the client."""
|
||||
# Ensure client is created first
|
||||
self.client._create_client()
|
||||
|
||||
self.client.stop()
|
||||
self.client._client.stop.assert_called_once()
|
||||
|
||||
@mock.patch("aprsd.client.base.LOG")
|
||||
def test_create_client_failure(self, mock_log):
|
||||
"""Test handling of client creation failure."""
|
||||
# Make setup_connection raise an exception
|
||||
with mock.patch.object(
|
||||
self.client, "setup_connection",
|
||||
side_effect=Exception("Connection failed"),
|
||||
):
|
||||
with self.assertRaises(Exception):
|
||||
self.client._create_client()
|
||||
|
||||
self.assertIsNone(self.client._client)
|
||||
mock_log.error.assert_called_once()
|
||||
|
||||
def test_client_property(self):
|
||||
"""Test the client property creates client if none exists."""
|
||||
self.client._client = None
|
||||
client = self.client.client
|
||||
self.assertIsNotNone(client)
|
||||
|
||||
def test_filter_applied_on_creation(self):
|
||||
"""Test that filter is applied when creating new client."""
|
||||
test_filter = "m/50"
|
||||
self.client.set_filter(test_filter)
|
||||
|
||||
# Force client recreation
|
||||
self.client.reset()
|
||||
|
||||
# Verify filter was applied to new client
|
||||
self.client._client.set_filter.assert_called_with(test_filter)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
75
tests/client/test_factory.py
Normal file
75
tests/client/test_factory.py
Normal file
@ -0,0 +1,75 @@
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from aprsd.client.factory import Client, ClientFactory
|
||||
|
||||
|
||||
class MockClient:
|
||||
"""Mock client for testing."""
|
||||
|
||||
@classmethod
|
||||
def is_enabled(cls):
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def is_configured(cls):
|
||||
return True
|
||||
|
||||
|
||||
class TestClientFactory(unittest.TestCase):
|
||||
"""Test cases for ClientFactory."""
|
||||
|
||||
def setUp(self):
|
||||
"""Set up test fixtures."""
|
||||
self.factory = ClientFactory()
|
||||
# Clear any registered clients from previous tests
|
||||
self.factory.clients = []
|
||||
|
||||
def test_singleton(self):
|
||||
"""Test that ClientFactory is a singleton."""
|
||||
factory2 = ClientFactory()
|
||||
self.assertEqual(self.factory, factory2)
|
||||
|
||||
def test_register_client(self):
|
||||
"""Test registering a client."""
|
||||
self.factory.register(MockClient)
|
||||
self.assertIn(MockClient, self.factory.clients)
|
||||
|
||||
def test_register_invalid_client(self):
|
||||
"""Test registering an invalid client raises error."""
|
||||
invalid_client = mock.MagicMock(spec=Client)
|
||||
with self.assertRaises(ValueError):
|
||||
self.factory.register(invalid_client)
|
||||
|
||||
def test_create_client(self):
|
||||
"""Test creating a client."""
|
||||
self.factory.register(MockClient)
|
||||
client = self.factory.create()
|
||||
self.assertIsInstance(client, MockClient)
|
||||
|
||||
def test_create_no_clients(self):
|
||||
"""Test creating a client with no registered clients."""
|
||||
with self.assertRaises(Exception):
|
||||
self.factory.create()
|
||||
|
||||
def test_is_client_enabled(self):
|
||||
"""Test checking if any client is enabled."""
|
||||
self.factory.register(MockClient)
|
||||
self.assertTrue(self.factory.is_client_enabled())
|
||||
|
||||
def test_is_client_enabled_none(self):
|
||||
"""Test checking if any client is enabled when none are."""
|
||||
MockClient.is_enabled = classmethod(lambda cls: False)
|
||||
self.factory.register(MockClient)
|
||||
self.assertFalse(self.factory.is_client_enabled())
|
||||
|
||||
def test_is_client_configured(self):
|
||||
"""Test checking if any client is configured."""
|
||||
self.factory.register(MockClient)
|
||||
self.assertTrue(self.factory.is_client_configured())
|
||||
|
||||
def test_is_client_configured_none(self):
|
||||
"""Test checking if any client is configured when none are."""
|
||||
MockClient.is_configured = classmethod(lambda cls: False)
|
||||
self.factory.register(MockClient)
|
||||
self.assertFalse(self.factory.is_client_configured())
|
@ -1,15 +1,9 @@
|
||||
import sys
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from aprsd.plugins import email
|
||||
|
||||
|
||||
if sys.version_info >= (3, 2):
|
||||
from unittest import mock
|
||||
else:
|
||||
from unittest import mock
|
||||
|
||||
|
||||
class TestMain(unittest.TestCase):
|
||||
@mock.patch("aprsd.plugins.email._imap_connect")
|
||||
@mock.patch("aprsd.plugins.email._smtp_connect")
|
||||
|
Loading…
Reference in New Issue
Block a user