Add tests to improve patch coverage for PR #148

Covers duplicate event dedup/observer paths, IntegrityError handling,
sender name resolution, observer lists in API responses, subscriber
dispatch lifecycle, MQTT retry, and CLI group/run/seed commands.

tests/test_api/test_advertisements.py

@@ -2,6 +2,8 @@
 
 from datetime import datetime, timedelta, timezone
 
+from meshcore_hub.common.models import Advertisement, EventObserver
+
 
 class TestListAdvertisements:
     """Tests for GET /advertisements endpoint."""
@@ -24,6 +26,64 @@ class TestListAdvertisements:
         assert data["items"][0]["public_key"] == sample_advertisement.public_key
         assert data["items"][0]["adv_type"] == sample_advertisement.adv_type
 
+    def test_list_advertisements_with_observers(
+        self,
+        client_no_auth,
+        api_db_session,
+        receiver_node,
+    ):
+        """Test that observers list is included in advertisement response."""
+        from hashlib import md5
+
+        event_hash = md5(b"test-ad-observers").hexdigest()
+        advert = Advertisement(
+            public_key="obs123obs123obs123obs123obs123ob",
+            name="ObservedAd",
+            adv_type="REPEATER",
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=receiver_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(advert)
+        api_db_session.commit()
+
+        observer = EventObserver(
+            event_type="advertisement",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+        api_db_session.add(observer)
+        api_db_session.commit()
+
+        response = client_no_auth.get("/api/v1/advertisements")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        item = data["items"][0]
+        assert "observers" in item
+        assert len(item["observers"]) == 1
+
+    def test_list_advertisements_with_node_tag_name(
+        self, client_no_auth, api_db_session, sample_node_with_name_tag
+    ):
+        """Test that node_tag_name is resolved from name tags."""
+        advert = Advertisement(
+            public_key=sample_node_with_name_tag.public_key,
+            name="AdName",
+            adv_type="CLIENT",
+            received_at=datetime.now(timezone.utc),
+            node_id=sample_node_with_name_tag.id,
+        )
+        api_db_session.add(advert)
+        api_db_session.commit()
+
+        response = client_no_auth.get("/api/v1/advertisements")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["node_tag_name"] == "Friendly Search Name"
+
     def test_list_advertisements_filter_by_public_key(
         self, client_no_auth, sample_advertisement
     ):
@@ -58,6 +118,62 @@ class TestGetAdvertisement:
         response = client_no_auth.get("/api/v1/advertisements/nonexistent-id")
         assert response.status_code == 404
 
+    def test_get_advertisement_with_observers(
+        self,
+        client_no_auth,
+        api_db_session,
+        receiver_node,
+    ):
+        """Test that get includes observers list."""
+        from hashlib import md5
+
+        event_hash = md5(b"test-get-ad-observers").hexdigest()
+        advert = Advertisement(
+            public_key="getobs123getobs123getobs123getob",
+            name="GetObservedAd",
+            adv_type="REPEATER",
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=receiver_node.id,
+            event_hash=event_hash,
+        )
+        api_db_session.add(advert)
+        api_db_session.commit()
+
+        observer = EventObserver(
+            event_type="advertisement",
+            event_hash=event_hash,
+            observer_node_id=receiver_node.id,
+            observed_at=datetime.now(timezone.utc),
+        )
+        api_db_session.add(observer)
+        api_db_session.commit()
+
+        response = client_no_auth.get(f"/api/v1/advertisements/{advert.id}")
+        assert response.status_code == 200
+        data = response.json()
+        assert "observers" in data
+        assert len(data["observers"]) == 1
+        assert data["observers"][0]["public_key"] == receiver_node.public_key
+
+    def test_get_advertisement_with_tag_names(
+        self, client_no_auth, api_db_session, sample_node_with_name_tag
+    ):
+        """Test that get includes node_tag_name and observer_tag_name."""
+        advert = Advertisement(
+            public_key=sample_node_with_name_tag.public_key,
+            name="AdName",
+            adv_type="CLIENT",
+            received_at=datetime.now(timezone.utc),
+            node_id=sample_node_with_name_tag.id,
+        )
+        api_db_session.add(advert)
+        api_db_session.commit()
+
+        response = client_no_auth.get(f"/api/v1/advertisements/{advert.id}")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["node_tag_name"] == "Friendly Search Name"
+
 
 class TestListAdvertisementsFilters:
     """Tests for advertisement list query filters."""
@@ -102,8 +218,6 @@ class TestListAdvertisementsFilters:
         self, client_no_auth, api_db_session, sample_node_with_member_tag
     ):
         """Test filtering advertisements by member_id tag."""
-        from meshcore_hub.common.models import Advertisement
-
         # Create an advertisement for the node with member tag
         advert = Advertisement(
             public_key=sample_node_with_member_tag.public_key,
@@ -129,8 +243,6 @@ class TestListAdvertisementsFilters:
 
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering advertisements by since timestamp."""
-        from meshcore_hub.common.models import Advertisement
-
         now = datetime.now(timezone.utc)
         old_time = now - timedelta(days=7)
 
@@ -153,8 +265,6 @@ class TestListAdvertisementsFilters:
 
     def test_filter_by_until(self, client_no_auth, api_db_session):
         """Test filtering advertisements by until timestamp."""
-        from meshcore_hub.common.models import Advertisement
-
         now = datetime.now(timezone.utc)
         old_time = now - timedelta(days=7)
 

tests/test_api/test_messages.py

@@ -1,6 +1,8 @@
 """Tests for message API routes."""
 
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
+
+from meshcore_hub.common.models import EventObserver, Message, Node, NodeTag
 
 
 class TestListMessages:
@@ -44,6 +46,96 @@ class TestListMessages:
         assert data["limit"] == 25
         assert data["offset"] == 10
 
+    def test_list_messages_sender_name_resolution(self, client_no_auth, api_db_session):
+        """Messages resolve sender name from matching pubkey_prefix."""
+        sender_node = Node(
+            public_key="abc123def456abc123def456abc123de",
+            name="SenderNode",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add(sender_node)
+        api_db_session.commit()
+
+        msg = Message(
+            message_type="contact",
+            pubkey_prefix="abc123def456",
+            text="Hello from sender",
+            received_at=datetime.now(timezone.utc),
+        )
+        api_db_session.add(msg)
+        api_db_session.commit()
+
+        response = client_no_auth.get("/api/v1/messages")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["sender_name"] == "SenderNode"
+
+    def test_list_messages_sender_tag_name_resolution(
+        self, client_no_auth, api_db_session
+    ):
+        """Messages resolve sender tag name from name tags."""
+        sender_node = Node(
+            public_key="tag123tag123tag123tag123tag123ta",
+            name="OriginalName",
+            first_seen=datetime.now(timezone.utc),
+        )
+        api_db_session.add(sender_node)
+        api_db_session.commit()
+
+        tag = NodeTag(
+            node_id=sender_node.id,
+            key="name",
+            value="TagSenderName",
+        )
+        api_db_session.add(tag)
+        api_db_session.commit()
+
+        msg = Message(
+            message_type="contact",
+            pubkey_prefix="tag123tag123",
+            text="Hello with tag",
+            received_at=datetime.now(timezone.utc),
+        )
+        api_db_session.add(msg)
+        api_db_session.commit()
+
+        response = client_no_auth.get("/api/v1/messages")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert data["items"][0]["sender_tag_name"] == "TagSenderName"
+
+    def test_list_messages_with_observers(
+        self, client_no_auth, api_db_session, receiver_node
+    ):
+        """Messages include observers list in response."""
+        msg = Message(
+            message_type="channel",
+            channel_idx=1,
+            text="Msg with observer",
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=receiver_node.id,
+        )
+        api_db_session.add(msg)
+        api_db_session.commit()
+
+        if msg.event_hash:
+            observer = EventObserver(
+                event_type="message",
+                event_hash=msg.event_hash,
+                observer_node_id=receiver_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+            api_db_session.add(observer)
+            api_db_session.commit()
+
+        response = client_no_auth.get("/api/v1/messages")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["items"]) == 1
+        assert "observers" in data["items"][0]
+
 
 class TestGetMessage:
     """Tests for GET /messages/{id} endpoint."""
@@ -60,6 +152,35 @@ class TestGetMessage:
         response = client_no_auth.get("/api/v1/messages/nonexistent-id")
         assert response.status_code == 404
 
+    def test_get_message_with_observers(
+        self, client_no_auth, api_db_session, receiver_node
+    ):
+        """Get message includes observers list."""
+        msg = Message(
+            message_type="channel",
+            channel_idx=1,
+            text="Msg for get observer test",
+            received_at=datetime.now(timezone.utc),
+            observer_node_id=receiver_node.id,
+        )
+        api_db_session.add(msg)
+        api_db_session.commit()
+
+        if msg.event_hash:
+            observer = EventObserver(
+                event_type="message",
+                event_hash=msg.event_hash,
+                observer_node_id=receiver_node.id,
+                observed_at=datetime.now(timezone.utc),
+            )
+            api_db_session.add(observer)
+            api_db_session.commit()
+
+        response = client_no_auth.get(f"/api/v1/messages/{msg.id}")
+        assert response.status_code == 200
+        data = response.json()
+        assert "observers" in data
+
 
 class TestListMessagesFilters:
     """Tests for message list query filters."""
@@ -113,10 +234,6 @@ class TestListMessagesFilters:
 
     def test_filter_by_since(self, client_no_auth, api_db_session):
         """Test filtering messages by since timestamp."""
-        from datetime import timedelta
-
-        from meshcore_hub.common.models import Message
-
         now = datetime.now(timezone.utc)
         old_time = now - timedelta(days=7)
 
@@ -139,10 +256,6 @@ class TestListMessagesFilters:
 
     def test_filter_by_until(self, client_no_auth, api_db_session):
         """Test filtering messages by until timestamp."""
-        from datetime import timedelta
-
-        from meshcore_hub.common.models import Message
-
         now = datetime.now(timezone.utc)
         old_time = now - timedelta(days=7)
 

tests/test_collector/test_cli.py

@@ -0,0 +1,110 @@
+"""Tests for collector CLI commands."""
+
+from unittest.mock import MagicMock, patch
+
+from click.testing import CliRunner
+
+from meshcore_hub.collector.cli import collector
+
+
+class TestCollectorGroup:
+    """Tests for the collector group command."""
+
+    def test_collector_without_subcommand_calls_run_service(self):
+        """Invoking collector without subcommand calls _run_collector_service."""
+        runner = CliRunner()
+        mock_settings = MagicMock(
+            data_home="/tmp/data",
+            effective_seed_home="/tmp/seed",
+            effective_database_url="sqlite:///tmp/test.db",
+        )
+        mock_settings.model_copy.return_value = mock_settings
+
+        with (
+            patch(
+                "meshcore_hub.common.config.get_collector_settings",
+                return_value=mock_settings,
+            ),
+            patch("meshcore_hub.collector.cli._run_collector_service") as mock_run,
+        ):
+            result = runner.invoke(
+                collector, ["--mqtt-host", "testhost"], catch_exceptions=False
+            )
+
+        assert result.exit_code == 0
+        mock_run.assert_called_once()
+
+    def test_collector_with_data_home_override(self):
+        """--data-home overrides the default data home."""
+        runner = CliRunner()
+        mock_settings = MagicMock(
+            data_home="/default",
+            effective_seed_home="/default/seed",
+            effective_database_url="sqlite:///default/db",
+        )
+        mock_settings.model_copy.return_value = mock_settings
+
+        with (
+            patch(
+                "meshcore_hub.common.config.get_collector_settings",
+                return_value=mock_settings,
+            ),
+            patch("meshcore_hub.collector.cli._run_collector_service"),
+        ):
+            result = runner.invoke(
+                collector, ["--data-home", "/custom/data"], catch_exceptions=False
+            )
+
+        assert result.exit_code == 0
+        mock_settings.model_copy.assert_called_once_with(
+            update={"data_home": "/custom/data"}
+        )
+
+
+class TestCollectorRunSubcommand:
+    """Tests for the 'collector run' subcommand."""
+
+    def test_run_subcommand_calls_run_service(self):
+        """'collector run' delegates to _run_collector_service."""
+        runner = CliRunner()
+        mock_settings = MagicMock(
+            data_home="/tmp/data",
+            effective_seed_home="/tmp/seed",
+            effective_database_url="sqlite:///tmp/test.db",
+        )
+        mock_settings.model_copy.return_value = mock_settings
+
+        with (
+            patch(
+                "meshcore_hub.common.config.get_collector_settings",
+                return_value=mock_settings,
+            ),
+            patch("meshcore_hub.collector.cli._run_collector_service") as mock_run,
+        ):
+            result = runner.invoke(collector, ["run"], catch_exceptions=False)
+
+        assert result.exit_code == 0
+        mock_run.assert_called_once()
+
+
+class TestCollectorSeedSubcommand:
+    """Tests for the 'collector seed' subcommand."""
+
+    def test_seed_command_help(self):
+        """'collector seed --help' shows usage."""
+        runner = CliRunner()
+        mock_settings = MagicMock(
+            data_home="/tmp/data",
+            effective_seed_home="/tmp/seed",
+            effective_database_url="sqlite:///tmp/test.db",
+        )
+        mock_settings.model_copy.return_value = mock_settings
+
+        with patch(
+            "meshcore_hub.common.config.get_collector_settings",
+            return_value=mock_settings,
+        ):
+            result = runner.invoke(collector, ["seed", "--help"])
+
+        assert result.exit_code == 0
+        assert "seed" in result.output.lower() or "import" in result.output.lower()

tests/test_collector/test_handlers/test_advertisement.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy import select
 
-from meshcore_hub.common.models import Advertisement, Node
+from meshcore_hub.common.models import Advertisement, EventObserver, Node
 from meshcore_hub.collector.handlers.advertisement import handle_advertisement
 
 
@@ -104,3 +104,98 @@ class TestHandleAdvertisement:
         # No advertisement should be created
         ads = db_session.execute(select(Advertisement)).scalars().all()
         assert len(ads) == 0
+
+    def test_duplicate_adds_observer(self, db_manager, db_session):
+        """Duplicate advertisement adds receiver to observers instead of new record."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+
+        receiver_pk = "c" * 64
+        handle_advertisement(receiver_pk, "advertisement", payload, db_manager)
+
+        ads = db_session.execute(select(Advertisement)).scalars().all()
+        assert len(ads) == 1
+
+        second_receiver_pk = "d" * 64
+        handle_advertisement(second_receiver_pk, "advertisement", payload, db_manager)
+
+        ads = db_session.execute(select(Advertisement)).scalars().all()
+        assert len(ads) == 1
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 2
+
+    def test_duplicate_updates_node_location(self, db_manager, db_session):
+        """Duplicate ad still updates advertised node lat/lon."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        payload_with_loc = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+            "lat": 10.0,
+            "lon": 20.0,
+        }
+        handle_advertisement("b" * 64, "advertisement", payload_with_loc, db_manager)
+
+        node = db_session.execute(
+            select(Node).where(Node.public_key == "a" * 64)
+        ).scalar_one()
+        assert node.lat == 10.0
+        assert node.lon == 20.0
+
+    def test_location_from_nested_dict(self, db_manager, db_session):
+        """Location extracted from nested location dict."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+            "location": {"latitude": 51.5, "longitude": -0.1},
+        }
+
+        handle_advertisement("b" * 64, "advertisement", payload, db_manager)
+
+        node = db_session.execute(
+            select(Node).where(Node.public_key == "a" * 64)
+        ).scalar_one()
+        assert node.lat == 51.5
+        assert node.lon == -0.1
+
+    def test_duplicate_same_receiver_skips_observer(self, db_manager, db_session):
+        """Duplicate ad from same receiver does not add duplicate observer."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+
+        receiver_pk = "c" * 64
+        handle_advertisement(receiver_pk, "advertisement", payload, db_manager)
+        handle_advertisement(receiver_pk, "advertisement", payload, db_manager)
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 1
+
+    def test_creates_receiver_node(self, db_manager, db_session):
+        """Receiver node is created if it does not exist."""
+        payload = {
+            "public_key": "a" * 64,
+            "name": "TestNode",
+            "adv_type": "chat",
+        }
+
+        receiver_pk = "e" * 64
+        handle_advertisement(receiver_pk, "advertisement", payload, db_manager)
+
+        receiver = db_session.execute(
+            select(Node).where(Node.public_key == receiver_pk)
+        ).scalar_one_or_none()
+        assert receiver is not None

tests/test_collector/test_handlers/test_message.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy import select
 
-from meshcore_hub.common.models import Message, Node
+from meshcore_hub.common.models import EventObserver, Message, Node
 from meshcore_hub.collector.handlers.message import (
     handle_contact_message,
     handle_channel_message,
@@ -46,6 +46,49 @@ class TestHandleContactMessage:
         msgs = db_session.execute(select(Message)).scalars().all()
         assert len(msgs) == 0
 
+    def test_duplicate_contact_message_adds_observer(self, db_manager, db_session):
+        """Duplicate contact message adds receiver to observers."""
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "text": "Hello!",
+        }
+
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+        handle_contact_message("b" * 64, "contact_msg_recv", payload, db_manager)
+
+        msgs = db_session.execute(select(Message)).scalars().all()
+        assert len(msgs) == 1
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 2
+
+    def test_contact_message_sender_timestamp(self, db_manager, db_session):
+        """Sender timestamp is parsed and stored."""
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "text": "Hello!",
+            "sender_timestamp": 1771695860,
+        }
+
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+
+        msg = db_session.execute(select(Message)).scalar_one()
+        assert msg.sender_timestamp is not None
+        assert msg.sender_timestamp.year == 2026
+
+    def test_contact_message_snr_lowercase_key(self, db_manager, db_session):
+        """SNR is read from lowercase 'snr' key when uppercase is absent."""
+        payload = {
+            "pubkey_prefix": "01ab2186c4d5",
+            "text": "Hello!",
+            "snr": 7.5,
+        }
+
+        handle_contact_message("a" * 64, "contact_msg_recv", payload, db_manager)
+
+        msg = db_session.execute(select(Message)).scalar_one()
+        assert msg.snr == 7.5
+
 
 class TestHandleChannelMessage:
     """Tests for handle_channel_message."""
@@ -86,3 +129,32 @@ class TestHandleChannelMessage:
         ).scalar_one_or_none()
 
         assert node is not None
+
+    def test_duplicate_channel_message_adds_observer(self, db_manager, db_session):
+        """Duplicate channel message adds receiver to observers."""
+        payload = {
+            "channel_idx": 4,
+            "text": "Channel msg",
+        }
+
+        handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager)
+        handle_channel_message("b" * 64, "channel_msg_recv", payload, db_manager)
+
+        msgs = db_session.execute(select(Message)).scalars().all()
+        assert len(msgs) == 1
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 2
+
+    def test_channel_message_signature_stored(self, db_manager, db_session):
+        """Signature field is stored when provided."""
+        payload = {
+            "channel_idx": 4,
+            "text": "Signed msg",
+            "signature": "abcdef1234567890",
+        }
+
+        handle_channel_message("a" * 64, "channel_msg_recv", payload, db_manager)
+
+        msg = db_session.execute(select(Message)).scalar_one()
+        assert msg.signature == "abcdef1234567890"

tests/test_collector/test_handlers/test_telemetry.py

@@ -2,7 +2,7 @@
 
 from sqlalchemy import select
 
-from meshcore_hub.common.models import Node, Telemetry
+from meshcore_hub.common.models import EventObserver, Node, Telemetry
 from meshcore_hub.collector.handlers.telemetry import handle_telemetry
 
 
@@ -58,3 +58,71 @@ class TestHandleTelemetry:
         # No telemetry should be created
         records = db_session.execute(select(Telemetry)).scalars().all()
         assert len(records) == 0
+
+    def test_duplicate_telemetry_adds_observer(self, db_manager, db_session):
+        """Duplicate telemetry adds receiver to observers."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {"temperature": 20.0},
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+        handle_telemetry("c" * 64, "telemetry_response", payload, db_manager)
+
+        records = db_session.execute(select(Telemetry)).scalars().all()
+        assert len(records) == 1
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 2
+
+    def test_lpp_data_as_hex_string(self, db_manager, db_session):
+        """lpp_data as hex string is converted to bytes."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {"battery": 85},
+            "lpp_data": "deadbeef",
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        telemetry = db_session.execute(select(Telemetry)).scalar_one()
+        assert telemetry.lpp_data == bytes.fromhex("deadbeef")
+
+    def test_lpp_data_as_list(self, db_manager, db_session):
+        """lpp_data as list of ints is converted to bytes."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {"battery": 85},
+            "lpp_data": [1, 2, 3],
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        telemetry = db_session.execute(select(Telemetry)).scalar_one()
+        assert telemetry.lpp_data == b"\x01\x02\x03"
+
+    def test_creates_receiver_node(self, db_manager, db_session):
+        """Receiver node is created if it does not exist."""
+        payload = {
+            "node_public_key": "b" * 64,
+            "parsed_data": {"temperature": 20.0},
+        }
+
+        receiver_pk = "e" * 64
+        handle_telemetry(receiver_pk, "telemetry_response", payload, db_manager)
+
+        receiver = db_session.execute(
+            select(Node).where(Node.public_key == receiver_pk)
+        ).scalar_one_or_none()
+        assert receiver is not None
+
+    def test_no_parsed_data_still_creates_record(self, db_manager, db_session):
+        """Telemetry without parsed_data is still stored."""
+        payload = {
+            "node_public_key": "b" * 64,
+        }
+
+        handle_telemetry("a" * 64, "telemetry_response", payload, db_manager)
+
+        telemetry = db_session.execute(select(Telemetry)).scalar_one()
+        assert telemetry.node_public_key == "b" * 64

tests/test_collector/test_handlers/test_trace.py

@@ -0,0 +1,90 @@
+"""Tests for trace data handler."""
+
+from sqlalchemy import select
+
+from meshcore_hub.common.models import EventObserver, Node, TracePath
+from meshcore_hub.collector.handlers.trace import handle_trace_data
+
+
+class TestHandleTraceData:
+    """Tests for handle_trace_data."""
+
+    def test_creates_trace_path_record(self, db_manager, db_session):
+        """Test that trace path records are stored."""
+        payload = {
+            "initiator_tag": 12345,
+            "path_len": 3,
+            "flags": 0,
+            "path_hashes": ["AA", "BB", "CC"],
+            "snr_values": [10.5, 8.0, 6.25],
+            "hop_count": 3,
+        }
+
+        handle_trace_data("a" * 64, "trace_data", payload, db_manager)
+
+        trace = db_session.execute(select(TracePath)).scalar_one_or_none()
+        assert trace is not None
+        assert trace.initiator_tag == 12345
+        assert trace.hop_count == 3
+        assert trace.path_hashes == ["AA", "BB", "CC"]
+        assert trace.snr_values == [10.5, 8.0, 6.25]
+
+    def test_handles_missing_initiator_tag(self, db_manager, db_session):
+        """Test that missing initiator_tag is handled gracefully."""
+        payload = {
+            "path_hashes": ["AA"],
+        }
+
+        handle_trace_data("a" * 64, "trace_data", payload, db_manager)
+
+        traces = db_session.execute(select(TracePath)).scalars().all()
+        assert len(traces) == 0
+
+    def test_duplicate_trace_adds_observer(self, db_manager, db_session):
+        """Duplicate trace adds receiver to observers instead of new record."""
+        payload = {
+            "initiator_tag": 99999,
+            "path_hashes": ["AA", "BB"],
+            "hop_count": 2,
+        }
+
+        handle_trace_data("a" * 64, "trace_data", payload, db_manager)
+
+        traces = db_session.execute(select(TracePath)).scalars().all()
+        assert len(traces) == 1
+
+        handle_trace_data("b" * 64, "trace_data", payload, db_manager)
+
+        traces = db_session.execute(select(TracePath)).scalars().all()
+        assert len(traces) == 1
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 2
+
+    def test_creates_receiver_node(self, db_manager, db_session):
+        """Receiver node is created if it does not exist."""
+        payload = {
+            "initiator_tag": 55555,
+            "hop_count": 1,
+        }
+
+        receiver_pk = "e" * 64
+        handle_trace_data(receiver_pk, "trace_data", payload, db_manager)
+
+        receiver = db_session.execute(
+            select(Node).where(Node.public_key == receiver_pk)
+        ).scalar_one_or_none()
+        assert receiver is not None
+
+    def test_creates_first_observer(self, db_manager, db_session):
+        """First trace event creates an observer entry."""
+        payload = {
+            "initiator_tag": 77777,
+            "hop_count": 2,
+        }
+
+        handle_trace_data("a" * 64, "trace_data", payload, db_manager)
+
+        observers = db_session.execute(select(EventObserver)).scalars().all()
+        assert len(observers) == 1
+        assert observers[0].event_type == "trace"

tests/test_collector/test_subscriber.py

@@ -850,6 +850,126 @@ class TestSubscriber:
         assert payload["pubkey_prefix"] == "1A2B3C4D5E6F"
 
 
+class TestSubscriberDispatch:
+    """Tests for _dispatch_event and lifecycle methods."""
+
+    @pytest.fixture
+    def mock_mqtt_client(self):
+        """Create a mock MQTT client."""
+        client = MagicMock()
+        client.topic_builder = MagicMock()
+        client.topic_builder.prefix = "meshcore"
+        client.topic_builder.all_events_topic.return_value = "meshcore/+/event/#"
+        client.topic_builder.parse_letsmesh_upload_topic.return_value = (
+            "a" * 64,
+            "status",
+        )
+        return client
+
+    @pytest.fixture
+    def subscriber(self, mock_mqtt_client, db_manager):
+        """Create a subscriber instance."""
+        return Subscriber(mock_mqtt_client, db_manager)
+
+    def test_dispatch_event_with_no_handler_falls_back_to_event_log(self, subscriber):
+        """Unregistered event types fall back to event_log handler."""
+        with patch(
+            "meshcore_hub.collector.handlers.event_log.handle_event_log"
+        ) as mock_log:
+            subscriber._dispatch_event("a" * 64, "unknown_type", {"data": 1})
+            mock_log.assert_called_once()
+
+    def test_dispatch_event_handler_exception_logged(self, subscriber):
+        """Handler exceptions are caught and logged, not re-raised."""
+        handler = MagicMock(side_effect=RuntimeError("boom"))
+        subscriber.register_handler("test_event", handler)
+
+        subscriber._dispatch_event("a" * 64, "test_event", {"data": 1})
+
+        handler.assert_called_once()
+
+    def test_dispatch_event_event_log_exception_logged(self, subscriber):
+        """Event log handler exceptions are caught and logged."""
+        with patch(
+            "meshcore_hub.collector.handlers.event_log.handle_event_log",
+            side_effect=RuntimeError("log boom"),
+        ):
+            subscriber._dispatch_event("a" * 64, "unknown_type", {"data": 1})
+
+    def test_dispatch_event_queues_webhook(self, subscriber):
+        """Events are queued for webhook when dispatcher is configured."""
+        mock_dispatcher = MagicMock()
+        mock_dispatcher.webhooks = [MagicMock()]
+        subscriber._webhook_dispatcher = mock_dispatcher
+
+        handler = MagicMock()
+        subscriber.register_handler("test_event", handler)
+
+        subscriber._dispatch_event("a" * 64, "test_event", {"data": 1})
+
+        assert len(subscriber._webhook_queue) == 1
+        assert subscriber._webhook_queue[0][0] == "test_event"
+
+    def test_dispatch_event_no_webhook_without_dispatcher(self, subscriber):
+        """No webhook queued when dispatcher is not configured."""
+        handler = MagicMock()
+        subscriber.register_handler("test_event", handler)
+
+        subscriber._dispatch_event("a" * 64, "test_event", {"data": 1})
+
+        assert len(subscriber._webhook_queue) == 0
+
+    def test_start_with_mqtt_retry(self, mock_mqtt_client, db_manager):
+        """MQTT connection is retried on failure."""
+        mock_mqtt_client.connect.side_effect = [
+            ConnectionError("fail"),
+            None,
+        ]
+
+        subscriber = Subscriber(mock_mqtt_client, db_manager)
+        with patch("meshcore_hub.collector.subscriber.time.sleep"):
+            subscriber.start()
+
+        assert mock_mqtt_client.connect.call_count == 2
+        assert subscriber._mqtt_connected is True
+        subscriber.stop()
+
+    def test_start_mqtt_all_retries_exhausted(self, mock_mqtt_client, db_manager):
+        """Subscriber raises when all MQTT retries fail."""
+        mock_mqtt_client.connect.side_effect = ConnectionError("fail")
+
+        subscriber = Subscriber(mock_mqtt_client, db_manager)
+        with (
+            patch("meshcore_hub.collector.subscriber.time.sleep"),
+            pytest.raises(ConnectionError),
+        ):
+            subscriber.start()
+
+    def test_run_calls_start_when_not_running(self, mock_mqtt_client, db_manager):
+        """run() calls start() if subscriber is not running."""
+        subscriber = Subscriber(mock_mqtt_client, db_manager)
+
+        with patch.object(subscriber, "start") as mock_start:
+            mock_start.side_effect = lambda: setattr(subscriber, "_running", True)
+            subscriber._shutdown_event.set()
+            subscriber.run()
+
+        mock_start.assert_called_once()
+
+    def test_stop_when_not_running(self, subscriber):
+        """stop() is a no-op when not running."""
+        subscriber._running = False
+        subscriber.stop()
+
+    def test_health_status(self, subscriber):
+        """Health status reports correct state."""
+        status = subscriber.get_health_status()
+        assert status["running"] is False
+        assert status["mqtt_connected"] is False
+        assert status["database_connected"] is False
+        assert status["healthy"] is False
+
+
 class TestCreateSubscriber:
     """Tests for create_subscriber factory function."""