From 367f838371b453fcc07e7cb5ae7946548778280d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Jan 2026 01:34:07 +0000 Subject: [PATCH] Add admin interface for managing node tags Implement CRUD operations for NodeTags in the admin interface: - Add NodeTagMove schema for moving tags between nodes - Add PUT /nodes/{public_key}/tags/{key}/move API endpoint - Add web routes at /a/node-tags for tag management - Create admin templates with node selector and tag management UI - Support editing, adding, moving, and deleting tags via API calls - Add comprehensive tests for new functionality The interface allows selecting a node from a dropdown, viewing its tags, and performing all CRUD operations including moving a tag to a different node without having to delete and recreate it. --- src/meshcore_hub/api/routes/node_tags.py | 59 ++- src/meshcore_hub/common/schemas/nodes.py | 11 + src/meshcore_hub/web/routes/admin.py | 240 ++++++++++++- .../{admin.html => admin/index.html} | 26 +- .../web/templates/admin/node_tags.html | 340 ++++++++++++++++++ tests/test_api/test_nodes.py | 130 +++++++ tests/test_web/conftest.py | 12 + tests/test_web/test_admin.py | 338 +++++++++++++++++ 8 files changed, 1142 insertions(+), 14 deletions(-) rename src/meshcore_hub/web/templates/{admin.html => admin/index.html} (54%) create mode 100644 src/meshcore_hub/web/templates/admin/node_tags.html create mode 100644 tests/test_web/test_admin.py diff --git a/src/meshcore_hub/api/routes/node_tags.py b/src/meshcore_hub/api/routes/node_tags.py index 116530a..ff93fc3 100644 --- a/src/meshcore_hub/api/routes/node_tags.py +++ b/src/meshcore_hub/api/routes/node_tags.py @@ -6,7 +6,12 @@ from sqlalchemy import select from meshcore_hub.api.auth import RequireAdmin, RequireRead from meshcore_hub.api.dependencies import DbSession from meshcore_hub.common.models import Node, NodeTag -from meshcore_hub.common.schemas.nodes import NodeTagCreate, NodeTagRead, NodeTagUpdate +from meshcore_hub.common.schemas.nodes import ( + NodeTagCreate, + NodeTagMove, + NodeTagRead, + NodeTagUpdate, +) router = APIRouter() @@ -130,6 +135,58 @@ async def update_node_tag( return NodeTagRead.model_validate(node_tag) +@router.put("/nodes/{public_key}/tags/{key}/move", response_model=NodeTagRead) +async def move_node_tag( + _: RequireAdmin, + session: DbSession, + public_key: str, + key: str, + data: NodeTagMove, +) -> NodeTagRead: + """Move a node tag to a different node.""" + # Find source node + source_query = select(Node).where(Node.public_key == public_key) + source_node = session.execute(source_query).scalar_one_or_none() + + if not source_node: + raise HTTPException(status_code=404, detail="Source node not found") + + # Find tag + tag_query = select(NodeTag).where( + (NodeTag.node_id == source_node.id) & (NodeTag.key == key) + ) + node_tag = session.execute(tag_query).scalar_one_or_none() + + if not node_tag: + raise HTTPException(status_code=404, detail="Tag not found") + + # Find destination node + dest_query = select(Node).where(Node.public_key == data.new_public_key) + dest_node = session.execute(dest_query).scalar_one_or_none() + + if not dest_node: + raise HTTPException(status_code=404, detail="Destination node not found") + + # Check if tag already exists on destination node + conflict_query = select(NodeTag).where( + (NodeTag.node_id == dest_node.id) & (NodeTag.key == key) + ) + conflict = session.execute(conflict_query).scalar_one_or_none() + + if conflict: + raise HTTPException( + status_code=409, + detail=f"Tag '{key}' already exists on destination node", + ) + + # Move tag to destination node + node_tag.node_id = dest_node.id + session.commit() + session.refresh(node_tag) + + return NodeTagRead.model_validate(node_tag) + + @router.delete("/nodes/{public_key}/tags/{key}", status_code=204) async def delete_node_tag( _: RequireAdmin, diff --git a/src/meshcore_hub/common/schemas/nodes.py b/src/meshcore_hub/common/schemas/nodes.py index 2f6fdbf..04ab6bc 100644 --- a/src/meshcore_hub/common/schemas/nodes.py +++ b/src/meshcore_hub/common/schemas/nodes.py @@ -38,6 +38,17 @@ class NodeTagUpdate(BaseModel): ) +class NodeTagMove(BaseModel): + """Schema for moving a node tag to a different node.""" + + new_public_key: str = Field( + ..., + min_length=64, + max_length=64, + description="Public key of the destination node", + ) + + class NodeTagRead(BaseModel): """Schema for reading a node tag.""" diff --git a/src/meshcore_hub/web/routes/admin.py b/src/meshcore_hub/web/routes/admin.py index dd5b7ef..3704db5 100644 --- a/src/meshcore_hub/web/routes/admin.py +++ b/src/meshcore_hub/web/routes/admin.py @@ -1,9 +1,10 @@ -"""Admin page route.""" +"""Admin page routes.""" import logging +from typing import Optional -from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import HTMLResponse +from fastapi import APIRouter, Form, HTTPException, Query, Request +from fastapi.responses import HTMLResponse, RedirectResponse from meshcore_hub.web.app import get_network_context, get_templates @@ -11,21 +12,236 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/a", tags=["admin"]) +def _check_admin_enabled(request: Request) -> None: + """Check if admin interface is enabled, raise 404 if not.""" + if not getattr(request.app.state, "admin_enabled", False): + raise HTTPException(status_code=404, detail="Not Found") + + +def _get_auth_context(request: Request) -> dict: + """Extract OAuth2Proxy authentication headers.""" + return { + "auth_user": request.headers.get("X-Forwarded-User"), + "auth_groups": request.headers.get("X-Forwarded-Groups"), + "auth_email": request.headers.get("X-Forwarded-Email"), + "auth_username": request.headers.get("X-Forwarded-Preferred-Username"), + } + + @router.get("/", response_class=HTMLResponse) async def admin_home(request: Request) -> HTMLResponse: """Render the admin page with OAuth2Proxy user info.""" - # Check if admin interface is enabled - if not getattr(request.app.state, "admin_enabled", False): - raise HTTPException(status_code=404, detail="Not Found") + _check_admin_enabled(request) templates = get_templates(request) context = get_network_context(request) context["request"] = request + context.update(_get_auth_context(request)) - # Extract OAuth2Proxy headers - context["auth_user"] = request.headers.get("X-Forwarded-User") - context["auth_groups"] = request.headers.get("X-Forwarded-Groups") - context["auth_email"] = request.headers.get("X-Forwarded-Email") - context["auth_username"] = request.headers.get("X-Forwarded-Preferred-Username") + return templates.TemplateResponse("admin/index.html", context) - return templates.TemplateResponse("admin.html", context) + +@router.get("/node-tags", response_class=HTMLResponse) +async def admin_node_tags( + request: Request, + public_key: Optional[str] = Query(None), + message: Optional[str] = Query(None), + error: Optional[str] = Query(None), +) -> HTMLResponse: + """Admin page for managing node tags.""" + _check_admin_enabled(request) + + templates = get_templates(request) + context = get_network_context(request) + context["request"] = request + context.update(_get_auth_context(request)) + + # Flash messages from redirects + context["message"] = message + context["error"] = error + + # Fetch all nodes for dropdown + nodes = [] + try: + response = await request.app.state.http_client.get( + "/api/v1/nodes", + params={"limit": 100}, + ) + if response.status_code == 200: + data = response.json() + nodes = data.get("items", []) + except Exception as e: + logger.exception("Failed to fetch nodes: %s", e) + context["error"] = "Failed to fetch nodes" + + context["nodes"] = nodes + context["selected_public_key"] = public_key + + # Fetch tags for selected node + tags = [] + selected_node = None + if public_key: + # Find the selected node in the list + for node in nodes: + if node.get("public_key") == public_key: + selected_node = node + break + + try: + response = await request.app.state.http_client.get( + f"/api/v1/nodes/{public_key}/tags", + ) + if response.status_code == 200: + tags = response.json() + elif response.status_code == 404: + context["error"] = "Node not found" + except Exception as e: + logger.exception("Failed to fetch tags: %s", e) + context["error"] = "Failed to fetch tags" + + context["tags"] = tags + context["selected_node"] = selected_node + + return templates.TemplateResponse("admin/node_tags.html", context) + + +@router.post("/node-tags", response_class=RedirectResponse) +async def admin_create_node_tag( + request: Request, + public_key: str = Form(...), + key: str = Form(...), + value: str = Form(""), + value_type: str = Form("string"), +) -> RedirectResponse: + """Create a new node tag.""" + _check_admin_enabled(request) + + redirect_url = f"/a/node-tags?public_key={public_key}" + + try: + response = await request.app.state.http_client.post( + f"/api/v1/nodes/{public_key}/tags", + json={ + "key": key, + "value": value or None, + "value_type": value_type, + }, + ) + if response.status_code == 201: + redirect_url += f"&message=Tag '{key}' created successfully" + elif response.status_code == 409: + redirect_url += f"&error=Tag '{key}' already exists" + elif response.status_code == 404: + redirect_url += "&error=Node not found" + else: + detail = response.json().get("detail", "Unknown error") + redirect_url += f"&error={detail}" + except Exception as e: + logger.exception("Failed to create tag: %s", e) + redirect_url += "&error=Failed to create tag" + + return RedirectResponse(url=redirect_url, status_code=303) + + +@router.post("/node-tags/update", response_class=RedirectResponse) +async def admin_update_node_tag( + request: Request, + public_key: str = Form(...), + key: str = Form(...), + value: str = Form(""), + value_type: str = Form("string"), +) -> RedirectResponse: + """Update an existing node tag.""" + _check_admin_enabled(request) + + redirect_url = f"/a/node-tags?public_key={public_key}" + + try: + response = await request.app.state.http_client.put( + f"/api/v1/nodes/{public_key}/tags/{key}", + json={ + "value": value or None, + "value_type": value_type, + }, + ) + if response.status_code == 200: + redirect_url += f"&message=Tag '{key}' updated successfully" + elif response.status_code == 404: + redirect_url += f"&error=Tag '{key}' not found" + else: + detail = response.json().get("detail", "Unknown error") + redirect_url += f"&error={detail}" + except Exception as e: + logger.exception("Failed to update tag: %s", e) + redirect_url += "&error=Failed to update tag" + + return RedirectResponse(url=redirect_url, status_code=303) + + +@router.post("/node-tags/move", response_class=RedirectResponse) +async def admin_move_node_tag( + request: Request, + public_key: str = Form(...), + key: str = Form(...), + new_public_key: str = Form(...), +) -> RedirectResponse: + """Move a node tag to a different node.""" + _check_admin_enabled(request) + + # Redirect to the destination node after move + redirect_url = f"/a/node-tags?public_key={new_public_key}" + + try: + response = await request.app.state.http_client.put( + f"/api/v1/nodes/{public_key}/tags/{key}/move", + json={"new_public_key": new_public_key}, + ) + if response.status_code == 200: + redirect_url += f"&message=Tag '{key}' moved successfully" + elif response.status_code == 404: + # Stay on source node if not found + redirect_url = f"/a/node-tags?public_key={public_key}" + detail = response.json().get("detail", "Not found") + redirect_url += f"&error={detail}" + elif response.status_code == 409: + redirect_url = f"/a/node-tags?public_key={public_key}" + redirect_url += f"&error=Tag '{key}' already exists on destination node" + else: + redirect_url = f"/a/node-tags?public_key={public_key}" + detail = response.json().get("detail", "Unknown error") + redirect_url += f"&error={detail}" + except Exception as e: + logger.exception("Failed to move tag: %s", e) + redirect_url = f"/a/node-tags?public_key={public_key}" + redirect_url += "&error=Failed to move tag" + + return RedirectResponse(url=redirect_url, status_code=303) + + +@router.post("/node-tags/delete", response_class=RedirectResponse) +async def admin_delete_node_tag( + request: Request, + public_key: str = Form(...), + key: str = Form(...), +) -> RedirectResponse: + """Delete a node tag.""" + _check_admin_enabled(request) + + redirect_url = f"/a/node-tags?public_key={public_key}" + + try: + response = await request.app.state.http_client.delete( + f"/api/v1/nodes/{public_key}/tags/{key}", + ) + if response.status_code == 204: + redirect_url += f"&message=Tag '{key}' deleted successfully" + elif response.status_code == 404: + redirect_url += f"&error=Tag '{key}' not found" + else: + detail = response.json().get("detail", "Unknown error") + redirect_url += f"&error={detail}" + except Exception as e: + logger.exception("Failed to delete tag: %s", e) + redirect_url += "&error=Failed to delete tag" + + return RedirectResponse(url=redirect_url, status_code=303) diff --git a/src/meshcore_hub/web/templates/admin.html b/src/meshcore_hub/web/templates/admin/index.html similarity index 54% rename from src/meshcore_hub/web/templates/admin.html rename to src/meshcore_hub/web/templates/admin/index.html index 7ab6db3..72ecd63 100644 --- a/src/meshcore_hub/web/templates/admin.html +++ b/src/meshcore_hub/web/templates/admin/index.html @@ -4,10 +4,34 @@ {% block content %}
-

Admin

+
+

Admin

+
+ +
+
Sign Out
+ +
+ +
+

+ + + + Node Tags +

+

Manage custom tags and metadata for network nodes.

+
+ +
+ +

Authenticated User

diff --git a/src/meshcore_hub/web/templates/admin/node_tags.html b/src/meshcore_hub/web/templates/admin/node_tags.html new file mode 100644 index 0000000..8ee7bef --- /dev/null +++ b/src/meshcore_hub/web/templates/admin/node_tags.html @@ -0,0 +1,340 @@ +{% extends "base.html" %} + +{% block title %}{{ network_name }} - Node Tags Admin{% endblock %} + +{% block content %} +
+
+

Node Tags

+
+ +
+
+ Sign Out +
+ + +{% if message %} +
+ + + + {{ message }} +
+{% endif %} + +{% if error %} +
+ + + + {{ error }} +
+{% endif %} + + +
+
+

Select Node

+
+
+ + +
+ +
+
+
+ +{% if selected_public_key and selected_node %} + +
+
+
+
+

{{ selected_node.name or 'Unnamed Node' }}

+

{{ selected_public_key }}

+ {% if selected_node.adv_type %} + {{ selected_node.adv_type }} + {% endif %} +
+ View Node +
+
+
+ + +
+
+

Tags ({{ tags|length }})

+ + {% if tags %} +
+ + + + + + + + + + + + {% for tag in tags %} + + + + + + + + {% endfor %} + +
KeyValueTypeUpdatedActions
{{ tag.key }}{{ tag.value or '-' }} + {{ tag.value_type }} + {{ tag.updated_at[:10] if tag.updated_at else '-' }} +
+ + + +
+
+
+ {% else %} +
+

No tags found for this node.

+

Add a new tag below.

+
+ {% endif %} +
+
+ + +
+
+

Add New Tag

+
+ + +
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+
+
+
+ + + +
+

Edit Tag

+
+ + + +
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+
+
+
+ +
+ + + + +
+

Move Tag to Another Node

+
+ + + +
+ + +
+ +
+ + +
+ +
+ + + + This will move the tag from the current node to the destination node. +
+ +
+ + +
+
+
+
+ +
+ + + + +
+

Delete Tag

+
+ + + +

Are you sure you want to delete the tag ""?

+ +
+ + + + This action cannot be undone. +
+ +
+ + +
+
+
+
+ +
+ + +{% elif selected_public_key and not selected_node %} +
+ + + + Node not found: {{ selected_public_key }} +
+{% else %} +
+
+ + + +

Select a Node

+

Choose a node from the dropdown above to view and manage its tags.

+
+
+{% endif %} +{% endblock %} + +{% block extra_scripts %} + +{% endblock %} diff --git a/tests/test_api/test_nodes.py b/tests/test_api/test_nodes.py index c0838bc..f8a47f2 100644 --- a/tests/test_api/test_nodes.py +++ b/tests/test_api/test_nodes.py @@ -158,3 +158,133 @@ class TestNodeTags: headers={"Authorization": "Bearer test-admin-key"}, ) assert response.status_code == 201 # Created + + +class TestMoveNodeTag: + """Tests for PUT /nodes/{public_key}/tags/{key}/move endpoint.""" + + # 64-character public key for testing + DEST_PUBLIC_KEY = "xyz789xyz789xyz789xyz789xyz789xyabc123abc123abc123abc123abc123ab" + + def test_move_node_tag_success( + self, client_no_auth, api_db_session, sample_node, sample_node_tag + ): + """Test successfully moving a tag to another node.""" + from meshcore_hub.common.models import Node + from datetime import datetime, timezone + + # Create a second node with 64-char public key + second_node = Node( + public_key=self.DEST_PUBLIC_KEY, + name="Second Node", + adv_type="CHAT", + first_seen=datetime.now(timezone.utc), + ) + api_db_session.add(second_node) + api_db_session.commit() + + response = client_no_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}/move", + json={"new_public_key": second_node.public_key}, + ) + assert response.status_code == 200 + data = response.json() + assert data["key"] == sample_node_tag.key + assert data["value"] == sample_node_tag.value + + # Verify tag is no longer on original node + response = client_no_auth.get( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}" + ) + assert response.status_code == 404 + + # Verify tag is now on new node + response = client_no_auth.get( + f"/api/v1/nodes/{second_node.public_key}/tags/{sample_node_tag.key}" + ) + assert response.status_code == 200 + + def test_move_node_tag_source_not_found(self, client_no_auth): + """Test moving a tag from a non-existent node.""" + response = client_no_auth.put( + "/api/v1/nodes/nonexistent123/tags/somekey/move", + json={"new_public_key": self.DEST_PUBLIC_KEY}, + ) + assert response.status_code == 404 + assert "Source node not found" in response.json()["detail"] + + def test_move_node_tag_tag_not_found(self, client_no_auth, sample_node): + """Test moving a non-existent tag.""" + response = client_no_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/nonexistent/move", + json={"new_public_key": self.DEST_PUBLIC_KEY}, + ) + assert response.status_code == 404 + assert "Tag not found" in response.json()["detail"] + + def test_move_node_tag_dest_not_found( + self, client_no_auth, sample_node, sample_node_tag + ): + """Test moving a tag to a non-existent destination node.""" + # 64-character nonexistent public key + nonexistent_key = ( + "1111111111111111111111111111111122222222222222222222222222222222" + ) + response = client_no_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}/move", + json={"new_public_key": nonexistent_key}, + ) + assert response.status_code == 404 + assert "Destination node not found" in response.json()["detail"] + + def test_move_node_tag_conflict( + self, client_no_auth, api_db_session, sample_node, sample_node_tag + ): + """Test moving a tag when destination already has that key.""" + from meshcore_hub.common.models import Node, NodeTag + from datetime import datetime, timezone + + # Create second node with same tag key + second_node = Node( + public_key=self.DEST_PUBLIC_KEY, + name="Second Node", + adv_type="CHAT", + first_seen=datetime.now(timezone.utc), + ) + api_db_session.add(second_node) + api_db_session.commit() + + # Add the same tag key to second node + existing_tag = NodeTag( + node_id=second_node.id, + key=sample_node_tag.key, # Same key + value="different value", + ) + api_db_session.add(existing_tag) + api_db_session.commit() + + response = client_no_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}/move", + json={"new_public_key": second_node.public_key}, + ) + assert response.status_code == 409 + assert "already exists on destination" in response.json()["detail"] + + def test_move_node_tag_requires_admin( + self, client_with_auth, sample_node, sample_node_tag + ): + """Test that move operation requires admin auth.""" + # Without auth + response = client_with_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}/move", + json={"new_public_key": self.DEST_PUBLIC_KEY}, + ) + assert response.status_code == 401 + + # With read key (not admin) + response = client_with_auth.put( + f"/api/v1/nodes/{sample_node.public_key}/tags/{sample_node_tag.key}/move", + json={"new_public_key": self.DEST_PUBLIC_KEY}, + headers={"Authorization": "Bearer test-read-key"}, + ) + assert response.status_code == 403 diff --git a/tests/test_web/conftest.py b/tests/test_web/conftest.py index 373f9b7..d069ce4 100644 --- a/tests/test_web/conftest.py +++ b/tests/test_web/conftest.py @@ -255,6 +255,18 @@ class MockHttpClient: key = f"POST:{path}" return self._create_response(key) + async def put( + self, path: str, json: dict | None = None, params: dict | None = None + ) -> Response: + """Mock PUT request.""" + key = f"PUT:{path}" + return self._create_response(key) + + async def delete(self, path: str, params: dict | None = None) -> Response: + """Mock DELETE request.""" + key = f"DELETE:{path}" + return self._create_response(key) + async def aclose(self) -> None: """Mock close method.""" pass diff --git a/tests/test_web/test_admin.py b/tests/test_web/test_admin.py new file mode 100644 index 0000000..d588775 --- /dev/null +++ b/tests/test_web/test_admin.py @@ -0,0 +1,338 @@ +"""Tests for admin web routes.""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from meshcore_hub.web.app import create_app + +from .conftest import MockHttpClient + + +@pytest.fixture +def mock_http_client_admin() -> MockHttpClient: + """Create a mock HTTP client for admin tests.""" + client = MockHttpClient() + + # Mock the nodes API response for admin dropdown + client.set_response( + "GET", + "/api/v1/nodes", + 200, + { + "items": [ + { + "public_key": "abc123def456abc123def456abc123de", + "name": "Node One", + "adv_type": "REPEATER", + "first_seen": "2024-01-01T00:00:00Z", + "last_seen": "2024-01-01T12:00:00Z", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + "tags": [], + }, + { + "public_key": "xyz789xyz789xyz789xyz789xyz789xy", + "name": "Node Two", + "adv_type": "CHAT", + "first_seen": "2024-01-01T00:00:00Z", + "last_seen": "2024-01-01T11:00:00Z", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + "tags": [], + }, + ], + "total": 2, + "limit": 100, + "offset": 0, + }, + ) + + # Mock node tags response + client.set_response( + "GET", + "/api/v1/nodes/abc123def456abc123def456abc123de/tags", + 200, + [ + { + "key": "environment", + "value": "production", + "value_type": "string", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + }, + { + "key": "location", + "value": "building-a", + "value_type": "string", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + }, + ], + ) + + # Mock create tag response + client.set_response( + "POST", + "/api/v1/nodes/abc123def456abc123def456abc123de/tags", + 201, + { + "key": "new_tag", + "value": "new_value", + "value_type": "string", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + }, + ) + + # Mock update tag response + client.set_response( + "PUT", + "/api/v1/nodes/abc123def456abc123def456abc123de/tags/environment", + 200, + { + "key": "environment", + "value": "staging", + "value_type": "string", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T12:00:00Z", + }, + ) + + # Mock move tag response + client.set_response( + "PUT", + "/api/v1/nodes/abc123def456abc123def456abc123de/tags/environment/move", + 200, + { + "key": "environment", + "value": "production", + "value_type": "string", + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T12:00:00Z", + }, + ) + + # Mock delete tag response + client.set_response( + "DELETE", + "/api/v1/nodes/abc123def456abc123def456abc123de/tags/environment", + 204, + None, + ) + + return client + + +@pytest.fixture +def admin_app(mock_http_client_admin: MockHttpClient) -> Any: + """Create a web app with admin enabled.""" + app = create_app( + api_url="http://localhost:8000", + api_key="test-api-key", + network_name="Test Network", + network_city="Test City", + network_country="Test Country", + network_radio_config="Test Radio Config", + network_contact_email="test@example.com", + admin_enabled=True, + ) + + app.state.http_client = mock_http_client_admin + + return app + + +@pytest.fixture +def admin_app_disabled(mock_http_client_admin: MockHttpClient) -> Any: + """Create a web app with admin disabled.""" + app = create_app( + api_url="http://localhost:8000", + api_key="test-api-key", + network_name="Test Network", + network_city="Test City", + network_country="Test Country", + network_radio_config="Test Radio Config", + network_contact_email="test@example.com", + admin_enabled=False, + ) + + app.state.http_client = mock_http_client_admin + + return app + + +@pytest.fixture +def admin_client(admin_app: Any, mock_http_client_admin: MockHttpClient) -> TestClient: + """Create a test client with admin enabled.""" + admin_app.state.http_client = mock_http_client_admin + return TestClient(admin_app, raise_server_exceptions=True) + + +@pytest.fixture +def admin_client_disabled( + admin_app_disabled: Any, mock_http_client_admin: MockHttpClient +) -> TestClient: + """Create a test client with admin disabled.""" + admin_app_disabled.state.http_client = mock_http_client_admin + return TestClient(admin_app_disabled, raise_server_exceptions=True) + + +class TestAdminHome: + """Tests for admin home page.""" + + def test_admin_home_enabled(self, admin_client): + """Test admin home page when enabled.""" + response = admin_client.get("/a/") + assert response.status_code == 200 + assert "Admin" in response.text + assert "Node Tags" in response.text + + def test_admin_home_disabled(self, admin_client_disabled): + """Test admin home page when disabled.""" + response = admin_client_disabled.get("/a/") + assert response.status_code == 404 + + +class TestAdminNodeTags: + """Tests for admin node tags page.""" + + def test_node_tags_page_no_selection(self, admin_client): + """Test node tags page without selecting a node.""" + response = admin_client.get("/a/node-tags") + assert response.status_code == 200 + assert "Node Tags" in response.text + assert "Select a Node" in response.text + # Should show node dropdown + assert "Node One" in response.text + assert "Node Two" in response.text + + def test_node_tags_page_with_selection(self, admin_client): + """Test node tags page with a node selected.""" + response = admin_client.get( + "/a/node-tags?public_key=abc123def456abc123def456abc123de" + ) + assert response.status_code == 200 + assert "Node Tags" in response.text + # Should show the selected node's tags + assert "environment" in response.text + assert "production" in response.text + assert "location" in response.text + assert "building-a" in response.text + + def test_node_tags_page_disabled(self, admin_client_disabled): + """Test node tags page when admin is disabled.""" + response = admin_client_disabled.get("/a/node-tags") + assert response.status_code == 404 + + def test_node_tags_page_with_message(self, admin_client): + """Test node tags page displays success message.""" + response = admin_client.get( + "/a/node-tags?public_key=abc123def456abc123def456abc123de" + "&message=Tag%20created%20successfully" + ) + assert response.status_code == 200 + assert "Tag created successfully" in response.text + + def test_node_tags_page_with_error(self, admin_client): + """Test node tags page displays error message.""" + response = admin_client.get( + "/a/node-tags?public_key=abc123def456abc123def456abc123de" + "&error=Tag%20already%20exists" + ) + assert response.status_code == 200 + assert "Tag already exists" in response.text + + +class TestAdminCreateTag: + """Tests for creating node tags.""" + + def test_create_tag_success(self, admin_client): + """Test creating a new tag.""" + response = admin_client.post( + "/a/node-tags", + data={ + "public_key": "abc123def456abc123def456abc123de", + "key": "new_tag", + "value": "new_value", + "value_type": "string", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + assert "message=" in response.headers["location"] + assert "created" in response.headers["location"] + + def test_create_tag_disabled(self, admin_client_disabled): + """Test creating tag when admin is disabled.""" + response = admin_client_disabled.post( + "/a/node-tags", + data={ + "public_key": "abc123def456abc123def456abc123de", + "key": "new_tag", + "value": "new_value", + "value_type": "string", + }, + follow_redirects=False, + ) + assert response.status_code == 404 + + +class TestAdminUpdateTag: + """Tests for updating node tags.""" + + def test_update_tag_success(self, admin_client): + """Test updating a tag.""" + response = admin_client.post( + "/a/node-tags/update", + data={ + "public_key": "abc123def456abc123def456abc123de", + "key": "environment", + "value": "staging", + "value_type": "string", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + assert "message=" in response.headers["location"] + assert "updated" in response.headers["location"] + + +class TestAdminMoveTag: + """Tests for moving node tags.""" + + def test_move_tag_success(self, admin_client): + """Test moving a tag to another node.""" + response = admin_client.post( + "/a/node-tags/move", + data={ + "public_key": "abc123def456abc123def456abc123de", + "key": "environment", + "new_public_key": "xyz789xyz789xyz789xyz789xyz789xy", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + # Should redirect to destination node + assert "xyz789xyz789xyz789xyz789xyz789xy" in response.headers["location"] + assert "message=" in response.headers["location"] + assert "moved" in response.headers["location"] + + +class TestAdminDeleteTag: + """Tests for deleting node tags.""" + + def test_delete_tag_success(self, admin_client): + """Test deleting a tag.""" + response = admin_client.post( + "/a/node-tags/delete", + data={ + "public_key": "abc123def456abc123def456abc123de", + "key": "environment", + }, + follow_redirects=False, + ) + assert response.status_code == 303 + assert "message=" in response.headers["location"] + assert "deleted" in response.headers["location"]