diff --git a/tests/test_api/test_advertisements.py b/tests/test_api/test_advertisements.py index 1b3d5d7..0a6d33d 100644 --- a/tests/test_api/test_advertisements.py +++ b/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) diff --git a/tests/test_api/test_messages.py b/tests/test_api/test_messages.py index 1ad3958..cc7736c 100644 --- a/tests/test_api/test_messages.py +++ b/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) diff --git a/tests/test_collector/test_cli.py b/tests/test_collector/test_cli.py new file mode 100644 index 0000000..4a0bd8a --- /dev/null +++ b/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() diff --git a/tests/test_collector/test_handlers/test_advertisement.py b/tests/test_collector/test_handlers/test_advertisement.py index 5f6d360..14041a3 100644 --- a/tests/test_collector/test_handlers/test_advertisement.py +++ b/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 diff --git a/tests/test_collector/test_handlers/test_message.py b/tests/test_collector/test_handlers/test_message.py index 20d5b50..afeea23 100644 --- a/tests/test_collector/test_handlers/test_message.py +++ b/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" diff --git a/tests/test_collector/test_handlers/test_telemetry.py b/tests/test_collector/test_handlers/test_telemetry.py index 60816b9..b993bca 100644 --- a/tests/test_collector/test_handlers/test_telemetry.py +++ b/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 diff --git a/tests/test_collector/test_handlers/test_trace.py b/tests/test_collector/test_handlers/test_trace.py new file mode 100644 index 0000000..2126614 --- /dev/null +++ b/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" diff --git a/tests/test_collector/test_subscriber.py b/tests/test_collector/test_subscriber.py index e25beb1..c3ad793 100644 --- a/tests/test_collector/test_subscriber.py +++ b/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."""