feat(portal): implement dynamic drawer/page navigation management

Replace hardcoded sidebar drawer configuration with admin-manageable
dynamic system. Extend page_status.json with drawer definitions and
page assignments, add drawer CRUD API endpoints, render portal sidebar
via Jinja2 loops, and extend /admin/pages UI with drawer management.
Fix multi-worker cache invalidation via mtime-based staleness detection.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
egg
2026-02-09 11:34:04 +08:00
parent 706c8ba52c
commit 9b1d2edc52
20 changed files with 2269 additions and 735 deletions

View File

@@ -280,8 +280,8 @@ class TestPermissionMiddleware:
assert response.status_code == 200
class TestAdminAPI:
"""Tests for admin API endpoints."""
class TestAdminAPI:
"""Tests for admin API endpoints."""
def test_get_pages_without_login(self, client):
"""Test get pages API requires login."""
@@ -289,22 +289,64 @@ class TestAdminAPI:
# Should redirect
assert response.status_code == 302
def test_get_pages_with_login(self, client):
"""Test get pages API with login."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
def test_get_pages_with_login(self, client):
"""Test get pages API with login."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.get("/admin/api/pages")
assert response.status_code == 200
data = json.loads(response.data)
assert data["success"] is True
assert "pages" in data
def test_update_page_status(self, client, temp_page_status):
"""Test updating page status via API."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
data = json.loads(response.data)
assert data["success"] is True
assert "pages" in data
def test_get_drawers_without_login(self, client):
"""Test drawer API requires login."""
response = client.get("/admin/api/drawers", follow_redirects=False)
assert response.status_code == 302
def test_mutate_drawers_without_login(self, client):
"""Test drawer mutations require login."""
response = client.post(
"/admin/api/drawers",
data=json.dumps({"name": "Unauthorized Drawer"}),
content_type="application/json",
follow_redirects=False,
)
assert response.status_code in (302, 401)
response = client.delete("/admin/api/drawers/reports", follow_redirects=False)
assert response.status_code == 302
def test_get_drawers_with_login(self, client):
"""Test list drawers API with login."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.get("/admin/api/drawers")
assert response.status_code == 200
data = json.loads(response.data)
assert data["success"] is True
assert "drawers" in data
assert any(drawer["id"] == "reports" for drawer in data["drawers"])
def test_create_drawer_duplicate_name_conflict(self, client):
"""Test creating duplicate drawer name returns 409."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.post(
"/admin/api/drawers",
data=json.dumps({"name": "報表類", "order": 99}),
content_type="application/json",
)
assert response.status_code == 409
def test_update_page_status(self, client, temp_page_status):
"""Test updating page status via API."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.put(
"/admin/api/pages/wip-overview",
@@ -329,10 +371,48 @@ class TestAdminAPI:
"/admin/api/pages/wip-overview",
data=json.dumps({"status": "invalid"}),
content_type="application/json"
)
assert response.status_code == 400
)
assert response.status_code == 400
def test_update_page_drawer_assignment(self, client):
"""Test assigning page drawer via page update API."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.put(
"/admin/api/pages/wip-overview",
data=json.dumps({"drawer_id": "queries", "order": 3}),
content_type="application/json",
)
assert response.status_code == 200
page_registry._cache = None
pages = page_registry.get_all_pages()
page = next(item for item in pages if item["route"] == "/wip-overview")
assert page["drawer_id"] == "queries"
assert page["order"] == 3
def test_update_page_invalid_drawer_assignment(self, client):
"""Test assigning a non-existent drawer returns bad request."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.put(
"/admin/api/pages/wip-overview",
data=json.dumps({"drawer_id": "missing-drawer"}),
content_type="application/json",
)
assert response.status_code == 400
def test_delete_drawer_with_assigned_pages_conflict(self, client):
"""Test deleting a non-empty drawer returns conflict."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin"}
response = client.delete("/admin/api/drawers/reports")
assert response.status_code == 409
class TestContextProcessor:
"""Tests for template context processor."""

View File

@@ -1,194 +1,243 @@
# -*- coding: utf-8 -*-
"""Unit tests for page_registry module."""
import json
import pytest
import tempfile
from pathlib import Path
from unittest.mock import patch
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from mes_dashboard.services import page_registry
@pytest.fixture
def temp_data_file(tmp_path):
"""Create a temporary data file for testing."""
data_file = tmp_path / "page_status.json"
initial_data = {
"pages": [
{"route": "/", "name": "Home", "status": "released"},
{"route": "/dev-page", "name": "Dev Page", "status": "dev"},
],
"api_public": True
}
data_file.write_text(json.dumps(initial_data), encoding="utf-8")
return data_file
@pytest.fixture
def mock_registry(temp_data_file):
"""Mock page_registry to use temp file."""
original_data_file = page_registry.DATA_FILE
original_cache = page_registry._cache
page_registry.DATA_FILE = temp_data_file
page_registry._cache = None # Clear cache
yield temp_data_file
# Restore original
page_registry.DATA_FILE = original_data_file
page_registry._cache = original_cache
class TestGetPageStatus:
"""Tests for get_page_status function."""
def test_get_released_page_status(self, mock_registry):
"""Test getting status of released page."""
status = page_registry.get_page_status("/")
assert status == "released"
def test_get_dev_page_status(self, mock_registry):
"""Test getting status of dev page."""
status = page_registry.get_page_status("/dev-page")
assert status == "dev"
def test_get_unregistered_page_status(self, mock_registry):
"""Test getting status of unregistered page returns None."""
status = page_registry.get_page_status("/not-registered")
assert status is None
class TestIsPageRegistered:
"""Tests for is_page_registered function."""
def test_registered_page(self, mock_registry):
"""Test checking registered page."""
assert page_registry.is_page_registered("/") is True
def test_unregistered_page(self, mock_registry):
"""Test checking unregistered page."""
assert page_registry.is_page_registered("/not-here") is False
class TestSetPageStatus:
"""Tests for set_page_status function."""
def test_update_existing_page(self, mock_registry):
"""Test updating existing page status."""
page_registry.set_page_status("/", "dev")
assert page_registry.get_page_status("/") == "dev"
def test_add_new_page(self, mock_registry):
"""Test adding new page."""
page_registry.set_page_status("/new-page", "released", "New Page")
assert page_registry.get_page_status("/new-page") == "released"
def test_invalid_status_raises_error(self, mock_registry):
"""Test setting invalid status raises ValueError."""
with pytest.raises(ValueError, match="Invalid status"):
page_registry.set_page_status("/", "invalid")
def test_update_page_name(self, mock_registry):
"""Test updating page name."""
page_registry.set_page_status("/", "released", "New Name")
pages = page_registry.get_all_pages()
home = next(p for p in pages if p["route"] == "/")
assert home["name"] == "New Name"
class TestGetAllPages:
"""Tests for get_all_pages function."""
def test_get_all_pages(self, mock_registry):
"""Test getting all pages."""
pages = page_registry.get_all_pages()
assert len(pages) == 2
routes = [p["route"] for p in pages]
assert "/" in routes
assert "/dev-page" in routes
class TestIsApiPublic:
"""Tests for is_api_public function."""
def test_api_public_true(self, mock_registry):
"""Test API public flag when true."""
assert page_registry.is_api_public() is True
def test_api_public_false(self, mock_registry, temp_data_file):
"""Test API public flag when false."""
data = json.loads(temp_data_file.read_text())
data["api_public"] = False
temp_data_file.write_text(json.dumps(data))
page_registry._cache = None # Clear cache
assert page_registry.is_api_public() is False
class TestReloadCache:
"""Tests for reload_cache function."""
def test_reload_cache(self, mock_registry, temp_data_file):
"""Test reloading cache from disk."""
# First load
assert page_registry.get_page_status("/") == "released"
# Modify file directly
data = json.loads(temp_data_file.read_text())
data["pages"][0]["status"] = "dev"
temp_data_file.write_text(json.dumps(data))
# Cache still has old value
assert page_registry.get_page_status("/") == "released"
# After reload, should have new value
page_registry.reload_cache()
assert page_registry.get_page_status("/") == "dev"
class TestConcurrency:
"""Tests for thread safety."""
def test_concurrent_access(self, mock_registry):
"""Test concurrent read/write operations."""
import threading
errors = []
def reader():
try:
for _ in range(100):
page_registry.get_page_status("/")
except Exception as e:
errors.append(e)
def writer():
try:
for i in range(100):
status = "released" if i % 2 == 0 else "dev"
page_registry.set_page_status("/", status)
except Exception as e:
errors.append(e)
threads = [
threading.Thread(target=reader) for _ in range(3)
] + [
threading.Thread(target=writer) for _ in range(2)
]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(errors) == 0, f"Errors occurred: {errors}"
if __name__ == "__main__":
pytest.main([__file__, "-v"])
# -*- coding: utf-8 -*-
"""Unit tests for page_registry module."""
import json
import os
import sys
import threading
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
from mes_dashboard.services import page_registry
@pytest.fixture
def temp_data_file(tmp_path):
"""Create a temporary legacy data file for migration tests."""
data_file = tmp_path / "page_status.json"
initial_data = {
"pages": [
{"route": "/", "name": "Home", "status": "released"},
{"route": "/wip-overview", "name": "WIP Overview", "status": "released"},
{"route": "/tables", "name": "Tables", "status": "dev"},
{"route": "/dev-page", "name": "Dev Page", "status": "dev"},
],
"api_public": True,
}
data_file.write_text(json.dumps(initial_data), encoding="utf-8")
return data_file
@pytest.fixture
def mock_registry(temp_data_file):
"""Mock page_registry to use temp file."""
original_data_file = page_registry.DATA_FILE
original_cache = page_registry._cache
page_registry.DATA_FILE = temp_data_file
page_registry._cache = None
yield temp_data_file
page_registry.DATA_FILE = original_data_file
page_registry._cache = original_cache
class TestSchemaMigration:
"""Tests for first-run drawers migration."""
def test_migration_adds_drawers_and_assignments(self, mock_registry):
drawers = page_registry.get_all_drawers()
drawer_ids = [drawer["id"] for drawer in drawers]
assert drawer_ids == ["reports", "queries", "dev-tools"]
pages = page_registry.get_all_pages()
page_by_route = {page["route"]: page for page in pages}
assert page_by_route["/wip-overview"]["drawer_id"] == "reports"
assert page_by_route["/wip-overview"]["order"] == 1
assert page_by_route["/tables"]["drawer_id"] == "queries"
assert page_by_route["/tables"]["order"] == 1
# Admin tools should be backfilled from legacy hardcoded sidebar mapping.
assert page_by_route["/admin/pages"]["drawer_id"] == "dev-tools"
assert page_by_route["/admin/performance"]["drawer_id"] == "dev-tools"
def test_subsequent_load_does_not_reset_drawers(self, mock_registry):
page_registry.get_all_drawers()
page_registry.create_drawer("custom", order=10, admin_only=False)
page_registry.reload_cache()
drawers = page_registry.get_all_drawers()
assert any(drawer["id"] == "custom" for drawer in drawers)
class TestGetPageStatus:
"""Tests for get_page_status function."""
def test_get_released_page_status(self, mock_registry):
status = page_registry.get_page_status("/")
assert status == "released"
def test_get_dev_page_status(self, mock_registry):
status = page_registry.get_page_status("/dev-page")
assert status == "dev"
def test_get_unregistered_page_status(self, mock_registry):
status = page_registry.get_page_status("/not-registered")
assert status is None
class TestSetPageStatus:
"""Tests for set_page_status function."""
def test_update_existing_page_status(self, mock_registry):
page_registry.set_page_status("/dev-page", "released")
assert page_registry.get_page_status("/dev-page") == "released"
def test_set_page_drawer_and_order(self, mock_registry):
page_registry.set_page_status("/dev-page", "dev", drawer_id="queries", order=9)
pages = page_registry.get_all_pages()
dev_page = next(page for page in pages if page["route"] == "/dev-page")
assert dev_page["drawer_id"] == "queries"
assert dev_page["order"] == 9
def test_clear_page_drawer_and_order(self, mock_registry):
page_registry.set_page_status("/dev-page", "dev", drawer_id="queries", order=9)
page_registry.set_page_status("/dev-page", "dev", drawer_id=None, order=None)
pages = page_registry.get_all_pages()
dev_page = next(page for page in pages if page["route"] == "/dev-page")
assert "drawer_id" not in dev_page
assert "order" not in dev_page
def test_set_invalid_drawer_raises_error(self, mock_registry):
with pytest.raises(ValueError, match="Drawer not found"):
page_registry.set_page_status("/dev-page", "dev", drawer_id="not-exists")
def test_invalid_status_raises_error(self, mock_registry):
with pytest.raises(ValueError, match="Invalid status"):
page_registry.set_page_status("/", "invalid")
class TestDrawerCrud:
"""Tests for drawer CRUD functions."""
def test_create_drawer(self, mock_registry):
created = page_registry.create_drawer("Custom Drawer", order=4, admin_only=True)
assert created["name"] == "Custom Drawer"
assert created["order"] == 4
assert created["admin_only"] is True
def test_create_duplicate_drawer_name_raises_conflict(self, mock_registry):
with pytest.raises(page_registry.DrawerConflictError):
page_registry.create_drawer("報表類", order=4)
def test_update_drawer(self, mock_registry):
updated = page_registry.update_drawer(
"reports",
name="報表中心",
order=7,
admin_only=True,
)
assert updated["name"] == "報表中心"
assert updated["order"] == 7
assert updated["admin_only"] is True
def test_delete_drawer_rejects_assigned_pages(self, mock_registry):
with pytest.raises(page_registry.DrawerConflictError, match="assigned pages"):
page_registry.delete_drawer("reports")
def test_delete_empty_drawer(self, mock_registry):
created = page_registry.create_drawer("Temporary", order=8)
page_registry.delete_drawer(created["id"])
drawers = page_registry.get_all_drawers()
assert all(drawer["id"] != created["id"] for drawer in drawers)
class TestNavigationConfig:
"""Tests for navigation config generation."""
def test_navigation_config_grouped_and_sorted(self, mock_registry):
page_registry.set_page_status("/dev-page", "dev", drawer_id="queries", order=5)
nav = page_registry.get_navigation_config()
assert [drawer["id"] for drawer in nav] == ["reports", "queries", "dev-tools"]
reports = next(drawer for drawer in nav if drawer["id"] == "reports")
assert [page["route"] for page in reports["pages"]] == ["/wip-overview"]
assert reports["pages"][0]["frame_id"] == "wipOverviewFrame"
assert reports["pages"][0]["tool_src"] is None
queries = next(drawer for drawer in nav if drawer["id"] == "queries")
assert queries["pages"][0]["route"] == "/tables"
assert queries["pages"][-1]["route"] == "/dev-page"
dev_tools = next(drawer for drawer in nav if drawer["id"] == "dev-tools")
assert all(page["frame_id"] == "toolFrame" for page in dev_tools["pages"])
assert dev_tools["pages"][0]["tool_src"] == "/admin/pages"
class TestIsApiPublic:
"""Tests for is_api_public function."""
def test_api_public_true(self, mock_registry):
assert page_registry.is_api_public() is True
def test_api_public_false(self, mock_registry, temp_data_file):
data = json.loads(temp_data_file.read_text())
data["api_public"] = False
temp_data_file.write_text(json.dumps(data))
page_registry._cache = None
assert page_registry.is_api_public() is False
class TestReloadCache:
"""Tests for reload_cache function."""
def test_reload_cache(self, mock_registry, temp_data_file):
assert page_registry.get_page_status("/") == "released"
data = json.loads(temp_data_file.read_text())
home = next(page for page in data["pages"] if page["route"] == "/")
home["status"] = "dev"
temp_data_file.write_text(json.dumps(data))
assert page_registry.get_page_status("/") == "released"
page_registry.reload_cache()
assert page_registry.get_page_status("/") == "dev"
class TestConcurrency:
"""Tests for thread safety."""
def test_concurrent_access(self, mock_registry):
errors = []
def reader():
try:
for _ in range(100):
page_registry.get_page_status("/")
except Exception as exc: # pragma: no cover - defensive
errors.append(exc)
def writer():
try:
for index in range(100):
status = "released" if index % 2 == 0 else "dev"
page_registry.set_page_status("/", status)
except Exception as exc: # pragma: no cover - defensive
errors.append(exc)
threads = [threading.Thread(target=reader) for _ in range(3)] + [
threading.Thread(target=writer) for _ in range(2)
]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
assert len(errors) == 0, f"Errors occurred: {errors}"

View File

@@ -100,6 +100,108 @@ class TestTemplateIntegration(unittest.TestCase):
self.assertIn('mes-toast-container', html)
class TestPortalDynamicDrawerRendering(unittest.TestCase):
"""Test dynamic portal drawer rendering."""
def setUp(self):
db._ENGINE = None
self.app = create_app('testing')
self.app.config['TESTING'] = True
self.client = self.app.test_client()
_login_as_admin(self.client)
def test_portal_uses_navigation_config_for_sidebar_and_iframes(self):
drawers = [
{
"id": "custom",
"name": "自訂分類",
"order": 1,
"admin_only": False,
"pages": [
{
"route": "/wip-overview",
"name": "自訂首頁",
"status": "released",
"order": 1,
"frame_id": "customFrame",
"tool_src": None,
}
],
},
{
"id": "dev-tools",
"name": "開發工具",
"order": 2,
"admin_only": True,
"pages": [
{
"route": "/admin/pages",
"name": "頁面管理",
"status": "dev",
"order": 1,
"frame_id": "toolFrame",
"tool_src": "/admin/pages",
}
],
},
]
with patch("mes_dashboard.app.get_navigation_config", return_value=drawers):
response = self.client.get("/")
self.assertEqual(response.status_code, 200)
html = response.data.decode("utf-8")
self.assertIn("自訂分類", html)
self.assertIn('data-target="customFrame"', html)
self.assertIn('id="customFrame"', html)
self.assertIn('data-tool-src="/admin/pages"', html)
self.assertIn('id="toolFrame"', html)
def test_portal_hides_admin_only_drawer_for_non_admin(self):
client = self.app.test_client()
drawers = [
{
"id": "custom",
"name": "自訂分類",
"order": 1,
"admin_only": False,
"pages": [
{
"route": "/wip-overview",
"name": "自訂首頁",
"status": "released",
"order": 1,
"frame_id": "customFrame",
"tool_src": None,
}
],
},
{
"id": "dev-tools",
"name": "開發工具",
"order": 2,
"admin_only": True,
"pages": [
{
"route": "/admin/pages",
"name": "頁面管理",
"status": "dev",
"order": 1,
"frame_id": "toolFrame",
"tool_src": "/admin/pages",
}
],
},
]
with patch("mes_dashboard.app.get_navigation_config", return_value=drawers):
response = client.get("/")
self.assertEqual(response.status_code, 200)
html = response.data.decode("utf-8")
self.assertIn("自訂分類", html)
self.assertNotIn("開發工具", html)
self.assertNotIn('data-tool-src="/admin/pages"', html)
class TestToastCSSIntegration(unittest.TestCase):
"""Test that Toast CSS styles are included in pages."""