mirror of
https://github.com/ipnet-mesh/meshcore-hub.git
synced 2026-03-28 17:42:56 +01:00
Compare commits
19 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fd3c3171ce | ||
|
|
345ffd219b | ||
|
|
9661b22390 | ||
|
|
31aa48c9a0 | ||
|
|
1a3649b3be | ||
|
|
33649a065b | ||
|
|
fd582bda35 | ||
|
|
c42b26c8f3 | ||
|
|
d52163949a | ||
|
|
ca101583f0 | ||
|
|
4af0f2ea80 | ||
|
|
0b3ac64845 | ||
|
|
3c7a8981ee | ||
|
|
238e28ae41 | ||
|
|
68d5049963 | ||
|
|
624fa458ac | ||
|
|
309d575fc0 | ||
|
|
f7b4df13a7 | ||
|
|
13bae5c8d7 |
16
.github/workflows/ci.yml
vendored
16
.github/workflows/ci.yml
vendored
@@ -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 }})
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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 "",
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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}
|
||||
|
||||
35
src/meshcore_hub/web/templates/errors/404.html
Normal file
35
src/meshcore_hub/web/templates/errors/404.html
Normal 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 %}
|
||||
@@ -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 %}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
285
tests/test_api/test_members.py
Normal file
285
tests/test_api/test_members.py
Normal 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
|
||||
@@ -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
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
},
|
||||
|
||||
364
tests/test_web/test_advertisements.py
Normal file
364
tests/test_web/test_advertisements.py
Normal 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
|
||||
87
tests/test_web/test_health.py
Normal file
87
tests/test_web/test_health.py
Normal 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"]
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user