Merge pull request #181 from craigerl/unit-tests

Added unit test for client base

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)

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:

aprsd/plugins/fortune.py

@@ -8,7 +8,7 @@
 
 LOG = logging.getLogger("APRSD")
 
-DEFAULT_FORTUNE_PATH = '/usr/games/fortune'
+DEFAULT_FORTUNE_PATH = "/usr/games/fortune"
 
 
 class FortunePlugin(plugin.APRSDRegexCommandPluginBase):
@@ -45,7 +45,7 @@ def process(self, packet: packets.MessagePacket):
                 command,
                 shell=True,
                 timeout=3,
-                universal_newlines=True,
+                text=True,
             )
             output = (
                 output.replace("\r", "")

aprsd/plugins/location.py

@@ -2,8 +2,10 @@
 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 @@ def reverse(self, coordinates):
         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")
 

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()

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()

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())