mirror of
https://github.com/ipnet-mesh/meshcore-hub.git
synced 2026-06-27 21:41:22 +02:00
Louis King
9af90efee4
- Add observer multi-select (<select multiple size=2>) to Advertisements and Messages filter bars, populated from /api/v1/nodes?observer=true - Make all filter sections collapsible via <details> on Nodes, Advertisements, and Messages pages; collapsed by default, auto-expands when active filters exist, preserves open state across auto-refresh ticks - Add backend observer=true|false query param to GET /api/v1/nodes for observer-only or non-observer-only node filtering via subquery - Change observed_by in Advertisements/Messages API from single public_key to list[str] with .in_() for multi-select support - Fix router.js and api.js to handle array query params (duplicate keys promoted to arrays, .append() per element) - Fix createFilterHandler to use FormData.getAll() for multi-value support - Replace DaisyUI form-control/label/label-text classes with Tailwind-native equivalents (flex flex-col gap-1, flex items-center py-1, opacity-80 text-sm) since DaisyUI CSS is tree-shaken from the build output - Thicker collapsible border (border-2 border-base-content/25) visible in both light and dark themes - Bottom-align Filter/Clear buttons via two-row form layout - Move Observer filter to last position on Advertisements page - Add filter_observer_label i18n key - Add tests for observer=true node filtering and multi observer params
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""Tests for advertisement API routes."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from meshcore_hub.common.models import Advertisement, EventObserver
|
|
|
|
|
|
class TestListAdvertisements:
|
|
"""Tests for GET /advertisements endpoint."""
|
|
|
|
def test_list_advertisements_empty(self, client_no_auth):
|
|
"""Test listing advertisements when database is empty."""
|
|
response = client_no_auth.get("/api/v1/advertisements")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["items"] == []
|
|
assert data["total"] == 0
|
|
|
|
def test_list_advertisements_with_data(self, client_no_auth, sample_advertisement):
|
|
"""Test listing advertisements with data in database."""
|
|
response = client_no_auth.get("/api/v1/advertisements")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
assert data["total"] == 1
|
|
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"
|
|
|
|
|
|
class TestGetAdvertisement:
|
|
"""Tests for GET /advertisements/{id} endpoint."""
|
|
|
|
def test_get_advertisement_success(self, client_no_auth, sample_advertisement):
|
|
"""Test getting a specific advertisement."""
|
|
response = client_no_auth.get(
|
|
f"/api/v1/advertisements/{sample_advertisement.id}"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["public_key"] == sample_advertisement.public_key
|
|
|
|
def test_get_advertisement_not_found(self, client_no_auth):
|
|
"""Test getting a non-existent advertisement."""
|
|
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."""
|
|
|
|
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_list_advertisements_filter_by_observed_by_single(
|
|
self,
|
|
client_no_auth,
|
|
sample_advertisement,
|
|
sample_advertisement_with_receiver,
|
|
receiver_node,
|
|
):
|
|
"""Test filtering advertisements by a single receiver node."""
|
|
response = client_no_auth.get(
|
|
f"/api/v1/advertisements?observed_by={receiver_node.public_key}"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
|
|
def test_list_advertisements_filter_by_observed_by_multiple(
|
|
self,
|
|
client_no_auth,
|
|
api_db_session,
|
|
receiver_node,
|
|
):
|
|
"""Test filtering advertisements by multiple receiver nodes."""
|
|
# Create second receiver node
|
|
second_receiver = receiver_node.__class__(
|
|
public_key="2nd1232nd1232nd1232nd1232nd1232n",
|
|
name="SecondObserver",
|
|
first_seen=datetime.now(timezone.utc),
|
|
)
|
|
api_db_session.add(second_receiver)
|
|
api_db_session.commit()
|
|
|
|
# Create two advertisements, each observed by a different receiver
|
|
ad1 = Advertisement(
|
|
public_key="ad1pubad1pubad1pubad1pubad1pubad",
|
|
name="AD1",
|
|
adv_type="CLIENT",
|
|
received_at=datetime.now(timezone.utc),
|
|
observer_node_id=receiver_node.id,
|
|
)
|
|
ad2 = Advertisement(
|
|
public_key="ad2pubad2pubad2pubad2pubad2pubad",
|
|
name="AD2",
|
|
adv_type="CLIENT",
|
|
received_at=datetime.now(timezone.utc),
|
|
observer_node_id=second_receiver.id,
|
|
)
|
|
api_db_session.add_all([ad1, ad2])
|
|
api_db_session.commit()
|
|
|
|
# Filter by both receivers
|
|
response = client_no_auth.get(
|
|
f"/api/v1/advertisements?observed_by={receiver_node.public_key}&observed_by={second_receiver.public_key}"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 2
|
|
|
|
# Filter by just the first receiver
|
|
response = client_no_auth.get(
|
|
f"/api/v1/advertisements?observed_by={receiver_node.public_key}"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["name"] == "AD1"
|
|
|
|
def test_filter_by_since(self, client_no_auth, api_db_session):
|
|
"""Test filtering advertisements by since timestamp."""
|
|
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."""
|
|
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
|