diff --git a/aprsd/cmds/fetch_stats.py b/aprsd/cmds/fetch_stats.py index f43386a..e037371 100644 --- a/aprsd/cmds/fetch_stats.py +++ b/aprsd/cmds/fetch_stats.py @@ -164,9 +164,10 @@ def fetch_stats(ctx, host, port): def dump_stats(ctx): """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__}") - ss = StatsStore() - ss.load() - stats = ss.data - console.print(stats) + with console.status("Dumping stats"): + ss = StatsStore() + ss.load() + stats = ss.data + console.print(stats) diff --git a/aprsd/packets/core.py b/aprsd/packets/core.py index fa43f3f..52bea31 100644 --- a/aprsd/packets/core.py +++ b/aprsd/packets/core.py @@ -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: diff --git a/aprsd/plugins/fortune.py b/aprsd/plugins/fortune.py index 0ac6bb1..d0f44ca 100644 --- a/aprsd/plugins/fortune.py +++ b/aprsd/plugins/fortune.py @@ -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", "") diff --git a/aprsd/plugins/location.py b/aprsd/plugins/location.py index 4855979..879f12b 100644 --- a/aprsd/plugins/location.py +++ b/aprsd/plugins/location.py @@ -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") diff --git a/tests/client/test_aprsis.py b/tests/client/test_aprsis.py new file mode 100644 index 0000000..e1c7bad --- /dev/null +++ b/tests/client/test_aprsis.py @@ -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() diff --git a/tests/client/test_client_base.py b/tests/client/test_client_base.py new file mode 100644 index 0000000..b6faf2c --- /dev/null +++ b/tests/client/test_client_base.py @@ -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() diff --git a/tests/client/test_factory.py b/tests/client/test_factory.py new file mode 100644 index 0000000..4c2257e --- /dev/null +++ b/tests/client/test_factory.py @@ -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()) diff --git a/tests/test_main.py b/tests/test_main.py index ab1303d..276a42c 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -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")