aprsd/tests/packets/test_tracker.py

import unittest

from aprsd.packets import tracker
from tests import fake


class TestPacketTrack(unittest.TestCase):
    """Unit tests for the PacketTrack class."""

    def setUp(self):
        """Set up test fixtures."""
        # Reset singleton instance
        tracker.PacketTrack._instance = None
        tracker.PacketTrack.data = {}
        tracker.PacketTrack.total_tracked = 0

    def tearDown(self):
        """Clean up after tests."""
        tracker.PacketTrack._instance = None
        tracker.PacketTrack.data = {}
        tracker.PacketTrack.total_tracked = 0

    def test_singleton_pattern(self):
        """Test that PacketTrack is a singleton."""
        pt1 = tracker.PacketTrack()
        pt2 = tracker.PacketTrack()
        self.assertIs(pt1, pt2)

    def test_init(self):
        """Test initialization."""
        pt = tracker.PacketTrack()
        self.assertIsInstance(pt.data, dict)
        self.assertIsNotNone(pt._start_time)

    def test_getitem(self):
        """Test __getitem__() method."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        result = pt['123']
        self.assertEqual(result, packet)

    def test_iter(self):
        """Test __iter__() method."""
        pt = tracker.PacketTrack()
        packet1 = fake.fake_packet(msg_number='123')
        packet2 = fake.fake_packet(msg_number='456')
        pt.tx(packet1)
        pt.tx(packet2)

        keys = list(iter(pt))
        self.assertIn('123', keys)
        self.assertIn('456', keys)

    def test_keys(self):
        """Test keys() method."""
        pt = tracker.PacketTrack()
        packet1 = fake.fake_packet(msg_number='123')
        packet2 = fake.fake_packet(msg_number='456')
        pt.tx(packet1)
        pt.tx(packet2)

        keys = list(pt.keys())
        self.assertIn('123', keys)
        self.assertIn('456', keys)

    def test_items(self):
        """Test items() method."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        items = list(pt.items())
        self.assertEqual(len(items), 1)
        self.assertEqual(items[0][0], '123')
        self.assertEqual(items[0][1], packet)

    def test_values(self):
        """Test values() method."""
        pt = tracker.PacketTrack()
        packet1 = fake.fake_packet(msg_number='123')
        packet2 = fake.fake_packet(msg_number='456')
        pt.tx(packet1)
        pt.tx(packet2)

        values = list(pt.values())
        self.assertEqual(len(values), 2)
        self.assertIn(packet1, values)
        self.assertIn(packet2, values)

    def test_tx(self):
        """Test tx() method."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        initial_total = pt.total_tracked

        pt.tx(packet)

        self.assertIn('123', pt.data)
        self.assertEqual(pt.data['123'], packet)
        self.assertEqual(pt.total_tracked, initial_total + 1)
        self.assertEqual(packet.send_count, 0)

    def test_rx_ack_packet(self):
        """Test rx() with AckPacket."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        ack = fake.fake_ack_packet()
        ack.msgNo = '123'
        pt.rx(ack)

        self.assertNotIn('123', pt.data)

    def test_rx_reject_packet(self):
        """Test rx() with RejectPacket."""
        from aprsd.packets import core

        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        # Create a proper RejectPacket
        reject_pkt = core.RejectPacket(from_call='TEST', to_call='TEST', msgNo='123')
        pt.rx(reject_pkt)

        self.assertNotIn('123', pt.data)

    def test_rx_piggyback_ack(self):
        """Test rx() with piggyback ACK."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        piggyback = fake.fake_packet()
        piggyback.ackMsgNo = '123'
        pt.rx(piggyback)

        self.assertNotIn('123', pt.data)

    def test_rx_no_match(self):
        """Test rx() with packet that doesn't match tracked packet."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        ack = fake.fake_ack_packet()
        ack.msgNo = '999'  # Different msgNo
        pt.rx(ack)

        # Should still have original packet
        self.assertIn('123', pt.data)

    def test_remove(self):
        """Test remove() method."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        pt.remove('123')
        self.assertNotIn('123', pt.data)

    def test_remove_nonexistent(self):
        """Test remove() with nonexistent key."""
        pt = tracker.PacketTrack()
        # Should not raise exception
        pt.remove('nonexistent')

    def test_stats(self):
        """Test stats() method."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        packet.retry_count = 3
        packet.last_send_time = 1000
        pt.tx(packet)
        # Note: tx() resets send_count to 0

        stats = pt.stats()
        self.assertIn('total_tracked', stats)
        self.assertIn('packets', stats)
        self.assertIn('123', stats['packets'])
        self.assertEqual(stats['packets']['123']['send_count'], 0)
        self.assertEqual(stats['packets']['123']['retry_count'], 3)

    def test_stats_serializable(self):
        """Test stats() with serializable=True."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        stats = pt.stats(serializable=True)
        # Should be JSON serializable
        import json

        json.dumps(stats)  # Should not raise exception

    def test_get(self):
        """Test get() method from ObjectStoreMixin."""
        pt = tracker.PacketTrack()
        packet = fake.fake_packet(msg_number='123')
        pt.tx(packet)

        result = pt.get('123')
        self.assertEqual(result, packet)

        result = pt.get('nonexistent')
        self.assertIsNone(result)

    def test_len(self):
        """Test __len__() method."""
        pt = tracker.PacketTrack()
        self.assertEqual(len(pt), 0)

        pt.tx(fake.fake_packet(msg_number='123'))
        self.assertEqual(len(pt), 1)

        pt.tx(fake.fake_packet(msg_number='456'))
        self.assertEqual(len(pt), 2)