From f7b4df13a75148fc68cece1587436aae04bcd511 Mon Sep 17 00:00:00 2001 From: Louis King Date: Mon, 12 Jan 2026 21:00:02 +0000 Subject: [PATCH] Added more test coverage --- tests/test_api/conftest.py | 145 +++++++++++++ tests/test_api/test_advertisements.py | 119 +++++++++++ tests/test_api/test_members.py | 285 ++++++++++++++++++++++++++ tests/test_api/test_messages.py | 126 ++++++++++++ tests/test_api/test_nodes.py | 60 ++++++ tests/test_api/test_telemetry.py | 67 ++++++ tests/test_api/test_trace_paths.py | 69 +++++++ 7 files changed, 871 insertions(+) create mode 100644 tests/test_api/test_members.py diff --git a/tests/test_api/conftest.py b/tests/test_api/conftest.py index 0e6483a..0529695 100644 --- a/tests/test_api/conftest.py +++ b/tests/test_api/conftest.py @@ -20,6 +20,7 @@ from meshcore_hub.common.database import DatabaseManager from meshcore_hub.common.models import ( Advertisement, Base, + Member, Message, Node, NodeTag, @@ -264,3 +265,147 @@ def sample_trace_path(api_db_session): api_db_session.commit() api_db_session.refresh(trace) return trace + + +@pytest.fixture +def sample_member(api_db_session): + """Create a sample member in the database.""" + member = Member( + member_id="alice", + name="Alice Smith", + callsign="W1ABC", + role="Admin", + description="Network administrator", + contact="alice@example.com", + ) + api_db_session.add(member) + api_db_session.commit() + api_db_session.refresh(member) + return member + + +@pytest.fixture +def receiver_node(api_db_session): + """Create a receiver node in the database.""" + node = Node( + public_key="receiver123receiver123receiver12", + name="Receiver Node", + adv_type="REPEATER", + first_seen=datetime.now(timezone.utc), + last_seen=datetime.now(timezone.utc), + ) + api_db_session.add(node) + api_db_session.commit() + api_db_session.refresh(node) + return node + + +@pytest.fixture +def sample_message_with_receiver(api_db_session, receiver_node): + """Create a message with a receiver node.""" + message = Message( + message_type="channel", + channel_idx=1, + pubkey_prefix="xyz789", + text="Channel message with receiver", + received_at=datetime.now(timezone.utc), + receiver_node_id=receiver_node.id, + ) + api_db_session.add(message) + api_db_session.commit() + api_db_session.refresh(message) + return message + + +@pytest.fixture +def sample_advertisement_with_receiver(api_db_session, sample_node, receiver_node): + """Create an advertisement with source and receiver nodes.""" + advert = Advertisement( + public_key=sample_node.public_key, + name="SourceNode", + adv_type="REPEATER", + received_at=datetime.now(timezone.utc), + node_id=sample_node.id, + receiver_node_id=receiver_node.id, + ) + api_db_session.add(advert) + api_db_session.commit() + api_db_session.refresh(advert) + return advert + + +@pytest.fixture +def sample_telemetry_with_receiver(api_db_session, receiver_node): + """Create a telemetry record with a receiver node.""" + telemetry = Telemetry( + node_public_key="xyz789xyz789xyz789xyz789xyz789xy", + parsed_data={"battery_level": 50.0}, + received_at=datetime.now(timezone.utc), + receiver_node_id=receiver_node.id, + ) + api_db_session.add(telemetry) + api_db_session.commit() + api_db_session.refresh(telemetry) + return telemetry + + +@pytest.fixture +def sample_trace_path_with_receiver(api_db_session, receiver_node): + """Create a trace path with a receiver node.""" + trace = TracePath( + initiator_tag=99999, + path_hashes=["aaa111", "bbb222"], + hop_count=2, + received_at=datetime.now(timezone.utc), + receiver_node_id=receiver_node.id, + ) + api_db_session.add(trace) + api_db_session.commit() + api_db_session.refresh(trace) + return trace + + +@pytest.fixture +def sample_node_with_name_tag(api_db_session): + """Create a node with a name tag for search testing.""" + node = Node( + public_key="searchable123searchable123searc", + name="Original Name", + adv_type="CLIENT", + first_seen=datetime.now(timezone.utc), + ) + api_db_session.add(node) + api_db_session.commit() + + tag = NodeTag( + node_id=node.id, + key="name", + value="Friendly Search Name", + ) + api_db_session.add(tag) + api_db_session.commit() + api_db_session.refresh(node) + return node + + +@pytest.fixture +def sample_node_with_member_tag(api_db_session): + """Create a node with a member_id tag for filter testing.""" + node = Node( + public_key="member123member123member123membe", + name="Member Node", + adv_type="CHAT", + first_seen=datetime.now(timezone.utc), + ) + api_db_session.add(node) + api_db_session.commit() + + tag = NodeTag( + node_id=node.id, + key="member_id", + value="alice", + ) + api_db_session.add(tag) + api_db_session.commit() + api_db_session.refresh(node) + return node diff --git a/tests/test_api/test_advertisements.py b/tests/test_api/test_advertisements.py index 89aaca2..eea0b3b 100644 --- a/tests/test_api/test_advertisements.py +++ b/tests/test_api/test_advertisements.py @@ -1,5 +1,7 @@ """Tests for advertisement API routes.""" +from datetime import datetime, timedelta, timezone + class TestListAdvertisements: """Tests for GET /advertisements endpoint.""" @@ -55,3 +57,120 @@ class TestGetAdvertisement: """Test getting a non-existent advertisement.""" response = client_no_auth.get("/api/v1/advertisements/nonexistent-id") assert response.status_code == 404 + + +class TestListAdvertisementsFilters: + """Tests for advertisement list query filters.""" + + def test_filter_by_search_public_key(self, client_no_auth, sample_advertisement): + """Test filtering advertisements by public key search.""" + # Partial public key match + response = client_no_auth.get("/api/v1/advertisements?search=abc123") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/advertisements?search=zzz999") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_search_name(self, client_no_auth, sample_advertisement): + """Test filtering advertisements by name search.""" + response = client_no_auth.get("/api/v1/advertisements?search=TestNode") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_received_by( + self, + client_no_auth, + sample_advertisement, + sample_advertisement_with_receiver, + receiver_node, + ): + """Test filtering advertisements by receiver node.""" + response = client_no_auth.get( + f"/api/v1/advertisements?received_by={receiver_node.public_key}" + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_member_id( + 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, + name="Member Node Ad", + adv_type="CHAT", + received_at=datetime.now(timezone.utc), + node_id=sample_node_with_member_tag.id, + ) + api_db_session.add(advert) + api_db_session.commit() + + # Filter by member_id + response = client_no_auth.get("/api/v1/advertisements?member_id=alice") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/advertisements?member_id=unknown") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + 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) + + # Create an old advertisement + old_advert = Advertisement( + public_key="old123old123old123old123old123ol", + name="Old Advertisement", + adv_type="CLIENT", + received_at=old_time, + ) + api_db_session.add(old_advert) + api_db_session.commit() + + # Filter since yesterday - should not include old advertisement + since = (now - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/advertisements?since={since}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + 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) + + # Create an old advertisement + old_advert = Advertisement( + public_key="until123until123until123until12", + name="Old Advertisement Until", + adv_type="CLIENT", + received_at=old_time, + ) + api_db_session.add(old_advert) + api_db_session.commit() + + # Filter until 5 days ago - should include old advertisement + until = (now - timedelta(days=5)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/advertisements?until={until}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 diff --git a/tests/test_api/test_members.py b/tests/test_api/test_members.py new file mode 100644 index 0000000..507824e --- /dev/null +++ b/tests/test_api/test_members.py @@ -0,0 +1,285 @@ +"""Tests for member API routes.""" + + +class TestListMembers: + """Tests for GET /members endpoint.""" + + def test_list_members_empty(self, client_no_auth): + """Test listing members when database is empty.""" + response = client_no_auth.get("/api/v1/members") + assert response.status_code == 200 + data = response.json() + assert data["items"] == [] + assert data["total"] == 0 + + def test_list_members_with_data(self, client_no_auth, sample_member): + """Test listing members with data in database.""" + response = client_no_auth.get("/api/v1/members") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["total"] == 1 + assert data["items"][0]["member_id"] == sample_member.member_id + assert data["items"][0]["name"] == sample_member.name + assert data["items"][0]["callsign"] == sample_member.callsign + + def test_list_members_pagination(self, client_no_auth, sample_member): + """Test member list pagination parameters.""" + response = client_no_auth.get("/api/v1/members?limit=25&offset=10") + assert response.status_code == 200 + data = response.json() + assert data["limit"] == 25 + assert data["offset"] == 10 + + def test_list_members_requires_read_auth(self, client_with_auth): + """Test listing members requires read auth when configured.""" + # Without auth header + response = client_with_auth.get("/api/v1/members") + assert response.status_code == 401 + + # With read key + response = client_with_auth.get( + "/api/v1/members", + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 200 + + +class TestGetMember: + """Tests for GET /members/{member_id} endpoint.""" + + def test_get_member_success(self, client_no_auth, sample_member): + """Test getting a specific member.""" + response = client_no_auth.get(f"/api/v1/members/{sample_member.id}") + assert response.status_code == 200 + data = response.json() + assert data["member_id"] == sample_member.member_id + assert data["name"] == sample_member.name + assert data["callsign"] == sample_member.callsign + assert data["role"] == sample_member.role + assert data["description"] == sample_member.description + assert data["contact"] == sample_member.contact + + def test_get_member_not_found(self, client_no_auth): + """Test getting a non-existent member.""" + response = client_no_auth.get("/api/v1/members/nonexistent-id") + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + def test_get_member_requires_read_auth(self, client_with_auth, sample_member): + """Test getting a member requires read auth when configured.""" + # Without auth header + response = client_with_auth.get(f"/api/v1/members/{sample_member.id}") + assert response.status_code == 401 + + # With read key + response = client_with_auth.get( + f"/api/v1/members/{sample_member.id}", + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 200 + + +class TestCreateMember: + """Tests for POST /members endpoint.""" + + def test_create_member_success(self, client_no_auth): + """Test creating a new member.""" + response = client_no_auth.post( + "/api/v1/members", + json={ + "member_id": "bob", + "name": "Bob Jones", + "callsign": "W2XYZ", + "role": "Member", + "description": "Regular member", + "contact": "bob@example.com", + }, + ) + assert response.status_code == 201 + data = response.json() + assert data["member_id"] == "bob" + assert data["name"] == "Bob Jones" + assert data["callsign"] == "W2XYZ" + assert data["role"] == "Member" + assert "id" in data + assert "created_at" in data + + def test_create_member_minimal(self, client_no_auth): + """Test creating a member with only required fields.""" + response = client_no_auth.post( + "/api/v1/members", + json={ + "member_id": "charlie", + "name": "Charlie Brown", + }, + ) + assert response.status_code == 201 + data = response.json() + assert data["member_id"] == "charlie" + assert data["name"] == "Charlie Brown" + assert data["callsign"] is None + assert data["role"] is None + + def test_create_member_duplicate_member_id(self, client_no_auth, sample_member): + """Test creating a member with duplicate member_id fails.""" + response = client_no_auth.post( + "/api/v1/members", + json={ + "member_id": sample_member.member_id, # "alice" already exists + "name": "Another Alice", + }, + ) + assert response.status_code == 400 + assert "already exists" in response.json()["detail"].lower() + + def test_create_member_requires_admin_auth(self, client_with_auth): + """Test creating a member requires admin auth.""" + # Without auth + response = client_with_auth.post( + "/api/v1/members", + json={"member_id": "test", "name": "Test User"}, + ) + assert response.status_code == 401 + + # With read key (not admin) + response = client_with_auth.post( + "/api/v1/members", + json={"member_id": "test", "name": "Test User"}, + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 403 + + # With admin key + response = client_with_auth.post( + "/api/v1/members", + json={"member_id": "test", "name": "Test User"}, + headers={"Authorization": "Bearer test-admin-key"}, + ) + assert response.status_code == 201 + + +class TestUpdateMember: + """Tests for PUT /members/{member_id} endpoint.""" + + def test_update_member_success(self, client_no_auth, sample_member): + """Test updating a member.""" + response = client_no_auth.put( + f"/api/v1/members/{sample_member.id}", + json={ + "name": "Alice Johnson", + "role": "Super Admin", + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["name"] == "Alice Johnson" + assert data["role"] == "Super Admin" + # Unchanged fields should remain + assert data["member_id"] == sample_member.member_id + assert data["callsign"] == sample_member.callsign + + def test_update_member_change_member_id(self, client_no_auth, sample_member): + """Test updating member_id.""" + response = client_no_auth.put( + f"/api/v1/members/{sample_member.id}", + json={"member_id": "alice2"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["member_id"] == "alice2" + + def test_update_member_member_id_collision( + self, client_no_auth, api_db_session, sample_member + ): + """Test updating member_id to one that already exists fails.""" + from meshcore_hub.common.models import Member + + # Create another member + other_member = Member( + member_id="bob", + name="Bob", + ) + api_db_session.add(other_member) + api_db_session.commit() + + # Try to change alice's member_id to "bob" + response = client_no_auth.put( + f"/api/v1/members/{sample_member.id}", + json={"member_id": "bob"}, + ) + assert response.status_code == 400 + assert "already exists" in response.json()["detail"].lower() + + def test_update_member_not_found(self, client_no_auth): + """Test updating a non-existent member.""" + response = client_no_auth.put( + "/api/v1/members/nonexistent-id", + json={"name": "New Name"}, + ) + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + def test_update_member_requires_admin_auth(self, client_with_auth, sample_member): + """Test updating a member requires admin auth.""" + # Without auth + response = client_with_auth.put( + f"/api/v1/members/{sample_member.id}", + json={"name": "New Name"}, + ) + assert response.status_code == 401 + + # With read key (not admin) + response = client_with_auth.put( + f"/api/v1/members/{sample_member.id}", + json={"name": "New Name"}, + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 403 + + # With admin key + response = client_with_auth.put( + f"/api/v1/members/{sample_member.id}", + json={"name": "New Name"}, + headers={"Authorization": "Bearer test-admin-key"}, + ) + assert response.status_code == 200 + + +class TestDeleteMember: + """Tests for DELETE /members/{member_id} endpoint.""" + + def test_delete_member_success(self, client_no_auth, sample_member): + """Test deleting a member.""" + response = client_no_auth.delete(f"/api/v1/members/{sample_member.id}") + assert response.status_code == 204 + + # Verify it's deleted + response = client_no_auth.get(f"/api/v1/members/{sample_member.id}") + assert response.status_code == 404 + + def test_delete_member_not_found(self, client_no_auth): + """Test deleting a non-existent member.""" + response = client_no_auth.delete("/api/v1/members/nonexistent-id") + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + def test_delete_member_requires_admin_auth(self, client_with_auth, sample_member): + """Test deleting a member requires admin auth.""" + # Without auth + response = client_with_auth.delete(f"/api/v1/members/{sample_member.id}") + assert response.status_code == 401 + + # With read key (not admin) + response = client_with_auth.delete( + f"/api/v1/members/{sample_member.id}", + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 403 + + # With admin key + response = client_with_auth.delete( + f"/api/v1/members/{sample_member.id}", + headers={"Authorization": "Bearer test-admin-key"}, + ) + assert response.status_code == 204 diff --git a/tests/test_api/test_messages.py b/tests/test_api/test_messages.py index 8924d01..ef35709 100644 --- a/tests/test_api/test_messages.py +++ b/tests/test_api/test_messages.py @@ -1,5 +1,7 @@ """Tests for message API routes.""" +from datetime import datetime, timezone + class TestListMessages: """Tests for GET /messages endpoint.""" @@ -57,3 +59,127 @@ class TestGetMessage: """Test getting a non-existent message.""" response = client_no_auth.get("/api/v1/messages/nonexistent-id") assert response.status_code == 404 + + +class TestListMessagesFilters: + """Tests for message list query filters.""" + + def test_filter_by_pubkey_prefix(self, client_no_auth, sample_message): + """Test filtering messages by pubkey_prefix.""" + # Match + response = client_no_auth.get("/api/v1/messages?pubkey_prefix=abc123") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/messages?pubkey_prefix=xyz999") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_channel_idx( + self, client_no_auth, sample_message, sample_message_with_receiver + ): + """Test filtering messages by channel_idx.""" + # Channel 1 should match sample_message_with_receiver + response = client_no_auth.get("/api/v1/messages?channel_idx=1") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["items"][0]["channel_idx"] == 1 + + # Channel 0 should return no results + response = client_no_auth.get("/api/v1/messages?channel_idx=0") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_received_by( + self, + client_no_auth, + sample_message, + sample_message_with_receiver, + receiver_node, + ): + """Test filtering messages by receiver node.""" + response = client_no_auth.get( + f"/api/v1/messages?received_by={receiver_node.public_key}" + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["items"][0]["text"] == sample_message_with_receiver.text + + 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) + + # Create an old message + old_msg = Message( + message_type="direct", + pubkey_prefix="old123", + text="Old message", + received_at=old_time, + ) + api_db_session.add(old_msg) + api_db_session.commit() + + # Filter since yesterday - should not include old message + since = (now - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/messages?since={since}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + 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) + + # Create an old message + old_msg = Message( + message_type="direct", + pubkey_prefix="old456", + text="Old message for until", + received_at=old_time, + ) + api_db_session.add(old_msg) + api_db_session.commit() + + # Filter until 5 days ago - should include old message + until = (now - timedelta(days=5)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/messages?until={until}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + assert data["items"][0]["text"] == "Old message for until" + + def test_filter_by_search(self, client_no_auth, sample_message): + """Test filtering messages by text search.""" + # Match + response = client_no_auth.get("/api/v1/messages?search=Hello") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # Case insensitive match + response = client_no_auth.get("/api/v1/messages?search=hello") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/messages?search=nonexistent") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 diff --git a/tests/test_api/test_nodes.py b/tests/test_api/test_nodes.py index 7e00deb..d4f6240 100644 --- a/tests/test_api/test_nodes.py +++ b/tests/test_api/test_nodes.py @@ -57,6 +57,66 @@ class TestListNodes: assert response.status_code == 200 +class TestListNodesFilters: + """Tests for node list query filters.""" + + def test_filter_by_search_public_key(self, client_no_auth, sample_node): + """Test filtering nodes by public key search.""" + # Partial public key match + response = client_no_auth.get("/api/v1/nodes?search=abc123") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/nodes?search=zzz999") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_search_node_name(self, client_no_auth, sample_node): + """Test filtering nodes by node name search.""" + response = client_no_auth.get("/api/v1/nodes?search=Test%20Node") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_search_name_tag(self, client_no_auth, sample_node_with_name_tag): + """Test filtering nodes by name tag search.""" + response = client_no_auth.get("/api/v1/nodes?search=Friendly%20Search") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_adv_type(self, client_no_auth, sample_node): + """Test filtering nodes by advertisement type.""" + # Match REPEATER + response = client_no_auth.get("/api/v1/nodes?adv_type=REPEATER") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/nodes?adv_type=CLIENT") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_member_id(self, client_no_auth, sample_node_with_member_tag): + """Test filtering nodes by member_id tag.""" + # Match alice + response = client_no_auth.get("/api/v1/nodes?member_id=alice") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + # No match + response = client_no_auth.get("/api/v1/nodes?member_id=unknown") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + class TestGetNode: """Tests for GET /nodes/{public_key} endpoint.""" diff --git a/tests/test_api/test_telemetry.py b/tests/test_api/test_telemetry.py index 73da843..e33e513 100644 --- a/tests/test_api/test_telemetry.py +++ b/tests/test_api/test_telemetry.py @@ -1,5 +1,7 @@ """Tests for telemetry API routes.""" +from datetime import datetime, timedelta, timezone + class TestListTelemetry: """Tests for GET /telemetry endpoint.""" @@ -51,3 +53,68 @@ class TestGetTelemetry: """Test getting a non-existent telemetry record.""" response = client_no_auth.get("/api/v1/telemetry/nonexistent-id") assert response.status_code == 404 + + +class TestListTelemetryFilters: + """Tests for telemetry list query filters.""" + + def test_filter_by_received_by( + self, + client_no_auth, + sample_telemetry, + sample_telemetry_with_receiver, + receiver_node, + ): + """Test filtering telemetry by receiver node.""" + response = client_no_auth.get( + f"/api/v1/telemetry?received_by={receiver_node.public_key}" + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_since(self, client_no_auth, api_db_session): + """Test filtering telemetry by since timestamp.""" + from meshcore_hub.common.models import Telemetry + + now = datetime.now(timezone.utc) + old_time = now - timedelta(days=7) + + # Create old telemetry + old_telemetry = Telemetry( + node_public_key="old123old123old123old123old123ol", + parsed_data={"battery_level": 10.0}, + received_at=old_time, + ) + api_db_session.add(old_telemetry) + api_db_session.commit() + + # Filter since yesterday - should not include old telemetry + since = (now - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/telemetry?since={since}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_until(self, client_no_auth, api_db_session): + """Test filtering telemetry by until timestamp.""" + from meshcore_hub.common.models import Telemetry + + now = datetime.now(timezone.utc) + old_time = now - timedelta(days=7) + + # Create old telemetry + old_telemetry = Telemetry( + node_public_key="until123until123until123until12", + parsed_data={"battery_level": 20.0}, + received_at=old_time, + ) + api_db_session.add(old_telemetry) + api_db_session.commit() + + # Filter until 5 days ago - should include old telemetry + until = (now - timedelta(days=5)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/telemetry?until={until}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 diff --git a/tests/test_api/test_trace_paths.py b/tests/test_api/test_trace_paths.py index a58e71e..f2c038f 100644 --- a/tests/test_api/test_trace_paths.py +++ b/tests/test_api/test_trace_paths.py @@ -1,5 +1,7 @@ """Tests for trace path API routes.""" +from datetime import datetime, timedelta, timezone + class TestListTracePaths: """Tests for GET /trace-paths endpoint.""" @@ -37,3 +39,70 @@ class TestGetTracePath: """Test getting a non-existent trace path.""" response = client_no_auth.get("/api/v1/trace-paths/nonexistent-id") assert response.status_code == 404 + + +class TestListTracePathsFilters: + """Tests for trace path list query filters.""" + + def test_filter_by_received_by( + self, + client_no_auth, + sample_trace_path, + sample_trace_path_with_receiver, + receiver_node, + ): + """Test filtering trace paths by receiver node.""" + response = client_no_auth.get( + f"/api/v1/trace-paths?received_by={receiver_node.public_key}" + ) + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1 + + def test_filter_by_since(self, client_no_auth, api_db_session): + """Test filtering trace paths by since timestamp.""" + from meshcore_hub.common.models import TracePath + + now = datetime.now(timezone.utc) + old_time = now - timedelta(days=7) + + # Create old trace path + old_trace = TracePath( + initiator_tag=11111, + path_hashes=["old1", "old2"], + hop_count=2, + received_at=old_time, + ) + api_db_session.add(old_trace) + api_db_session.commit() + + # Filter since yesterday - should not include old trace path + since = (now - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/trace-paths?since={since}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 0 + + def test_filter_by_until(self, client_no_auth, api_db_session): + """Test filtering trace paths by until timestamp.""" + from meshcore_hub.common.models import TracePath + + now = datetime.now(timezone.utc) + old_time = now - timedelta(days=7) + + # Create old trace path + old_trace = TracePath( + initiator_tag=22222, + path_hashes=["until1", "until2"], + hop_count=2, + received_at=old_time, + ) + api_db_session.add(old_trace) + api_db_session.commit() + + # Filter until 5 days ago - should include old trace path + until = (now - timedelta(days=5)).strftime("%Y-%m-%dT%H:%M:%S") + response = client_no_auth.get(f"/api/v1/trace-paths?until={until}") + assert response.status_code == 200 + data = response.json() + assert len(data["items"]) == 1