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 %}
+
+
+
+{% if message %}
+
+{% endif %}
+
+{% if error %}
+
+{% endif %}
+
+
+
+
+{% 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 %}
+
+
+
+
+ | Key |
+ Value |
+ Type |
+ Updated |
+ Actions |
+
+
+
+ {% for tag in tags %}
+
+ | {{ tag.key }} |
+ {{ tag.value or '-' }} |
+
+ {{ tag.value_type }}
+ |
+ {{ tag.updated_at[:10] if tag.updated_at else '-' }} |
+
+
+
+
+
+
+ |
+
+ {% endfor %}
+
+
+
+ {% else %}
+
+
No tags found for this node.
+
Add a new tag below.
+
+ {% endif %}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+{% 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"]