Compare commits

19 Commits

Author SHA1 Message Date
Louis King
fd3c3171ce Fix FastAPI response model for union return type 2026-01-26 22:29:13 +00:00
Louis King
345ffd219b Separate API prefix search from exact match endpoint
- Add /api/v1/nodes/prefix/{prefix} for prefix-based node lookup
- Change /api/v1/nodes/{public_key} to exact match only
- /n/{prefix} now simply redirects to /nodes/{prefix}
- /nodes/{key} resolves prefixes via API and redirects to full key
2026-01-26 22:27:15 +00:00
Louis King
9661b22390 Fix node detail 404 to use custom error page 2026-01-26 22:11:48 +00:00
Louis King
31aa48c9a0 Return 404 page when node not found in detail view 2026-01-26 22:08:01 +00:00
Louis King
1a3649b3be Revert "Simplify 404 page design"
This reverts commit 33649a065b.
2026-01-26 22:07:29 +00:00
Louis King
33649a065b Simplify 404 page design 2026-01-26 22:05:31 +00:00
Louis King
fd582bda35 Add custom 404 error page 2026-01-26 22:01:00 +00:00
Louis King
c42b26c8f3 Make /n/ short link resolve prefix to full public key 2026-01-26 21:57:04 +00:00
Louis King
d52163949a Change /n/ short link to redirect to /nodes/ 2026-01-26 21:48:55 +00:00
Louis King
ca101583f0 Add /n/ short link alias and simplify CI lint job
- Add /n/{public_key} route as alias for /nodes/{public_key} for shorter URLs
- Replace individual lint tools in CI with pre-commit/action for consistency
2026-01-26 21:41:33 +00:00
JingleManSweep
4af0f2ea80 Merge pull request #70 from ipnet-mesh/chore/node-page-prefix-support
Add prefix matching support to node API endpoint
2026-01-26 21:28:43 +00:00
Louis King
0b3ac64845 Add prefix matching support to node API endpoint
Allow users to navigate to a node using any prefix of its public key
instead of requiring the full 64-character key. If multiple nodes match
the prefix, the first one alphabetically is returned.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-26 21:27:36 +00:00
Louis King
3c7a8981ee Increased dedup bucket window to 120s 2026-01-17 20:15:46 +00:00
JingleManSweep
238e28ae41 Merge pull request #67 from ipnet-mesh/chore/tidyup-message-filters-columns
Message Filter and Table Tidying
2026-01-15 18:19:30 +00:00
Louis King
68d5049963 Removed pointless channel number filter and tidied column headings/values 2026-01-15 18:16:31 +00:00
JingleManSweep
624fa458ac Merge pull request #66 from ipnet-mesh/chore/fix-sqlite-path-exists
Ensure SQLite database path/subdirectories exist before initialising …
2026-01-15 17:36:58 +00:00
Louis King
309d575fc0 Ensure SQLite database path/subdirectories exist before initialising database 2026-01-15 17:32:56 +00:00
Louis King
f7b4df13a7 Added more test coverage 2026-01-12 21:00:02 +00:00
Louis King
13bae5c8d7 Added more test coverage 2026-01-12 20:34:53 +00:00
20 changed files with 1541 additions and 79 deletions

View File

@@ -18,20 +18,8 @@ jobs:
with:
python-version: "3.13"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install black flake8 mypy
pip install -e ".[dev]"
- name: Check formatting with black
run: black --check src/ tests/
- name: Lint with flake8
run: flake8 src/ tests/
- name: Type check with mypy
run: mypy src/
- name: Run pre-commit
uses: pre-commit/action@v3.0.1
test:
name: Test (Python ${{ matrix.python-version }})

View File

@@ -2,7 +2,7 @@
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from fastapi import APIRouter, HTTPException, Path, Query
from sqlalchemy import func, or_, select
from sqlalchemy.orm import selectinload
@@ -77,14 +77,43 @@ async def list_nodes(
)
@router.get("/{public_key}", response_model=NodeRead)
async def get_node(
@router.get("/prefix/{prefix}", response_model=NodeRead)
async def get_node_by_prefix(
_: RequireRead,
session: DbSession,
public_key: str,
prefix: str = Path(description="Public key prefix to search for"),
) -> NodeRead:
"""Get a single node by public key."""
query = select(Node).where(Node.public_key == public_key)
"""Get a single node by public key prefix.
Returns the first node (alphabetically by public_key) that matches the prefix.
"""
query = (
select(Node)
.options(selectinload(Node.tags))
.where(Node.public_key.startswith(prefix))
.order_by(Node.public_key)
.limit(1)
)
node = session.execute(query).scalar_one_or_none()
if not node:
raise HTTPException(status_code=404, detail="Node not found")
return NodeRead.model_validate(node)
@router.get("/{public_key}", response_model=NodeRead)
async def get_node(
_: RequireRead,
session: DbSession,
public_key: str = Path(description="Full 64-character public key"),
) -> NodeRead:
"""Get a single node by exact public key match."""
query = (
select(Node)
.options(selectinload(Node.tags))
.where(Node.public_key == public_key)
)
node = session.execute(query).scalar_one_or_none()
if not node:

View File

@@ -98,6 +98,15 @@ class DatabaseManager:
echo: Enable SQL query logging
"""
self.database_url = database_url
# Ensure parent directory exists for SQLite databases
if database_url.startswith("sqlite:///"):
from pathlib import Path
# Extract path from sqlite:///path/to/db.sqlite
db_path = Path(database_url.replace("sqlite:///", ""))
db_path.parent.mkdir(parents=True, exist_ok=True)
self.engine = create_database_engine(database_url, echo=echo)
self.session_factory = create_session_factory(self.engine)

View File

@@ -49,7 +49,7 @@ def compute_advertisement_hash(
adv_type: Optional[str] = None,
flags: Optional[int] = None,
received_at: Optional[datetime] = None,
bucket_seconds: int = 30,
bucket_seconds: int = 120,
) -> str:
"""Compute a deterministic hash for an advertisement.
@@ -104,7 +104,7 @@ def compute_telemetry_hash(
node_public_key: str,
parsed_data: Optional[dict] = None,
received_at: Optional[datetime] = None,
bucket_seconds: int = 30,
bucket_seconds: int = 120,
) -> str:
"""Compute a deterministic hash for a telemetry record.

View File

@@ -7,8 +7,10 @@ from typing import AsyncGenerator
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from starlette.exceptions import HTTPException as StarletteHTTPException
from meshcore_hub import __version__
from meshcore_hub.common.schemas import RadioConfig
@@ -150,6 +152,24 @@ def create_app(
except Exception as e:
return {"status": "not_ready", "api": str(e)}
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(
request: Request, exc: StarletteHTTPException
) -> HTMLResponse:
"""Handle HTTP exceptions with custom error pages."""
if exc.status_code == 404:
context = get_network_context(request)
context["request"] = request
context["detail"] = exc.detail if exc.detail != "Not Found" else None
return templates.TemplateResponse(
"errors/404.html", context, status_code=404
)
# For other errors, return a simple response
return HTMLResponse(
content=f"<h1>{exc.status_code}</h1><p>{exc.detail}</p>",
status_code=exc.status_code,
)
return app

View File

@@ -15,7 +15,6 @@ router = APIRouter()
async def messages_list(
request: Request,
message_type: str | None = Query(None, description="Filter by message type"),
channel_idx: str | None = Query(None, description="Filter by channel"),
search: str | None = Query(None, description="Search in message text"),
page: int = Query(1, ge=1, description="Page number"),
limit: int = Query(50, ge=1, le=100, description="Items per page"),
@@ -28,20 +27,10 @@ async def messages_list(
# Calculate offset
offset = (page - 1) * limit
# Parse channel_idx, treating empty string as None
channel_idx_int: int | None = None
if channel_idx and channel_idx.strip():
try:
channel_idx_int = int(channel_idx)
except ValueError:
logger.warning(f"Invalid channel_idx value: {channel_idx}")
# Build query params
params: dict[str, int | str] = {"limit": limit, "offset": offset}
if message_type:
params["message_type"] = message_type
if channel_idx_int is not None:
params["channel_idx"] = channel_idx_int
# Fetch messages from API
messages = []
@@ -70,7 +59,6 @@ async def messages_list(
"limit": limit,
"total_pages": total_pages,
"message_type": message_type or "",
"channel_idx": channel_idx_int,
"search": search or "",
}
)

View File

@@ -2,8 +2,8 @@
import logging
from fastapi import APIRouter, Query, Request
from fastapi.responses import HTMLResponse
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from meshcore_hub.web.app import get_network_context, get_templates
@@ -81,25 +81,45 @@ async def nodes_list(
return templates.TemplateResponse("nodes.html", context)
@router.get("/nodes/{public_key}", response_class=HTMLResponse)
async def node_detail(request: Request, public_key: str) -> HTMLResponse:
"""Render the node detail page."""
@router.get("/n/{prefix}")
async def node_short_link(prefix: str) -> RedirectResponse:
"""Redirect short link to nodes page."""
return RedirectResponse(url=f"/nodes/{prefix}", status_code=302)
@router.get("/nodes/{public_key}", response_model=None)
async def node_detail(
request: Request, public_key: str
) -> HTMLResponse | RedirectResponse:
"""Render the node detail page.
If the key is not a full 64-character public key, uses the prefix API
to resolve it and redirects to the canonical URL.
"""
# If not a full public key, resolve via prefix API and redirect
if len(public_key) != 64:
response = await request.app.state.http_client.get(
f"/api/v1/nodes/prefix/{public_key}"
)
if response.status_code == 200:
node = response.json()
return RedirectResponse(url=f"/nodes/{node['public_key']}", status_code=302)
raise HTTPException(status_code=404, detail="Node not found")
templates = get_templates(request)
context = get_network_context(request)
context["request"] = request
node = None
advertisements = []
telemetry = []
try:
# Fetch node details
response = await request.app.state.http_client.get(
f"/api/v1/nodes/{public_key}"
)
if response.status_code == 200:
node = response.json()
# Fetch node details (exact match)
response = await request.app.state.http_client.get(f"/api/v1/nodes/{public_key}")
if response.status_code != 200:
raise HTTPException(status_code=404, detail="Node not found")
node = response.json()
try:
# Fetch recent advertisements for this node
response = await request.app.state.http_client.get(
"/api/v1/advertisements", params={"public_key": public_key, "limit": 10}

View File

@@ -0,0 +1,35 @@
{% extends "base.html" %}
{% block title %}Page Not Found - {{ network_name }}{% endblock %}
{% block content %}
<div class="hero min-h-[60vh]">
<div class="hero-content text-center">
<div class="max-w-md">
<div class="text-9xl font-bold text-primary opacity-20">404</div>
<h1 class="text-4xl font-bold -mt-8">Page Not Found</h1>
<p class="py-6 text-base-content/70">
{% if detail %}
{{ detail }}
{% else %}
The page you're looking for doesn't exist or has been moved.
{% endif %}
</p>
<div class="flex gap-4 justify-center">
<a href="/" class="btn btn-primary">
<svg xmlns="http://www.w3.org/2000/svg" class="h-5 w-5 mr-2" fill="none" viewBox="0 0 24 24" stroke="currentColor">
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M3 12l2-2m0 0l7-7 7 7M5 10v10a1 1 0 001 1h3m10-11l2 2m-2-2v10a1 1 0 01-1 1h-3m-6 0a1 1 0 001-1v-4a1 1 0 011-1h2a1 1 0 011 1v4a1 1 0 001 1m-6 0h6" />
</svg>
Go Home
</a>
<a href="/nodes" class="btn btn-outline">
<svg xmlns="http://www.w3.org/2000/svg" class="h-5 w-5 mr-2" fill="none" viewBox="0 0 24 24" stroke="currentColor">
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M19 11H5m14 0a2 2 0 012 2v6a2 2 0 01-2 2H5a2 2 0 01-2-2v-6a2 2 0 012-2m14 0V9a2 2 0 00-2-2M5 11V9a2 2 0 012-2m0 0V5a2 2 0 012-2h6a2 2 0 012 2v2M7 7h10" />
</svg>
Browse Nodes
</a>
</div>
</div>
</div>
</div>
{% endblock %}

View File

@@ -28,21 +28,10 @@
</label>
<select name="message_type" class="select select-bordered select-sm">
<option value="">All Types</option>
<option value="direct" {% if message_type == 'direct' %}selected{% endif %}>Direct</option>
<option value="contact" {% if message_type == 'contact' %}selected{% endif %}>Direct</option>
<option value="channel" {% if message_type == 'channel' %}selected{% endif %}>Channel</option>
</select>
</div>
<div class="form-control">
<label class="label py-1">
<span class="label-text">Channel</span>
</label>
<select name="channel_idx" class="select select-bordered select-sm">
<option value="">All Channels</option>
{% for i in range(8) %}
<option value="{{ i }}" {% if channel_idx == i %}selected{% endif %}>Channel {{ i }}</option>
{% endfor %}
</select>
</div>
<div class="flex gap-2 w-full sm:w-auto">
<button type="submit" class="btn btn-primary btn-sm">Filter</button>
<a href="/messages" class="btn btn-ghost btn-sm">Clear</a>
@@ -64,7 +53,7 @@
<div class="min-w-0">
<div class="font-medium text-sm truncate">
{% if msg.message_type == 'channel' %}
<span class="font-mono">CH{{ msg.channel_idx }}</span>
<span class="opacity-60">Public</span>
{% else %}
{% if msg.sender_tag_name or msg.sender_name %}
{{ msg.sender_tag_name or msg.sender_name }}
@@ -105,7 +94,7 @@
<tr>
<th>Type</th>
<th>Time</th>
<th>From/Channel</th>
<th>From</th>
<th>Message</th>
<th>Receivers</th>
</tr>
@@ -121,7 +110,7 @@
</td>
<td class="text-sm whitespace-nowrap">
{% if msg.message_type == 'channel' %}
<span class="font-mono">CH{{ msg.channel_idx }}</span>
<span class="opacity-60">Public</span>
{% else %}
{% if msg.sender_tag_name or msg.sender_name %}
<span class="font-medium">{{ msg.sender_tag_name or msg.sender_name }}</span>
@@ -154,5 +143,5 @@
</table>
</div>
{{ pagination(page, total_pages, {"message_type": message_type, "channel_idx": channel_idx, "limit": limit}) }}
{{ pagination(page, total_pages, {"message_type": message_type, "limit": limit}) }}
{% endblock %}

View File

@@ -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

View File

@@ -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

View File

@@ -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

View File

@@ -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

View File

@@ -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."""
@@ -86,6 +146,54 @@ class TestGetNode:
response = client_no_auth.get("/api/v1/nodes/nonexistent123")
assert response.status_code == 404
def test_get_node_by_prefix(self, client_no_auth, sample_node):
"""Test getting a node by public key prefix."""
prefix = sample_node.public_key[:8] # First 8 chars
response = client_no_auth.get(f"/api/v1/nodes/prefix/{prefix}")
assert response.status_code == 200
data = response.json()
assert data["public_key"] == sample_node.public_key
def test_get_node_by_single_char_prefix(self, client_no_auth, sample_node):
"""Test getting a node by single character prefix."""
prefix = sample_node.public_key[0]
response = client_no_auth.get(f"/api/v1/nodes/prefix/{prefix}")
assert response.status_code == 200
data = response.json()
assert data["public_key"] == sample_node.public_key
def test_get_node_prefix_returns_first_alphabetically(
self, client_no_auth, api_db_session
):
"""Test that prefix match returns first node alphabetically."""
from datetime import datetime, timezone
from meshcore_hub.common.models import Node
# Create two nodes with same prefix but different suffixes
# abc... should come before abd...
node_a = Node(
public_key="abc0000000000000000000000000000000000000000000000000000000000000",
name="Node A",
adv_type="REPEATER",
first_seen=datetime.now(timezone.utc),
)
node_b = Node(
public_key="abc1111111111111111111111111111111111111111111111111111111111111",
name="Node B",
adv_type="REPEATER",
first_seen=datetime.now(timezone.utc),
)
api_db_session.add(node_a)
api_db_session.add(node_b)
api_db_session.commit()
# Request with prefix should return first alphabetically
response = client_no_auth.get("/api/v1/nodes/prefix/abc")
assert response.status_code == 200
data = response.json()
assert data["public_key"] == node_a.public_key
class TestNodeTags:
"""Tests for node tag endpoints."""

View File

@@ -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

View File

@@ -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

View File

@@ -40,7 +40,7 @@ class MockHttpClient:
"items": [
{
"id": "node-1",
"public_key": "abc123def456abc123def456abc123de",
"public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1",
"name": "Node One",
"adv_type": "REPEATER",
"last_seen": "2024-01-01T12:00:00Z",
@@ -48,7 +48,7 @@ class MockHttpClient:
},
{
"id": "node-2",
"public_key": "def456abc123def456abc123def456ab",
"public_key": "def456abc123def456abc123def456abc123def456abc123def456abc123def4",
"name": "Node Two",
"adv_type": "CLIENT",
"last_seen": "2024-01-01T11:00:00Z",
@@ -62,12 +62,14 @@ class MockHttpClient:
},
}
# Default single node response
self._responses["GET:/api/v1/nodes/abc123def456abc123def456abc123de"] = {
# Default single node response (exact match)
self._responses[
"GET:/api/v1/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
] = {
"status_code": 200,
"json": {
"id": "node-1",
"public_key": "abc123def456abc123def456abc123de",
"public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1",
"name": "Node One",
"adv_type": "REPEATER",
"last_seen": "2024-01-01T12:00:00Z",
@@ -110,7 +112,7 @@ class MockHttpClient:
"items": [
{
"id": "adv-1",
"public_key": "abc123def456abc123def456abc123de",
"public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1",
"name": "Node One",
"adv_type": "REPEATER",
"received_at": "2024-01-01T12:00:00Z",
@@ -127,7 +129,7 @@ class MockHttpClient:
"items": [
{
"id": "tel-1",
"node_public_key": "abc123def456abc123def456abc123de",
"node_public_key": "abc123def456abc123def456abc123def456abc123def456abc123def456abc1",
"parsed_data": {"battery_level": 85.5},
"received_at": "2024-01-01T12:00:00Z",
},

View File

@@ -0,0 +1,364 @@
"""Tests for the advertisements page route."""
from typing import Any
from fastapi.testclient import TestClient
from tests.test_web.conftest import MockHttpClient
class TestAdvertisementsPage:
"""Tests for the advertisements page."""
def test_advertisements_returns_200(self, client: TestClient) -> None:
"""Test that advertisements page returns 200 status code."""
response = client.get("/advertisements")
assert response.status_code == 200
def test_advertisements_returns_html(self, client: TestClient) -> None:
"""Test that advertisements page returns HTML content."""
response = client.get("/advertisements")
assert "text/html" in response.headers["content-type"]
def test_advertisements_contains_network_name(self, client: TestClient) -> None:
"""Test that advertisements page contains the network name."""
response = client.get("/advertisements")
assert "Test Network" in response.text
def test_advertisements_displays_advertisement_list(
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page displays advertisements from API."""
response = client.get("/advertisements")
assert response.status_code == 200
# Check for advertisement data from mock
assert "Node One" in response.text
def test_advertisements_displays_adv_type(
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page displays advertisement types."""
response = client.get("/advertisements")
# Should show adv type from mock data
assert "REPEATER" in response.text
class TestAdvertisementsPageFilters:
"""Tests for advertisements page filtering."""
def test_advertisements_with_search(self, client: TestClient) -> None:
"""Test advertisements page with search parameter."""
response = client.get("/advertisements?search=node")
assert response.status_code == 200
def test_advertisements_with_member_filter(self, client: TestClient) -> None:
"""Test advertisements page with member_id filter."""
response = client.get("/advertisements?member_id=alice")
assert response.status_code == 200
def test_advertisements_with_public_key_filter(self, client: TestClient) -> None:
"""Test advertisements page with public_key filter."""
response = client.get(
"/advertisements?public_key=abc123def456abc123def456abc123de"
)
assert response.status_code == 200
def test_advertisements_with_pagination(self, client: TestClient) -> None:
"""Test advertisements page with pagination parameters."""
response = client.get("/advertisements?page=1&limit=25")
assert response.status_code == 200
def test_advertisements_page_2(self, client: TestClient) -> None:
"""Test advertisements page 2."""
response = client.get("/advertisements?page=2")
assert response.status_code == 200
def test_advertisements_with_all_filters(self, client: TestClient) -> None:
"""Test advertisements page with multiple filters."""
response = client.get(
"/advertisements?search=test&member_id=alice&page=1&limit=10"
)
assert response.status_code == 200
class TestAdvertisementsPageDropdowns:
"""Tests for advertisements page dropdown data."""
def test_advertisements_loads_members_for_dropdown(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page loads members for filter dropdown."""
# Set up members response
mock_http_client.set_response(
"GET",
"/api/v1/members",
200,
{
"items": [
{"id": "m1", "member_id": "alice", "name": "Alice"},
{"id": "m2", "member_id": "bob", "name": "Bob"},
],
"total": 2,
},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
# Members should be available for dropdown
assert "Alice" in response.text or "alice" in response.text
def test_advertisements_loads_nodes_for_dropdown(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page loads nodes for filter dropdown."""
# Set up nodes response with tags
mock_http_client.set_response(
"GET",
"/api/v1/nodes",
200,
{
"items": [
{
"id": "n1",
"public_key": "abc123",
"name": "Node Alpha",
"tags": [{"key": "name", "value": "Custom Name"}],
},
{
"id": "n2",
"public_key": "def456",
"name": "Node Beta",
"tags": [],
},
],
"total": 2,
},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
class TestAdvertisementsNodeSorting:
"""Tests for node sorting in advertisements dropdown."""
def test_nodes_sorted_by_display_name(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that nodes are sorted alphabetically by display name."""
# Set up nodes with tags - "Zebra" should come after "Alpha"
mock_http_client.set_response(
"GET",
"/api/v1/nodes",
200,
{
"items": [
{
"id": "n1",
"public_key": "abc123",
"name": "Zebra Node",
"tags": [],
},
{
"id": "n2",
"public_key": "def456",
"name": "Alpha Node",
"tags": [],
},
],
"total": 2,
},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
# Both nodes should appear
text = response.text
assert "Alpha Node" in text or "alpha" in text.lower()
assert "Zebra Node" in text or "zebra" in text.lower()
def test_nodes_sorted_by_tag_name_when_present(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that nodes use tag name for sorting when available."""
mock_http_client.set_response(
"GET",
"/api/v1/nodes",
200,
{
"items": [
{
"id": "n1",
"public_key": "abc123",
"name": "Zebra",
"tags": [{"key": "name", "value": "Alpha Custom"}],
},
],
"total": 1,
},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
def test_nodes_fallback_to_public_key_when_no_name(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that nodes fall back to public_key when no name."""
mock_http_client.set_response(
"GET",
"/api/v1/nodes",
200,
{
"items": [
{
"id": "n1",
"public_key": "abc123def456",
"name": None,
"tags": [],
},
],
"total": 1,
},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
class TestAdvertisementsPageAPIErrors:
"""Tests for advertisements page handling API errors."""
def test_advertisements_handles_api_error(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page handles API errors gracefully."""
mock_http_client.set_response(
"GET", "/api/v1/advertisements", status_code=500, json_data=None
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
# Should still return 200 (page renders with empty list)
assert response.status_code == 200
def test_advertisements_handles_api_not_found(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that advertisements page handles API 404 gracefully."""
mock_http_client.set_response(
"GET",
"/api/v1/advertisements",
status_code=404,
json_data={"detail": "Not found"},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
# Should still return 200 (page renders with empty list)
assert response.status_code == 200
def test_advertisements_handles_members_api_error(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that page handles members API error gracefully."""
mock_http_client.set_response(
"GET", "/api/v1/members", status_code=500, json_data=None
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
# Should still return 200 (page renders without member dropdown)
assert response.status_code == 200
def test_advertisements_handles_nodes_api_error(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that page handles nodes API error gracefully."""
mock_http_client.set_response(
"GET", "/api/v1/nodes", status_code=500, json_data=None
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
# Should still return 200 (page renders without node dropdown)
assert response.status_code == 200
def test_advertisements_handles_empty_response(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that page handles empty advertisements list."""
mock_http_client.set_response(
"GET",
"/api/v1/advertisements",
200,
{"items": [], "total": 0},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200
class TestAdvertisementsPagination:
"""Tests for advertisements pagination calculations."""
def test_pagination_calculates_total_pages(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that pagination correctly calculates total pages."""
mock_http_client.set_response(
"GET",
"/api/v1/advertisements",
200,
{"items": [], "total": 150},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
# With limit=50 and total=150, should have 3 pages
response = client.get("/advertisements?limit=50")
assert response.status_code == 200
def test_pagination_with_zero_total(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test pagination with zero results shows at least 1 page."""
mock_http_client.set_response(
"GET",
"/api/v1/advertisements",
200,
{"items": [], "total": 0},
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/advertisements")
assert response.status_code == 200

View File

@@ -0,0 +1,87 @@
"""Tests for the health check endpoints."""
from typing import Any
from fastapi.testclient import TestClient
from meshcore_hub import __version__
from tests.test_web.conftest import MockHttpClient
class TestHealthEndpoint:
"""Tests for the /health endpoint."""
def test_health_returns_200(self, client: TestClient) -> None:
"""Test that health endpoint returns 200 status code."""
response = client.get("/health")
assert response.status_code == 200
def test_health_returns_json(self, client: TestClient) -> None:
"""Test that health endpoint returns JSON content."""
response = client.get("/health")
assert "application/json" in response.headers["content-type"]
def test_health_returns_healthy_status(self, client: TestClient) -> None:
"""Test that health endpoint returns healthy status."""
response = client.get("/health")
data = response.json()
assert data["status"] == "healthy"
def test_health_returns_version(self, client: TestClient) -> None:
"""Test that health endpoint returns version."""
response = client.get("/health")
data = response.json()
assert data["version"] == __version__
class TestHealthReadyEndpoint:
"""Tests for the /health/ready endpoint."""
def test_health_ready_returns_200(self, client: TestClient) -> None:
"""Test that health/ready endpoint returns 200 status code."""
response = client.get("/health/ready")
assert response.status_code == 200
def test_health_ready_returns_json(self, client: TestClient) -> None:
"""Test that health/ready endpoint returns JSON content."""
response = client.get("/health/ready")
assert "application/json" in response.headers["content-type"]
def test_health_ready_returns_ready_status(self, client: TestClient) -> None:
"""Test that health/ready returns ready status when API is connected."""
response = client.get("/health/ready")
data = response.json()
assert data["status"] == "ready"
assert data["api"] == "connected"
def test_health_ready_with_api_error(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that health/ready handles API errors gracefully."""
mock_http_client.set_response("GET", "/health", status_code=500, json_data=None)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/health/ready")
assert response.status_code == 200
data = response.json()
assert data["status"] == "not_ready"
assert "status 500" in data["api"]
def test_health_ready_with_api_404(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that health/ready handles API 404 response."""
mock_http_client.set_response(
"GET", "/health", status_code=404, json_data={"detail": "Not found"}
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
response = client.get("/health/ready")
assert response.status_code == 200
data = response.json()
assert data["status"] == "not_ready"
assert "status 404" in data["api"]

View File

@@ -73,21 +73,27 @@ class TestNodeDetailPage:
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that node detail page returns 200 status code."""
response = client.get("/nodes/abc123def456abc123def456abc123de")
response = client.get(
"/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
)
assert response.status_code == 200
def test_node_detail_returns_html(
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that node detail page returns HTML content."""
response = client.get("/nodes/abc123def456abc123def456abc123de")
response = client.get(
"/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
)
assert "text/html" in response.headers["content-type"]
def test_node_detail_displays_node_info(
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that node detail page displays node information."""
response = client.get("/nodes/abc123def456abc123def456abc123de")
response = client.get(
"/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
)
assert response.status_code == 200
# Should display node details
assert "Node One" in response.text
@@ -98,8 +104,13 @@ class TestNodeDetailPage:
self, client: TestClient, mock_http_client: MockHttpClient
) -> None:
"""Test that node detail page displays the full public key."""
response = client.get("/nodes/abc123def456abc123def456abc123de")
assert "abc123def456abc123def456abc123de" in response.text
response = client.get(
"/nodes/abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
)
assert (
"abc123def456abc123def456abc123def456abc123def456abc123def456abc1"
in response.text
)
class TestNodesPageAPIErrors:
@@ -123,7 +134,7 @@ class TestNodesPageAPIErrors:
def test_node_detail_handles_not_found(
self, web_app: Any, mock_http_client: MockHttpClient
) -> None:
"""Test that node detail page handles 404 from API."""
"""Test that node detail page returns 404 when node not found."""
mock_http_client.set_response(
"GET",
"/api/v1/nodes/nonexistent",
@@ -132,8 +143,9 @@ class TestNodesPageAPIErrors:
)
web_app.state.http_client = mock_http_client
client = TestClient(web_app, raise_server_exceptions=True)
client = TestClient(web_app, raise_server_exceptions=False)
response = client.get("/nodes/nonexistent")
# Should still return 200 (page renders but shows no node)
assert response.status_code == 200
# Should return 404 with custom error page
assert response.status_code == 404
assert "Page Not Found" in response.text