egg/DashBoard
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: finalize no-iframe portal shell route-view migration

Browse Source
This commit is contained in:
egg
2026-02-11 17:07:50 +08:00
parent ccab10bee8
commit 1e7f8f4498
100 changed files with 8794 additions and 642 deletions
Download Patch File Download Diff File Expand all files Collapse all files

520
tests/e2e/test_global_connection.py
View File

@@ -1,377 +1,173 @@
# -*- coding: utf-8 -*-
"""E2E tests for global connection management features.
Tests the MesApi client, Toast notifications, and page functionality
using Playwright.
Run with: pytest tests/e2e/ --headed (to see browser)
"""
import pytest
import re
from playwright.sync_api import Page, expect
@pytest.mark.e2e
class TestPortalPage:
"""E2E tests for the Portal page."""
def test_portal_loads_successfully(self, page: Page, app_server: str):
"""Portal page should load without errors."""
page.goto(app_server)
# Wait for page to load
expect(page.locator('h1')).to_contain_text('MES 報表入口')
def test_portal_has_all_sidebar_routes(self, page: Page, app_server: str):
"""Portal should expose route-based sidebar entries."""
# -*- coding: utf-8 -*-
"""E2E tests for SPA shell navigation/runtime contracts."""
import pytest
import re
from playwright.sync_api import Page, expect
def _sidebar_links(page: Page):
"""Support both legacy and current shell nav selectors."""
return page.locator("a.drawer-link[href], a.sidebar-item[data-route]")
def _fetch_json_status(page: Page, url: str):
"""Run fetch in browser context and return status/payload metadata."""
return page.evaluate(
"""
async (targetUrl) => {
const response = await fetch(targetUrl, { cache: 'no-store' });
let payload = null;
try {
payload = await response.json();
} catch (_) {
payload = null;
}
return { ok: response.ok, status: response.status, payload };
}
""",
url,
)
@pytest.mark.e2e
class TestPortalPage:
"""E2E tests for portal shell routing and drawer navigation."""
def test_portal_loads_successfully(self, page: Page, app_server: str):
page.goto(app_server)
expect(page.locator("h1")).to_contain_text("MES 報表入口")
def test_portal_has_sidebar_routes(self, page: Page, app_server: str):
page.goto(app_server)
expect(page.locator('.sidebar-item:has-text("WIP 即時概況")')).to_be_visible()
expect(page.locator('.sidebar-item:has-text("設備即時概況")')).to_be_visible()
expect(page.locator('.sidebar-item:has-text("設備歷史績效")')).to_be_visible()
expect(page.locator('.sidebar-item:has-text("設備維修查詢")')).to_be_visible()
expect(_sidebar_links(page).first).to_be_visible()
expect(page.locator(".drawer-link:has-text('WIP 即時概況')")).to_be_visible()
expect(page.locator(".drawer-link:has-text('設備即時概況')")).to_be_visible()
expect(page.locator(".drawer-link:has-text('設備歷史績效')")).to_be_visible()
expect(page.locator(".drawer-link:has-text('設備維修查詢')")).to_be_visible()
def test_portal_sidebar_navigation_uses_direct_routes(self, page: Page, app_server: str):
"""Sidebar click should navigate to direct route without iframe switching."""
page.goto(app_server)
first_route = page.locator('.sidebar-item[data-route]').first
first_route = _sidebar_links(page).first
expect(first_route).to_be_visible()
target_href = first_route.get_attribute('href')
assert target_href and target_href.startswith('/'), "sidebar route href missing"
target_href = first_route.get_attribute("href")
assert target_href, "sidebar route href missing"
first_route.click()
expect(page).to_have_url(re.compile(f".*{re.escape(target_href)}$"))
assert page.locator("iframe").count() == 0, "Shell content must not use iframe"
def test_portal_health_popup_clickable(self, page: Page, app_server: str):
"""Health status pill should toggle popup visibility on click."""
page.goto(app_server)
popup = page.locator('#healthPopup')
expect(popup).not_to_have_class(re.compile(r'show'))
trigger = page.locator(".health-trigger")
expect(trigger).to_be_visible()
expect(page.locator("#shellHealthPopup")).to_have_count(0)
page.locator('#healthStatus').click()
expect(popup).to_have_class(re.compile(r'show'))
@pytest.mark.e2e
class TestToastNotifications:
"""E2E tests for Toast notification system."""
def test_toast_container_exists(self, page: Page, app_server: str):
"""Toast container should be present in the DOM."""
page.goto(f"{app_server}/wip-overview")
# Toast container should exist in DOM (hidden when empty, which is expected)
page.wait_for_selector('#mes-toast-container', state='attached', timeout=5000)
def test_toast_info_display(self, page: Page, app_server: str):
"""Toast.info() should display info notification."""
page.goto(f"{app_server}/wip-overview")
# Execute Toast.info() in browser context
page.evaluate("Toast.info('Test info message')")
# Verify toast appears
toast = page.locator('.mes-toast-info')
expect(toast).to_be_visible()
expect(toast).to_contain_text('Test info message')
def test_toast_success_display(self, page: Page, app_server: str):
"""Toast.success() should display success notification."""
page.goto(f"{app_server}/wip-overview")
page.evaluate("Toast.success('Operation successful')")
toast = page.locator('.mes-toast-success')
expect(toast).to_be_visible()
expect(toast).to_contain_text('Operation successful')
def test_toast_error_display(self, page: Page, app_server: str):
"""Toast.error() should display error notification."""
page.goto(f"{app_server}/wip-overview")
page.evaluate("Toast.error('An error occurred')")
toast = page.locator('.mes-toast-error')
expect(toast).to_be_visible()
expect(toast).to_contain_text('An error occurred')
def test_toast_error_with_retry(self, page: Page, app_server: str):
"""Toast.error() with retry callback should show retry button."""
page.goto(f"{app_server}/wip-overview")
page.evaluate("Toast.error('Connection failed', { retry: () => console.log('retry clicked') })")
# Verify retry button exists
retry_btn = page.locator('.mes-toast-retry')
expect(retry_btn).to_be_visible()
expect(retry_btn).to_contain_text('重試')
def test_toast_loading_display(self, page: Page, app_server: str):
"""Toast.loading() should display loading notification."""
page.goto(f"{app_server}/wip-overview")
page.evaluate("Toast.loading('Loading data...')")
toast = page.locator('.mes-toast-loading')
expect(toast).to_be_visible()
def test_toast_dismiss(self, page: Page, app_server: str):
"""Toast.dismiss() should remove toast."""
page.goto(f"{app_server}/wip-overview")
# Create and dismiss a toast
toast_id = page.evaluate("Toast.info('Will be dismissed')")
page.evaluate(f"Toast.dismiss({toast_id})")
# Wait for animation
page.wait_for_timeout(500)
# Toast should be gone
expect(page.locator('.mes-toast-info')).not_to_be_visible()
def test_toast_max_limit(self, page: Page, app_server: str):
"""Toast system should enforce max 5 toasts."""
page.goto(f"{app_server}/wip-overview")
# Create 7 toasts
for i in range(7):
page.evaluate(f"Toast.info('Toast {i}')")
# Should only have 5 toasts visible
toasts = page.locator('.mes-toast')
expect(toasts).to_have_count(5)
@pytest.mark.e2e
class TestMesApiClient:
"""E2E tests for MesApi client."""
def test_mesapi_exists_on_page(self, page: Page, app_server: str):
"""MesApi should be available in window scope."""
page.goto(f"{app_server}/wip-overview")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined"
def test_mesapi_has_get_method(self, page: Page, app_server: str):
"""MesApi should have get() method."""
page.goto(f"{app_server}/wip-overview")
has_get = page.evaluate("typeof MesApi.get === 'function'")
assert has_get, "MesApi.get should be a function"
def test_mesapi_has_post_method(self, page: Page, app_server: str):
"""MesApi should have post() method."""
page.goto(f"{app_server}/wip-overview")
has_post = page.evaluate("typeof MesApi.post === 'function'")
assert has_post, "MesApi.post should be a function"
def test_mesapi_request_logging(self, page: Page, app_server: str):
"""MesApi should log requests to console."""
page.goto(f"{app_server}/wip-overview")
# Capture console messages
console_messages = []
page.on("console", lambda msg: console_messages.append(msg.text))
# Make a request (will fail but should log)
page.evaluate("""
(async () => {
try {
await MesApi.get('/api/test-endpoint');
} catch (e) {
// Expected to fail
}
})()
""")
page.wait_for_timeout(1000)
# Check for MesApi log pattern
mesapi_logs = [m for m in console_messages if '[MesApi]' in m]
assert len(mesapi_logs) > 0, "MesApi should log requests with [MesApi] prefix"
@pytest.mark.e2e
class TestWIPOverviewPage:
"""E2E tests for WIP Overview page."""
def test_wip_overview_loads(self, page: Page, app_server: str):
"""WIP Overview page should load."""
page.goto(f"{app_server}/wip-overview")
# Page should have the header
expect(page.locator('body')).to_be_visible()
def test_wip_overview_has_toast_system(self, page: Page, app_server: str):
"""WIP Overview should have Toast system loaded."""
page.goto(f"{app_server}/wip-overview")
has_toast = page.evaluate("typeof Toast !== 'undefined'")
assert has_toast, "Toast should be defined on WIP Overview page"
def test_wip_overview_has_mesapi(self, page: Page, app_server: str):
"""WIP Overview should have MesApi loaded."""
page.goto(f"{app_server}/wip-overview")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined on WIP Overview page"
@pytest.mark.e2e
class TestWIPDetailPage:
"""E2E tests for WIP Detail page."""
def test_wip_detail_loads(self, page: Page, app_server: str):
"""WIP Detail page should load."""
page.goto(f"{app_server}/wip-detail")
expect(page.locator('body')).to_be_visible()
def test_wip_detail_has_toast_system(self, page: Page, app_server: str):
"""WIP Detail should have Toast system loaded."""
page.goto(f"{app_server}/wip-detail")
has_toast = page.evaluate("typeof Toast !== 'undefined'")
assert has_toast, "Toast should be defined on WIP Detail page"
def test_wip_detail_has_mesapi(self, page: Page, app_server: str):
"""WIP Detail should have MesApi loaded."""
page.goto(f"{app_server}/wip-detail")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined on WIP Detail page"
@pytest.mark.e2e
class TestTablesPage:
"""E2E tests for Tables page."""
def test_tables_page_loads(self, page: Page, app_server: str):
"""Tables page should load."""
page.goto(f"{app_server}/tables")
header = page.locator('h1')
expect(header).to_be_visible()
text = header.inner_text()
assert (
'MES 數據表查詢工具' in text
or '頁面開發中' in text
trigger.click()
expect(page.locator("#shellHealthPopup")).to_be_visible()
page.keyboard.press("Escape")
expect(page.locator("#shellHealthPopup")).to_have_count(0)
@pytest.mark.e2e
class TestFrontendApiRuntime:
"""E2E tests for runtime API availability in browser context."""
def test_wip_overview_can_call_summary_api_via_fetch(self, page: Page, app_server: str):
page.goto(f"{app_server}/wip-overview")
result = _fetch_json_status(page, "/api/wip/overview/summary")
assert result["ok"] is True
assert result["status"] == 200
assert isinstance(result.get("payload"), dict)
def test_wip_detail_can_call_workcenter_api_via_fetch(self, page: Page, app_server: str):
page.goto(f"{app_server}/wip-detail")
result = _fetch_json_status(page, "/api/wip/meta/workcenters")
assert result["ok"] is True
assert result["status"] == 200
assert isinstance(result.get("payload"), dict)
def test_global_mesapi_bridge_is_optional(self, page: Page, app_server: str):
page.goto(f"{app_server}/wip-overview")
runtime = page.evaluate(
"""
() => ({
hasFetch: typeof window.fetch === 'function',
hasMesApi: typeof window.MesApi !== 'undefined',
hasMesApiGet: Boolean(window.MesApi && typeof window.MesApi.get === 'function'),
})
"""
)
def test_tables_has_toast_system(self, page: Page, app_server: str):
"""Tables page should have Toast system loaded."""
page.goto(f"{app_server}/tables")
has_toast = page.evaluate("typeof Toast !== 'undefined'")
assert has_toast, "Toast should be defined on Tables page"
def test_tables_has_mesapi(self, page: Page, app_server: str):
"""Tables page should have MesApi loaded."""
page.goto(f"{app_server}/tables")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined on Tables page"
@pytest.mark.e2e
class TestResourcePage:
"""E2E tests for Resource Status page."""
def test_resource_page_loads(self, page: Page, app_server: str):
"""Resource page should load."""
page.goto(f"{app_server}/resource")
expect(page.locator('body')).to_be_visible()
def test_resource_has_toast_system(self, page: Page, app_server: str):
"""Resource page should have Toast system loaded."""
page.goto(f"{app_server}/resource")
has_toast = page.evaluate("typeof Toast !== 'undefined'")
assert has_toast, "Toast should be defined on Resource page"
def test_resource_has_mesapi(self, page: Page, app_server: str):
"""Resource page should have MesApi loaded."""
page.goto(f"{app_server}/resource")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined on Resource page"
@pytest.mark.e2e
class TestExcelQueryPage:
"""E2E tests for Excel Query page."""
def test_excel_query_page_loads(self, page: Page, app_server: str):
"""Excel Query page should load."""
page.goto(f"{app_server}/excel-query")
expect(page.locator('body')).to_be_visible()
def test_excel_query_has_toast_system(self, page: Page, app_server: str):
"""Excel Query page should have Toast system loaded."""
page.goto(f"{app_server}/excel-query")
has_toast = page.evaluate("typeof Toast !== 'undefined'")
assert has_toast, "Toast should be defined on Excel Query page"
def test_excel_query_has_mesapi(self, page: Page, app_server: str):
"""Excel Query page should have MesApi loaded."""
page.goto(f"{app_server}/excel-query")
has_mesapi = page.evaluate("typeof MesApi !== 'undefined'")
assert has_mesapi, "MesApi should be defined on Excel Query page"
@pytest.mark.e2e
class TestConsoleLogVerification:
"""E2E tests for console log verification (Phase 4.2 tasks)."""
def test_request_has_request_id(self, page: Page, app_server: str):
"""API requests should log with req_xxx ID format."""
page.goto(f"{app_server}/wip-overview")
console_messages = []
page.on("console", lambda msg: console_messages.append(msg.text))
# Trigger an API request
page.evaluate("""
(async () => {
try {
await MesApi.get('/api/wip/overview/summary');
} catch (e) {}
})()
""")
page.wait_for_timeout(2000)
# Check for request ID pattern
req_id_pattern = re.compile(r'req_\d{4}')
has_req_id = any(req_id_pattern.search(m) for m in console_messages)
assert has_req_id, "Console should show request ID like req_0001"
def test_successful_request_shows_checkmark(self, page: Page, app_server: str):
"""Successful requests should show checkmark in console."""
page.goto(f"{app_server}/wip-overview")
console_messages = []
page.on("console", lambda msg: console_messages.append(msg.text))
# Make request to a working endpoint
page.evaluate("""
(async () => {
try {
await MesApi.get('/api/wip/overview/summary');
} catch (e) {}
})()
""")
page.wait_for_timeout(3000)
# Filter for MesApi logs
mesapi_logs = [m for m in console_messages if '[MesApi]' in m]
# The exact checkmark depends on implementation (✓ or similar)
assert len(mesapi_logs) > 0, "Should have MesApi console logs"
assert runtime["hasFetch"] is True
if runtime["hasMesApi"]:
assert runtime["hasMesApiGet"] is True
@pytest.mark.e2e
class TestRoutePagesSmoke:
"""Basic smoke checks for key route pages."""
def test_wip_overview_loads(self, page: Page, app_server: str):
response = page.goto(f"{app_server}/wip-overview")
assert response is not None and response.ok
expect(page.locator("body")).to_be_visible()
def test_wip_detail_loads(self, page: Page, app_server: str):
response = page.goto(f"{app_server}/wip-detail")
assert response is not None and response.ok
expect(page.locator("body")).to_be_visible()
def test_resource_page_loads(self, page: Page, app_server: str):
response = page.goto(f"{app_server}/resource")
assert response is not None and response.ok
expect(page.locator("body")).to_be_visible()
def test_tables_page_loads(self, page: Page, app_server: str):
response = page.goto(f"{app_server}/tables")
assert response is not None
assert response.status in {200, 403}
expect(page.locator("body")).to_be_visible()
if response.status == 200:
header = page.locator("h1")
expect(header).to_be_visible()
text = header.inner_text()
assert "MES 數據表查詢工具" in text or "頁面開發中" in text
def test_excel_query_page_loads(self, page: Page, app_server: str):
response = page.goto(f"{app_server}/excel-query")
assert response is not None
assert response.status in {200, 403}
expect(page.locator("body")).to_be_visible()
@pytest.mark.e2e
class TestConsoleAndErrorSignals:
"""Console/pageerror checks for SPA runtime stability."""
def test_wip_overview_has_no_uncaught_page_errors(self, page: Page, app_server: str):
errors = []
page.on("pageerror", lambda error: errors.append(str(error)))
page.goto(f"{app_server}/wip-overview")
page.wait_for_timeout(2000)
assert errors == [], f"Unexpected page errors: {errors[:3]}"
def test_wip_overview_triggers_expected_api_requests(self, page: Page, app_server: str):
observed = set()
def on_response(resp):
if "/api/wip/overview/summary" in resp.url:
observed.add("summary")
if "/api/wip/overview/matrix" in resp.url:
observed.add("matrix")
page.on("response", on_response)
page.goto(f"{app_server}/wip-overview", wait_until="domcontentloaded")
page.wait_for_timeout(5000)
assert "summary" in observed
assert "matrix" in observed

65
tests/e2e/test_wip_hold_pages_e2e.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import time
from urllib.parse import parse_qs, quote, urlparse
import re
import pytest
import requests
@@ -166,3 +167,67 @@ class TestWipAndHoldPagesE2E:
page.wait_for_timeout(200)
assert seen == {"summary", "distribution", "lots"}
def test_portal_shell_deep_links_keep_detail_routes(self, page: Page, app_server: str):
workcenter = _pick_workcenter(app_server)
page.goto(
f"{app_server}/portal-shell/wip-detail?workcenter={quote(workcenter)}&status=queue",
wait_until="commit",
timeout=60000,
)
expect(page).to_have_url(re.compile(r".*/portal-shell/wip-detail\\?.*workcenter=.*"))
detail_response = _wait_for_response(
page,
lambda resp: (
"/api/wip/detail/" in resp.url
and parse_qs(urlparse(resp.url).query).get("status", [None])[0] in {"QUEUE", "queue"}
),
timeout_seconds=30.0,
)
assert detail_response is not None
assert detail_response.ok
reason = _pick_hold_reason(app_server)
page.goto(
f"{app_server}/portal-shell/hold-detail?reason={quote(reason)}",
wait_until="commit",
timeout=60000,
)
expect(page).to_have_url(re.compile(r".*/portal-shell/hold-detail\\?.*reason=.*"))
summary_response = _wait_for_response(
page,
lambda resp: (
"/api/wip/hold-detail/summary" in resp.url
and parse_qs(urlparse(resp.url).query).get("reason", [None])[0] == reason
),
timeout_seconds=30.0,
)
assert summary_response is not None
assert summary_response.ok
def test_portal_shell_wip_overview_drilldown_routes_to_detail_pages(self, page: Page, app_server: str):
page.goto(
f"{app_server}/portal-shell/wip-overview",
wait_until="commit",
timeout=60000,
)
page.wait_for_timeout(3000)
matrix_links = page.locator("td.clickable")
if matrix_links.count() == 0:
pytest.skip("No matrix rows available for WIP drilldown")
matrix_links.first.click()
expect(page).to_have_url(re.compile(r".*/portal-shell/wip-detail\\?.*workcenter=.*"))
page.goto(
f"{app_server}/portal-shell/wip-overview",
wait_until="commit",
timeout=60000,
)
page.wait_for_timeout(3000)
reason_links = page.locator("a.reason-link")
if reason_links.count() == 0:
pytest.skip("No pareto reason links available for HOLD drilldown")
reason_links.first.click()
expect(page).to_have_url(re.compile(r".*/portal-shell/hold-detail\\?.*reason=.*"))

23
tests/stress/test_frontend_stress.py
View File

@@ -15,9 +15,9 @@ import time
import requests
from urllib.parse import quote
from playwright.sync_api import Page, expect
@pytest.fixture(scope="session")
@pytest.fixture(scope="session")
def app_server() -> str:
"""Get the base URL for stress testing."""
import os
@@ -34,10 +34,15 @@ def browser_context_args(browser_context_args):
}
def load_page_with_js(page: Page, url: str, timeout: int = 60000):
"""Load page and wait for JS to initialize."""
page.goto(url, wait_until='domcontentloaded', timeout=timeout)
page.wait_for_timeout(1000) # Allow JS initialization
def load_page_with_js(page: Page, url: str, timeout: int = 60000):
"""Load page and wait for JS to initialize."""
page.goto(url, wait_until='domcontentloaded', timeout=timeout)
page.wait_for_timeout(1000) # Allow JS initialization
def locate_portal_nav_links(page: Page):
"""Locate portal navigation links across legacy/new shell DOM contracts."""
return page.locator('.drawer-link[href], .sidebar-item[data-route]')
@pytest.mark.stress
@@ -264,7 +269,7 @@ class TestPageNavigationStress:
def test_rapid_route_switching(self, page: Page, app_server: str):
"""Rapid direct-route switching should remain responsive."""
page.goto(app_server, wait_until='domcontentloaded', timeout=30000)
sidebar_items = page.locator('.sidebar-item[data-route]')
sidebar_items = locate_portal_nav_links(page)
expect(sidebar_items.first).to_be_visible()
item_count = sidebar_items.count()
assert item_count >= 1, "No portal sidebar routes available for stress test"
@@ -294,7 +299,7 @@ class TestPageNavigationStress:
def test_portal_navigation_contract_without_iframe(self, page: Page, app_server: str):
"""Portal sidebar should expose route metadata and no iframe DOM."""
page.goto(app_server, wait_until='domcontentloaded', timeout=30000)
sidebar_items = page.locator('.sidebar-item[data-route]')
sidebar_items = locate_portal_nav_links(page)
expect(sidebar_items.first).to_be_visible()
assert sidebar_items.count() >= 1, "No route sidebar items found"

6
tests/test_app_factory.py
View File

@@ -87,7 +87,8 @@ class AppFactoryTests(unittest.TestCase):
client = app.test_client()
response = client.get("/", follow_redirects=False)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.headers.get("Location"), "/portal-shell")
location = response.headers.get("Location", "")
self.assertTrue(location.startswith("/portal-shell"))
finally:
if old is not None:
os.environ["PORTAL_SPA_ENABLED"] = old
@@ -119,7 +120,8 @@ class AppFactoryTests(unittest.TestCase):
client = app.test_client()
response = client.get("/", follow_redirects=False)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.headers.get("Location"), "/portal-shell")
location = response.headers.get("Location", "")
self.assertTrue(location.startswith("/portal-shell"))
finally:
if old is None:
os.environ.pop("PORTAL_SPA_ENABLED", None)

45
tests/test_auth_integration.py
View File

@@ -414,27 +414,30 @@ class TestAdminAPI:
assert response.status_code == 409
class TestContextProcessor:
"""Tests for template context processor."""
def test_is_admin_in_context_when_logged_in(self, client):
"""Test is_admin is True in context when logged in."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin", "displayName": "Admin"}
response = client.get("/")
content = response.data.decode("utf-8")
# Should show admin-related content (logout link, etc.)
assert "登出" in content or "logout" in content.lower() or "Admin" in content
def test_is_admin_in_context_when_not_logged_in(self, client):
"""Test is_admin is False in context when not logged in."""
response = client.get("/")
content = response.data.decode("utf-8")
# Should show login link, not logout
assert "管理員登入" in content or "login" in content.lower()
class TestContextProcessor:
"""Tests for SPA shell auth context surface."""
def test_is_admin_in_context_when_logged_in(self, client):
"""Test navigation API exposes admin context when logged in."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin", "displayName": "Admin"}
response = client.get("/api/portal/navigation")
assert response.status_code == 200
payload = response.get_json()
assert payload["is_admin"] is True
assert payload["admin_user"]["username"] == "admin"
assert payload["admin_links"]["logout"] == "/admin/logout"
def test_is_admin_in_context_when_not_logged_in(self, client):
"""Test navigation API hides admin context when not logged in."""
response = client.get("/api/portal/navigation")
assert response.status_code == 200
payload = response.get_json()
assert payload["is_admin"] is False
assert payload["admin_user"] is None
assert payload["admin_links"]["logout"] is None
assert payload["admin_links"]["login"].startswith("/admin/login?next=")
if __name__ == "__main__":

150
tests/test_cutover_gates.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""Cutover gate enforcement tests for portal no-iframe migration."""
"""Cutover gate enforcement tests for portal shell route-view migration."""
from __future__ import annotations
@@ -9,14 +9,24 @@ from pathlib import Path
from mes_dashboard.app import create_app
ROOT = Path(__file__).resolve().parents[1]
BASELINE_VISIBILITY_FILE = ROOT / "docs" / "migration" / "portal-no-iframe" / "baseline_drawer_visibility.json"
BASELINE_API_FILE = ROOT / "docs" / "migration" / "portal-no-iframe" / "baseline_api_payload_contracts.json"
ROLLBACK_RUNBOOK = ROOT / "docs" / "migration" / "portal-no-iframe" / "rollback_rehearsal_runbook.md"
ROLLBACK_STRATEGY = ROOT / "docs" / "migration" / "portal-no-iframe" / "rollback_strategy_shell_and_wrappers.md"
LEGACY_REWRITE_SMOKE_CHECKLIST = ROOT / "docs" / "migration" / "portal-no-iframe" / "legacy_rewrite_smoke_checklists.md"
BASELINE_DIR = ROOT / "docs" / "migration" / "portal-shell-route-view-integration"
BASELINE_VISIBILITY_FILE = BASELINE_DIR / "baseline_drawer_visibility.json"
BASELINE_API_FILE = BASELINE_DIR / "baseline_api_payload_contracts.json"
GATE_REPORT_FILE = BASELINE_DIR / "cutover-gates-report.json"
WAVE_A_EVIDENCE_FILE = BASELINE_DIR / "wave-a-smoke-evidence.json"
WAVE_B_EVIDENCE_FILE = BASELINE_DIR / "wave-b-native-smoke-evidence.json"
WAVE_B_PARITY_FILE = BASELINE_DIR / "wave-b-parity-evidence.json"
VISUAL_SNAPSHOT_FILE = BASELINE_DIR / "visual-regression-snapshots.json"
ROLLBACK_RUNBOOK = BASELINE_DIR / "rollback-rehearsal-shell-route-view.md"
KILL_SWITCH_DOC = BASELINE_DIR / "kill-switch-operations.md"
OBSERVABILITY_REPORT = BASELINE_DIR / "migration-observability-report.md"
STRESS_SUITE = ROOT / "tests" / "stress" / "test_frontend_stress.py"
def _read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def _login_as_admin(client) -> None:
with client.session_transaction() as sess:
sess["admin"] = {"displayName": "Admin", "employeeNo": "A001"}
@@ -35,6 +45,7 @@ def test_g1_route_availability_gate_p0_routes_are_2xx_or_3xx():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
_login_as_admin(client)
p0_routes = [
"/",
@@ -43,6 +54,10 @@ def test_g1_route_availability_gate_p0_routes_are_2xx_or_3xx():
"/wip-overview",
"/resource",
"/qc-gate",
"/job-query",
"/excel-query",
"/query-tool",
"/tmtt-defect",
]
statuses = [client.get(route).status_code for route in p0_routes]
@@ -50,56 +65,47 @@ def test_g1_route_availability_gate_p0_routes_are_2xx_or_3xx():
def test_g2_drawer_parity_gate_matches_baseline_for_admin_and_non_admin():
baseline = json.loads(BASELINE_VISIBILITY_FILE.read_text(encoding="utf-8"))
baseline = _read_json(BASELINE_VISIBILITY_FILE)
app = create_app("testing")
app.config["TESTING"] = True
non_admin_client = app.test_client()
non_admin_payload = json.loads(non_admin_client.get("/api/portal/navigation").data.decode("utf-8"))
non_admin_payload = _read_json_response(non_admin_client.get("/api/portal/navigation"))
admin_client = app.test_client()
_login_as_admin(admin_client)
admin_payload = json.loads(admin_client.get("/api/portal/navigation").data.decode("utf-8"))
admin_payload = _read_json_response(admin_client.get("/api/portal/navigation"))
assert _route_set(non_admin_payload["drawers"]) == _route_set(baseline["non_admin"])
assert _route_set(admin_payload["drawers"]) == _route_set(baseline["admin"])
def test_g3_workflow_smoke_gate_critical_routes_reachable_for_admin():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
_login_as_admin(client)
def test_g3_smoke_evidence_gate_requires_wave_a_and_wave_b_pass():
wave_a = _read_json(WAVE_A_EVIDENCE_FILE)
wave_b = _read_json(WAVE_B_EVIDENCE_FILE)
smoke_routes = [
"/",
"/wip-overview",
"/wip-detail?workcenter=TMTT&type=PJA3460&status=queue",
"/hold-detail?reason=YieldLimit",
"/hold-overview",
"/hold-history",
"/resource",
"/resource-history?start_date=2026-01-01&end_date=2026-01-31",
"/qc-gate",
"/job-query",
"/excel-query",
"/query-tool",
"/tmtt-defect",
]
statuses = [client.get(route).status_code for route in smoke_routes]
assert all(200 <= status < 400 for status in statuses), statuses
for payload in (wave_a, wave_b):
assert payload["execution"]["automated_runs"]
for run in payload["execution"]["automated_runs"]:
assert run["status"] == "pass"
for route, result in payload["pages"].items():
assert result["status"] == "pass", f"smoke evidence failed: {route}"
assert result["critical_failures"] == []
def test_g4_client_stability_gate_assertion_present_in_stress_suite():
content = STRESS_SUITE.read_text(encoding="utf-8")
assert 'page.on("pageerror"' in content
assert 'assert len(js_errors) == 0' in content
def test_g4_no_iframe_gate_blocks_if_shell_uses_iframe():
stress_source = STRESS_SUITE.read_text(encoding="utf-8")
assert "Portal should not render iframe after migration" in stress_source
assert "iframe_count = page.locator('iframe').count()" in stress_source
report = _read_json(GATE_REPORT_FILE)
g4 = next(g for g in report["gates"] if g["id"] == "G4")
assert g4["status"] == "pass"
assert g4["block_on_fail"] is True
def test_g5_data_contract_gate_baseline_keys_are_defined_for_registered_apis():
baseline = json.loads(BASELINE_API_FILE.read_text(encoding="utf-8"))
def test_g5_route_query_compatibility_gate_checks_contracts():
baseline = _read_json(BASELINE_API_FILE)
app = create_app("testing")
app.config["TESTING"] = True
registered_routes = {rule.rule for rule in app.url_map.iter_rules()}
@@ -110,21 +116,63 @@ def test_g5_data_contract_gate_baseline_keys_are_defined_for_registered_apis():
assert required_keys, f"No required_keys defined for {api_route}"
assert all(isinstance(key, str) and key for key in required_keys)
report = _read_json(GATE_REPORT_FILE)
g5 = next(g for g in report["gates"] if g["id"] == "G5")
assert g5["status"] == "pass"
assert g5["block_on_fail"] is True
def test_g7_rollback_readiness_gate_has_15_minute_slo_and_operator_steps():
def test_g6_parity_gate_requires_table_chart_filter_interaction_matrix_pass():
parity = _read_json(WAVE_B_PARITY_FILE)
for route, checks in parity["pages"].items():
for dimension in ("table", "chart", "filter", "interaction", "matrix"):
status = checks[dimension]["status"]
assert status in {"pass", "n/a"}, f"{route} parity failed on {dimension}: {status}"
snapshots = _read_json(VISUAL_SNAPSHOT_FILE)
assert snapshots["critical_diff_policy"]["block_release"] is True
assert len(snapshots["snapshots"]) >= 4
report = _read_json(GATE_REPORT_FILE)
g6 = next(g for g in report["gates"] if g["id"] == "G6")
assert g6["status"] == "pass"
assert g6["block_on_fail"] is True
def test_g7_rollback_gate_has_recovery_slo_and_kill_switch_steps():
rehearsal = ROLLBACK_RUNBOOK.read_text(encoding="utf-8")
strategy = ROLLBACK_STRATEGY.read_text(encoding="utf-8")
kill_switch = KILL_SWITCH_DOC.read_text(encoding="utf-8")
assert "15" in rehearsal
assert "PORTAL_SPA_ENABLED=false" in strategy
assert "/api/portal/navigation" in strategy
assert "15 minutes" in rehearsal
assert "PORTAL_SPA_ENABLED=false" in rehearsal
assert "PORTAL_SPA_ENABLED=false" in kill_switch
assert "/api/portal/navigation" in kill_switch
assert "/health" in kill_switch
report = _read_json(GATE_REPORT_FILE)
g7 = next(g for g in report["gates"] if g["id"] == "G7")
assert g7["status"] == "pass"
assert g7["block_on_fail"] is True
def test_legacy_rewrite_smoke_checklist_covers_all_wrapped_pages():
content = LEGACY_REWRITE_SMOKE_CHECKLIST.read_text(encoding="utf-8")
def test_release_block_semantics_enforced_by_gate_report():
report = _read_json(GATE_REPORT_FILE)
assert report["policy"]["block_on_any_failed_gate"] is True
assert report["policy"]["block_on_incomplete_smoke_evidence"] is True
assert report["policy"]["block_on_critical_parity_failure"] is True
assert "tmtt-defect" in content
assert "job-query" in content
assert "excel-query" in content
assert "query-tool" in content
assert "SMOKE-01" in content
for gate in report["gates"]:
assert gate["status"] == "pass"
assert gate["block_on_fail"] is True
assert report["release_blocked"] is False
def test_observability_report_covers_route_errors_health_and_fallback_usage():
content = OBSERVABILITY_REPORT.read_text(encoding="utf-8")
assert "route errors" in content.lower()
assert "health regressions" in content.lower()
assert "wrapper fallback usage" in content.lower()
def _read_json_response(response) -> dict:
return json.loads(response.data.decode("utf-8"))

6
tests/test_health_routes.py
View File

@@ -142,6 +142,9 @@ def test_frontend_shell_health_endpoint_healthy(mock_status):
payload = response.get_json()
assert payload["status"] == "healthy"
assert payload["checks"]["portal_shell_css"]["exists"] is True
assert payload["summary"]["status"] == "healthy"
assert payload["summary"]["error_count"] == 0
assert payload["detail"]["checks"]["portal_shell_css"]["exists"] is True
@patch('mes_dashboard.routes.health_routes.get_portal_shell_asset_status')
@@ -170,6 +173,9 @@ def test_frontend_shell_health_endpoint_unhealthy(mock_status):
payload = response.get_json()
assert payload["status"] == "unhealthy"
assert any("portal-shell.css" in error for error in payload.get("errors", []))
assert payload["summary"]["status"] == "unhealthy"
assert payload["summary"]["error_count"] >= 1
assert any("portal-shell.css" in error for error in payload["detail"].get("errors", []))
def test_get_portal_shell_asset_status_reports_nested_html_as_healthy(tmp_path):

151
tests/test_portal_shell_routes.py
View File

@@ -4,6 +4,7 @@
from __future__ import annotations
import json
from unittest.mock import patch
from mes_dashboard.app import create_app
@@ -28,6 +29,9 @@ def test_portal_shell_fallback_html_served_when_dist_missing(monkeypatch):
assert "/static/dist/portal-shell.css" in html
assert "/static/dist/tailwind.css" in html
response_with_trailing_slash = client.get("/portal-shell/")
assert response_with_trailing_slash.status_code == 200
def test_portal_shell_uses_nested_dist_html_when_top_level_missing(monkeypatch):
app = create_app("testing")
@@ -103,14 +107,17 @@ def test_wrapper_telemetry_endpoint_removed_after_wrapper_decommission():
app.config["TESTING"] = True
client = app.test_client()
response = client.post(
post_response = client.post(
"/api/portal/wrapper-telemetry",
json={
"route": "/job-query",
"event_type": "wrapper_loaded",
"event_type": "wrapper_load_success",
},
)
assert response.status_code == 404
assert post_response.status_code == 404
get_response = client.get("/api/portal/wrapper-telemetry")
assert get_response.status_code == 404
def test_navigation_drawer_and_page_order_deterministic_non_admin():
@@ -129,6 +136,64 @@ def test_navigation_drawer_and_page_order_deterministic_non_admin():
assert reports_routes == ["/wip-overview", "/resource", "/qc-gate"]
def test_navigation_contract_page_metadata_fields_present_and_typed():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
payload = json.loads(client.get("/api/portal/navigation").data.decode("utf-8"))
assert isinstance(payload["drawers"], list)
for drawer in payload["drawers"]:
assert isinstance(drawer["id"], str) and drawer["id"]
assert isinstance(drawer["name"], str) and drawer["name"]
assert isinstance(drawer["order"], int)
assert isinstance(drawer["admin_only"], bool)
assert isinstance(drawer["pages"], list)
for page in drawer["pages"]:
assert isinstance(page["route"], str) and page["route"].startswith("/")
assert isinstance(page["name"], str) and page["name"]
assert page["status"] in {"released", "dev"}
assert isinstance(page["order"], int)
def test_navigation_duplicate_order_values_still_resolve_deterministically():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
config = [
{
"id": "reports",
"name": "Reports",
"order": 1,
"admin_only": False,
"pages": [
{"route": "/qc-gate", "name": "QC", "status": "released", "order": 1},
{"route": "/resource", "name": "Resource", "status": "released", "order": 1},
{"route": "/wip-overview", "name": "WIP", "status": "released", "order": 1},
],
},
{
"id": "tools",
"name": "Tools",
"order": 1,
"admin_only": False,
"pages": [
{"route": "/job-query", "name": "Job", "status": "released", "order": 1},
],
},
]
with patch("mes_dashboard.app.get_navigation_config", return_value=config):
payload = json.loads(client.get("/api/portal/navigation").data.decode("utf-8"))
# Drawer tie breaks by name and page tie breaks by name.
assert [drawer["id"] for drawer in payload["drawers"]] == ["reports", "tools"]
assert [page["route"] for page in payload["drawers"][0]["pages"]] == ["/qc-gate", "/resource", "/wip-overview"]
def test_navigation_mixed_release_dev_visibility_admin_vs_non_admin():
app = create_app("testing")
app.config["TESTING"] = True
@@ -159,7 +224,85 @@ def test_navigation_mixed_release_dev_visibility_admin_vs_non_admin():
assert "/hold-history" in admin_routes
def test_legacy_wrapper_routes_are_reachable():
def test_portal_navigation_includes_admin_links_by_auth_state():
app = create_app("testing")
app.config["TESTING"] = True
non_admin_client = app.test_client()
non_admin_payload = json.loads(non_admin_client.get("/api/portal/navigation").data.decode("utf-8"))
assert non_admin_payload["admin_links"]["login"].startswith("/admin/login?next=")
assert non_admin_payload["admin_links"]["pages"] is None
assert non_admin_payload["admin_links"]["logout"] is None
admin_client = app.test_client()
_login_as_admin(admin_client)
admin_payload = json.loads(admin_client.get("/api/portal/navigation").data.decode("utf-8"))
assert admin_payload["admin_links"]["pages"] == "/admin/pages"
assert admin_payload["admin_links"]["logout"] == "/admin/logout"
def test_portal_navigation_emits_diagnostics_for_invalid_navigation_payload():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
malformed = [
{"id": "", "name": "bad-drawer", "pages": []},
{
"id": "reports",
"name": "Reports",
"order": 1,
"admin_only": False,
"pages": [
{"route": "", "name": "invalid-route"},
{"route": "missing-leading-slash", "name": "invalid-route-2"},
"not-a-dict",
],
},
]
with patch("mes_dashboard.app.get_navigation_config", return_value=malformed):
response = client.get("/api/portal/navigation")
assert response.status_code == 200
payload = json.loads(response.data.decode("utf-8"))
diagnostics = payload["diagnostics"]
assert diagnostics["invalid_drawers"] >= 1
assert diagnostics["invalid_pages"] >= 2
assert payload["drawers"] == []
def test_portal_navigation_logs_contract_mismatch_route():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()
with (
patch("mes_dashboard.app._load_shell_route_contract_routes", return_value={"/wip-overview"}),
patch(
"mes_dashboard.app.get_navigation_config",
return_value=[
{
"id": "reports",
"name": "Reports",
"order": 1,
"admin_only": False,
"pages": [
{"route": "/wip-overview", "name": "WIP", "status": "released", "order": 1},
{"route": "/resource", "name": "Resource", "status": "released", "order": 2},
],
}
],
),
):
response = client.get("/api/portal/navigation")
assert response.status_code == 200
payload = json.loads(response.data.decode("utf-8"))
assert payload["diagnostics"]["contract_mismatch_routes"] == ["/resource"]
def test_wave_b_native_routes_are_reachable():
app = create_app("testing")
app.config["TESTING"] = True
client = app.test_client()

253
tests/test_portal_shell_wave_b_native_smoke.py Normal file
View File

@@ -0,0 +1,253 @@
# -*- coding: utf-8 -*-
"""Wave B native-route smoke coverage for shell migration."""
from __future__ import annotations
import io
from unittest.mock import patch
import pytest
from mes_dashboard.app import create_app
@pytest.fixture
def app():
app = create_app("testing")
app.config["TESTING"] = True
return app
@pytest.fixture
def client(app):
return app.test_client()
def _login_as_admin(client) -> None:
with client.session_transaction() as sess:
sess["admin"] = {"displayName": "Admin", "employeeNo": "A001"}
def _build_excel_file() -> io.BytesIO:
import openpyxl
workbook = openpyxl.Workbook()
sheet = workbook.active
sheet["A1"] = "LOT_ID"
sheet["B1"] = "QTY"
sheet["A2"] = "LOT001"
sheet["B2"] = 100
sheet["A3"] = "LOT002"
sheet["B3"] = 200
buffer = io.BytesIO()
workbook.save(buffer)
buffer.seek(0)
return buffer
def test_job_query_native_smoke_query_search_export(client):
shell = client.get("/portal-shell/job-query?start_date=2026-02-01&end_date=2026-02-11")
assert shell.status_code == 200
page = client.get("/job-query")
assert page.status_code == 200
with (
patch(
"mes_dashboard.services.resource_cache.get_all_resources",
return_value=[
{
"RESOURCEID": "EQ-01",
"RESOURCENAME": "Machine-01",
"WORKCENTERNAME": "WC-A",
"RESOURCEFAMILYNAME": "FAMILY-A",
}
],
),
patch(
"mes_dashboard.routes.job_query_routes.get_jobs_by_resources",
return_value={
"data": [{"JOBID": "JOB001", "RESOURCENAME": "Machine-01"}],
"total": 1,
"resource_count": 1,
},
),
patch(
"mes_dashboard.routes.job_query_routes.export_jobs_with_history",
return_value=iter(["JOBID,RESOURCEID\n", "JOB001,EQ-01\n"]),
),
):
resources = client.get("/api/job-query/resources")
assert resources.status_code == 200
assert resources.get_json()["total"] == 1
query = client.post(
"/api/job-query/jobs",
json={
"resource_ids": ["EQ-01"],
"start_date": "2026-02-01",
"end_date": "2026-02-11",
},
)
assert query.status_code == 200
assert query.get_json()["total"] == 1
export = client.post(
"/api/job-query/export",
json={
"resource_ids": ["EQ-01"],
"start_date": "2026-02-01",
"end_date": "2026-02-11",
},
)
assert export.status_code == 200
assert "text/csv" in export.content_type
def test_excel_query_native_smoke_upload_detect_query_export(client):
_login_as_admin(client)
shell = client.get("/portal-shell/excel-query?mode=upload")
assert shell.status_code == 200
page = client.get("/excel-query")
assert page.status_code == 200
from mes_dashboard.routes.excel_query_routes import _uploaded_excel_cache
_uploaded_excel_cache.clear()
upload = client.post(
"/api/excel-query/upload",
data={"file": (_build_excel_file(), "smoke.xlsx")},
content_type="multipart/form-data",
)
assert upload.status_code == 200
assert "LOT_ID" in upload.get_json()["columns"]
detect = client.post(
"/api/excel-query/column-type",
json={"column_name": "LOT_ID"},
)
assert detect.status_code == 200
assert detect.get_json()["column_name"] == "LOT_ID"
with (
patch(
"mes_dashboard.routes.excel_query_routes.execute_advanced_batch_query",
return_value={
"data": [{"LOT_ID": "LOT001", "QTY": 100}],
"columns": ["LOT_ID", "QTY"],
"total": 1,
},
),
patch(
"mes_dashboard.routes.excel_query_routes.execute_batch_query",
return_value={
"data": [{"LOT_ID": "LOT001", "QTY": 100}],
"columns": ["LOT_ID", "QTY"],
"total": 1,
},
),
):
query = client.post(
"/api/excel-query/execute-advanced",
json={
"table_name": "DWH.DW_MES_WIP",
"search_column": "LOT_ID",
"return_columns": ["LOT_ID", "QTY"],
"search_values": ["LOT001"],
"query_type": "in",
},
)
assert query.status_code == 200
assert query.get_json()["total"] == 1
export = client.post(
"/api/excel-query/export-csv",
json={
"table_name": "DWH.DW_MES_WIP",
"search_column": "LOT_ID",
"return_columns": ["LOT_ID", "QTY"],
"search_values": ["LOT001"],
},
)
assert export.status_code == 200
assert "text/csv" in export.content_type
def test_query_tool_native_smoke_resolve_history_association(client):
_login_as_admin(client)
shell = client.get("/portal-shell/query-tool?input_type=lot_id")
assert shell.status_code == 200
page = client.get("/query-tool")
assert page.status_code == 200
with (
patch(
"mes_dashboard.routes.query_tool_routes.resolve_lots",
return_value={
"data": [{"container_id": "488103800029578b"}],
"total": 1,
"input_count": 1,
"not_found": [],
},
),
patch(
"mes_dashboard.routes.query_tool_routes.get_lot_history",
return_value={"data": [{"CONTAINERID": "488103800029578b"}], "total": 1},
),
patch(
"mes_dashboard.routes.query_tool_routes.get_lot_materials",
return_value={"data": [{"MATERIALLOTID": "MAT001"}], "total": 1},
),
):
resolve = client.post(
"/api/query-tool/resolve",
json={"input_type": "lot_id", "values": ["GA23100020-A00-001"]},
)
assert resolve.status_code == 200
assert resolve.get_json()["total"] == 1
history = client.get("/api/query-tool/lot-history?container_id=488103800029578b")
assert history.status_code == 200
assert history.get_json()["total"] == 1
associations = client.get(
"/api/query-tool/lot-associations?container_id=488103800029578b&type=materials"
)
assert associations.status_code == 200
assert associations.get_json()["total"] == 1
def test_tmtt_defect_native_smoke_range_query_and_csv_export(client):
shell = client.get("/portal-shell/tmtt-defect?start_date=2026-02-01&end_date=2026-02-11")
assert shell.status_code == 200
page = client.get("/tmtt-defect")
assert page.status_code == 200
with (
patch(
"mes_dashboard.routes.tmtt_defect_routes.query_tmtt_defect_analysis",
return_value={
"kpi": {"total_input": 10},
"charts": {"by_workflow": []},
"detail": [],
},
),
patch(
"mes_dashboard.routes.tmtt_defect_routes.export_csv",
return_value=iter(["LOT_ID,TYPE\n", "LOT001,PRINT\n"]),
),
):
query = client.get("/api/tmtt-defect/analysis?start_date=2026-02-01&end_date=2026-02-11")
assert query.status_code == 200
assert query.get_json()["success"] is True
export = client.get("/api/tmtt-defect/export?start_date=2026-02-01&end_date=2026-02-11")
assert export.status_code == 200
assert "text/csv" in export.content_type

19
tests/test_query_tool_routes.py
View File

@@ -28,14 +28,17 @@ def client(app):
return app.test_client()
class TestQueryToolPage:
"""Tests for /query-tool page route."""
def test_page_returns_html(self, client):
"""Should return the query tool page."""
response = client.get('/query-tool')
assert response.status_code == 200
assert b'html' in response.data.lower()
class TestQueryToolPage:
"""Tests for /query-tool page route."""
def test_page_returns_html(self, client):
"""Should return the query tool page."""
with client.session_transaction() as sess:
sess["admin"] = {"username": "admin", "displayName": "Admin"}
response = client.get('/query-tool')
assert response.status_code == 200
assert b'html' in response.data.lower()
class TestResolveEndpoint:

48
tests/test_route_query_compatibility.py Normal file
View File

@@ -0,0 +1,48 @@
# -*- coding: utf-8 -*-
"""Route/query compatibility tests for shell list-detail workflows."""
from __future__ import annotations
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
BASELINE_ROUTE_QUERY_FILE = (
ROOT / "docs" / "migration" / "portal-shell-route-view-integration" / "baseline_route_query_contracts.json"
)
def _read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def test_wip_list_detail_query_contract_compatibility():
routes = _read_json(BASELINE_ROUTE_QUERY_FILE)["routes"]
overview_keys = set(routes["/wip-overview"]["query_keys"])
detail_keys = set(routes["/wip-detail"]["query_keys"])
assert {"workorder", "lotid", "package", "type", "status"}.issubset(overview_keys)
assert overview_keys.issubset(detail_keys)
assert "workcenter" in detail_keys
def test_hold_list_detail_query_contract_compatibility():
routes = _read_json(BASELINE_ROUTE_QUERY_FILE)["routes"]
detail_keys = set(routes["/hold-detail"]["query_keys"])
history_keys = set(routes["/hold-history"]["query_keys"])
assert "reason" in detail_keys
# Hold history route intentionally supports optional query keys at runtime.
assert routes["/hold-history"]["render_mode"] == "native"
assert routes["/hold-detail"]["render_mode"] == "native"
assert isinstance(history_keys, set)
def test_wave_b_routes_keep_native_render_mode_with_query_contract_object():
routes = _read_json(BASELINE_ROUTE_QUERY_FILE)["routes"]
for route in ["/job-query", "/excel-query", "/query-tool", "/tmtt-defect"]:
entry = routes[route]
assert entry["render_mode"] == "native"
assert isinstance(entry["query_keys"], list)

132
tests/test_route_view_migration_baseline.py Normal file
View File

@@ -0,0 +1,132 @@
# -*- coding: utf-8 -*-
"""Validation tests for shell route-view migration baseline artifacts."""
from __future__ import annotations
import json
import copy
from pathlib import Path
from mes_dashboard.app import create_app
from mes_dashboard.services.navigation_contract import (
compute_drawer_visibility,
validate_route_migration_contract,
validate_wave_b_rewrite_entry_criteria,
)
ROOT = Path(__file__).resolve().parent.parent
PAGE_STATUS_FILE = ROOT / "data" / "page_status.json"
BASELINE_DIR = ROOT / "docs" / "migration" / "portal-shell-route-view-integration"
BASELINE_VISIBILITY_FILE = BASELINE_DIR / "baseline_drawer_visibility.json"
BASELINE_ROUTE_QUERY_FILE = BASELINE_DIR / "baseline_route_query_contracts.json"
BASELINE_INTERACTION_FILE = BASELINE_DIR / "baseline_interaction_evidence.json"
ROUTE_CONTRACT_FILE = BASELINE_DIR / "route_migration_contract.json"
ROUTE_CONTRACT_VALIDATION_FILE = BASELINE_DIR / "route_migration_contract_validation.json"
WAVE_B_REWRITE_ENTRY_FILE = BASELINE_DIR / "wave-b-rewrite-entry-criteria.json"
REQUIRED_ROUTES = {
"/wip-overview",
"/wip-detail",
"/hold-overview",
"/hold-detail",
"/hold-history",
"/resource",
"/resource-history",
"/qc-gate",
"/job-query",
"/excel-query",
"/query-tool",
"/tmtt-defect",
}
def _read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def test_route_migration_contract_has_no_validation_errors():
contract = _read_json(ROUTE_CONTRACT_FILE)
errors = validate_route_migration_contract(contract, required_routes=REQUIRED_ROUTES)
assert errors == []
validation_payload = _read_json(ROUTE_CONTRACT_VALIDATION_FILE)
assert validation_payload["errors"] == []
def test_wave_b_rewrite_entry_criteria_blocks_premature_native_cutover():
contract = _read_json(ROUTE_CONTRACT_FILE)
rewrite_entry = _read_json(WAVE_B_REWRITE_ENTRY_FILE)
# Current baseline has complete evidence for Wave B native routes.
assert validate_wave_b_rewrite_entry_criteria(contract, rewrite_entry) == []
# Simulate incomplete criteria while route already in native mode.
mutated_criteria = copy.deepcopy(rewrite_entry)
mutated_criteria["pages"]["/job-query"]["evidence"]["parity"] = "pending"
mutated_criteria["pages"]["/job-query"]["native_cutover_ready"] = False
mutated_criteria["pages"]["/job-query"]["block_reason"] = "pending parity"
errors = validate_wave_b_rewrite_entry_criteria(contract, mutated_criteria)
assert "native cutover blocked for /job-query: rewrite criteria incomplete" in errors
def test_baseline_visibility_matches_current_registry_state():
page_status = _read_json(PAGE_STATUS_FILE)
baseline = _read_json(BASELINE_VISIBILITY_FILE)
assert baseline["admin"] == compute_drawer_visibility(page_status, is_admin=True)
assert baseline["non_admin"] == compute_drawer_visibility(page_status, is_admin=False)
def test_baseline_route_query_contract_covers_all_target_routes():
baseline = _read_json(BASELINE_ROUTE_QUERY_FILE)
routes = baseline["routes"]
assert set(routes.keys()) == REQUIRED_ROUTES
for route in REQUIRED_ROUTES:
assert "query_keys" in routes[route]
assert "render_mode" in routes[route]
assert routes[route]["render_mode"] in {"native", "wrapper"}
def test_interaction_evidence_contains_required_sections_for_all_routes():
payload = _read_json(BASELINE_INTERACTION_FILE)
routes = payload["routes"]
assert set(routes.keys()) == REQUIRED_ROUTES
for route in REQUIRED_ROUTES:
entry = routes[route]
assert "table" in entry
assert "chart" in entry
assert "filter" in entry
assert "matrix" in entry
def test_navigation_api_drawer_parity_matches_shell_baseline_for_admin_and_non_admin():
app = create_app("testing")
app.config["TESTING"] = True
baseline = _read_json(BASELINE_VISIBILITY_FILE)
non_admin_client = app.test_client()
non_admin_payload = _read_response_json(non_admin_client.get("/api/portal/navigation"))
assert _route_set(non_admin_payload["drawers"]) == _route_set(baseline["non_admin"])
admin_client = app.test_client()
with admin_client.session_transaction() as sess:
sess["admin"] = {"displayName": "Admin", "employeeNo": "A001"}
admin_payload = _read_response_json(admin_client.get("/api/portal/navigation"))
assert _route_set(admin_payload["drawers"]) == _route_set(baseline["admin"])
def _read_response_json(response) -> dict:
return json.loads(response.data.decode("utf-8"))
def _route_set(drawers: list[dict]) -> set[str]:
return {
page["route"]
for drawer in drawers
for page in drawer.get("pages", [])
}

4
tests/test_runtime_hardening.py
View File

@@ -154,7 +154,7 @@ def test_health_reports_pool_saturation_degraded_reason(
def test_security_headers_applied_globally(testing_app_factory):
app = testing_app_factory(csrf_enabled=False)
response = app.test_client().get("/")
response = app.test_client().get("/", follow_redirects=True)
assert response.status_code == 200
assert "Content-Security-Policy" in response.headers
@@ -175,7 +175,7 @@ def test_hsts_header_enabled_in_production(monkeypatch):
app = create_app("production")
app.config["TESTING"] = True
response = app.test_client().get("/")
response = app.test_client().get("/", follow_redirects=True)
assert response.status_code == 200
assert "Strict-Transport-Security" in response.headers

54
tests/test_visual_regression_snapshots.py Normal file
View File

@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
"""Visual regression snapshot contract checks for migration-critical states."""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SNAPSHOT_FILE = (
ROOT / "docs" / "migration" / "portal-shell-route-view-integration" / "visual-regression-snapshots.json"
)
def _read_json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def _sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def _compute_fingerprint(files: list[str]) -> str:
lines: list[str] = []
for rel in files:
path = ROOT / rel
assert path.exists(), f"snapshot file missing: {rel}"
digest = hashlib.sha256(path.read_bytes()).hexdigest()
lines.append(rel)
lines.append(digest)
payload = "\n".join(lines) + "\n"
return _sha256_text(payload)
def test_visual_snapshot_policy_blocks_release_on_critical_diff():
payload = _read_json(SNAPSHOT_FILE)
policy = payload["critical_diff_policy"]
assert policy["block_release"] is True
assert policy["severity"] == "critical"
def test_visual_snapshot_fingerprints_match_current_sources():
payload = _read_json(SNAPSHOT_FILE)
snapshots = payload.get("snapshots", [])
assert snapshots, "no visual snapshot entries"
for item in snapshots:
files = item.get("files", [])
expected = str(item.get("fingerprint", "")).strip()
assert files and expected, f"invalid snapshot entry: {item.get('id')}"
actual = _compute_fingerprint(files)
assert actual == expected, f"critical visual snapshot diff: {item.get('id')}"