feat(modernization): full architecture blueprint with hardening follow-up
Implement phased modernization infrastructure for transitioning from multi-page legacy routing to SPA portal-shell architecture, plus post-delivery hardening fixes for policy loading, fallback consistency, and governance drift detection. Key changes: - Add route contract enrichment with scope/visibility/compatibility policies - Canonical 302 redirects from legacy direct-entry to /portal-shell/ routes - Asset readiness enforcement and runtime fallback retirement for in-scope routes - Shared feature-flag helpers (env > config > default) replacing duplicated _to_bool - Defensive copy for lru_cached policy payloads preventing mutation corruption - Unified retired-fallback response helper across app and blueprint routes - Frontend/backend route-contract cross-validation in governance gates - Shell CSS token fallback values for routes rendered outside shell scope - Local-safe .env.example defaults with production recommendation comments - Legacy contract fallback warning logging and single-hop redirect optimization Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -22,15 +22,24 @@ os.environ.setdefault('WATCHDOG_STATE_FILE', os.path.join(_TMP_DIR, 'mes_dashboa
|
||||
|
||||
import mes_dashboard.core.database as db
|
||||
from mes_dashboard.app import create_app
|
||||
from mes_dashboard.core.modernization_policy import clear_modernization_policy_cache
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
@pytest.fixture
|
||||
def app():
|
||||
"""Create application for testing."""
|
||||
db._ENGINE = None
|
||||
app = create_app('testing')
|
||||
app.config['TESTING'] = True
|
||||
return app
|
||||
app.config['TESTING'] = True
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_modernization_policy_cache():
|
||||
"""Keep policy-cache state isolated across tests."""
|
||||
clear_modernization_policy_cache()
|
||||
yield
|
||||
clear_modernization_policy_cache()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -130,14 +130,21 @@ class TestWipAndHoldPagesE2E:
|
||||
back_href = page.locator("a.btn-back").get_attribute("href") or ""
|
||||
parsed = urlparse(back_href)
|
||||
params = parse_qs(parsed.query)
|
||||
assert parsed.path == "/wip-overview"
|
||||
assert parsed.path in {"/wip-overview", "/portal-shell/wip-overview"}
|
||||
assert params.get("type", [None])[0] == "PJA3460"
|
||||
assert params.get("status", [None])[0] in {"queue", "QUEUE"}
|
||||
|
||||
def test_hold_detail_without_reason_redirects_to_overview(self, page: Page, app_server: str):
|
||||
nav_resp = _get_with_retry(f"{app_server}/api/portal/navigation", attempts=3, timeout=10.0)
|
||||
nav_payload = nav_resp.json() if nav_resp.ok else {}
|
||||
spa_enabled = bool(nav_payload.get("portal_spa_enabled"))
|
||||
|
||||
response = _get_with_retry(f"{app_server}/hold-detail", attempts=3, timeout=10.0)
|
||||
assert response.status_code == 302
|
||||
assert response.headers.get("Location") == "/wip-overview"
|
||||
if spa_enabled:
|
||||
assert response.headers.get("Location") == "/portal-shell/wip-overview"
|
||||
else:
|
||||
assert response.headers.get("Location") == "/wip-overview"
|
||||
|
||||
def test_hold_detail_calls_summary_distribution_and_lots(self, page: Page, app_server: str):
|
||||
reason = _pick_hold_reason(app_server)
|
||||
|
||||
72
tests/test_asset_readiness_policy.py
Normal file
72
tests/test_asset_readiness_policy.py
Normal file
@@ -0,0 +1,72 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for modernization asset-readiness and fallback-retirement policy."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from mes_dashboard.app import create_app
|
||||
|
||||
|
||||
def test_asset_readiness_enforcement_blocks_startup_when_missing_assets(monkeypatch):
|
||||
monkeypatch.setenv("MODERNIZATION_ENFORCE_ASSET_READINESS", "true")
|
||||
|
||||
with patch(
|
||||
"mes_dashboard.app.get_missing_in_scope_assets",
|
||||
return_value=["/wip-overview:wip-overview.html"],
|
||||
):
|
||||
with pytest.raises(RuntimeError, match="In-scope asset readiness check failed"):
|
||||
create_app("testing")
|
||||
|
||||
|
||||
def test_in_scope_fallback_retirement_returns_503_when_dist_asset_missing(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "false")
|
||||
monkeypatch.setenv("MODERNIZATION_RETIRE_IN_SCOPE_RUNTIME_FALLBACK", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
|
||||
with patch("mes_dashboard.routes.hold_overview_routes.os.path.exists", return_value=False):
|
||||
response = client.get("/hold-overview")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert "系統發生錯誤" in response.data.decode("utf-8")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("route", "exists_patch"),
|
||||
[
|
||||
("/hold-overview", "mes_dashboard.routes.hold_overview_routes.os.path.exists"),
|
||||
("/hold-history", "mes_dashboard.routes.hold_history_routes.os.path.exists"),
|
||||
("/hold-detail?reason=YieldLimit", "mes_dashboard.routes.hold_routes.os.path.exists"),
|
||||
],
|
||||
)
|
||||
def test_hold_blueprints_share_retired_fallback_template(monkeypatch, route, exists_patch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "false")
|
||||
monkeypatch.setenv("MODERNIZATION_RETIRE_IN_SCOPE_RUNTIME_FALLBACK", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
|
||||
with patch(exists_patch, return_value=False):
|
||||
response = client.get(route)
|
||||
|
||||
assert response.status_code == 503
|
||||
assert "系統發生錯誤" in response.data.decode("utf-8")
|
||||
|
||||
|
||||
def test_deferred_route_keeps_fallback_posture_when_in_scope_retirement_enabled(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "false")
|
||||
monkeypatch.setenv("MODERNIZATION_RETIRE_IN_SCOPE_RUNTIME_FALLBACK", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
with client.session_transaction() as sess:
|
||||
sess["admin"] = {"displayName": "Admin", "employeeNo": "A001"}
|
||||
|
||||
with patch("mes_dashboard.app.os.path.exists", return_value=False):
|
||||
response = client.get("/tables")
|
||||
|
||||
assert response.status_code == 200
|
||||
25
tests/test_feature_flags.py
Normal file
25
tests/test_feature_flags.py
Normal file
@@ -0,0 +1,25 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for shared feature-flag resolution helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from mes_dashboard.core.feature_flags import parse_bool, resolve_bool_flag
|
||||
|
||||
|
||||
def test_parse_bool_supports_true_and_false_tokens():
|
||||
assert parse_bool("true", default=False) is True
|
||||
assert parse_bool(" yes ", default=False) is True
|
||||
assert parse_bool("0", default=True) is False
|
||||
assert parse_bool("off", default=True) is False
|
||||
|
||||
|
||||
def test_resolve_bool_flag_prefers_environment_over_config():
|
||||
env = {"FEATURE_X": "false"}
|
||||
config = {"FEATURE_X": True}
|
||||
assert resolve_bool_flag("FEATURE_X", config=config, default=True, environ=env) is False
|
||||
|
||||
|
||||
def test_resolve_bool_flag_uses_config_then_default_when_env_missing():
|
||||
config = {"FEATURE_X": "true"}
|
||||
assert resolve_bool_flag("FEATURE_X", config=config, default=False, environ={}) is True
|
||||
assert resolve_bool_flag("MISSING", config=config, default=False, environ={}) is False
|
||||
97
tests/test_full_modernization_gates.py
Normal file
97
tests/test_full_modernization_gates.py
Normal file
@@ -0,0 +1,97 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for full modernization governance gate runner."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = ROOT / "scripts" / "check_full_modernization_gates.py"
|
||||
REPORT = ROOT / "docs" / "migration" / "full-modernization-architecture-blueprint" / "quality_gate_report.json"
|
||||
SCOPE_MATRIX = ROOT / "docs" / "migration" / "full-modernization-architecture-blueprint" / "route_scope_matrix.json"
|
||||
|
||||
_GATES_SPEC = importlib.util.spec_from_file_location("check_full_modernization_gates", SCRIPT)
|
||||
assert _GATES_SPEC and _GATES_SPEC.loader
|
||||
gates = importlib.util.module_from_spec(_GATES_SPEC)
|
||||
sys.modules[_GATES_SPEC.name] = gates
|
||||
_GATES_SPEC.loader.exec_module(gates)
|
||||
|
||||
|
||||
def _run(mode: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["python", str(SCRIPT), "--mode", mode, "--report", str(REPORT)],
|
||||
cwd=str(ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def test_gate_runner_warn_mode_passes_and_generates_report():
|
||||
result = _run("warn")
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
|
||||
payload = json.loads(REPORT.read_text(encoding="utf-8"))
|
||||
assert payload["mode"] == "warn"
|
||||
assert payload["passed"] is True
|
||||
|
||||
|
||||
def test_gate_runner_block_mode_passes_current_baseline():
|
||||
result = _run("block")
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
|
||||
payload = json.loads(REPORT.read_text(encoding="utf-8"))
|
||||
assert payload["mode"] == "block"
|
||||
assert payload["passed"] is True
|
||||
|
||||
|
||||
def test_scope_matrix_keeps_deferred_routes_for_follow_up_only():
|
||||
matrix = json.loads(SCOPE_MATRIX.read_text(encoding="utf-8"))
|
||||
deferred = {item["route"] for item in matrix["deferred"]}
|
||||
assert deferred == {"/tables", "/excel-query", "/query-tool", "/mid-section-defect"}
|
||||
|
||||
|
||||
def test_route_contract_parity_check_detects_route_set_drift():
|
||||
report = gates.CheckReport(mode="block")
|
||||
backend = {"/wip-overview": {"scope": "in-scope"}}
|
||||
frontend = {
|
||||
"/wip-overview": "in-scope",
|
||||
"/extra": "deferred",
|
||||
}
|
||||
|
||||
gates._check_frontend_backend_route_contract_parity(backend, frontend, report)
|
||||
|
||||
assert any("/extra" in error for error in report.errors)
|
||||
|
||||
|
||||
def test_route_contract_parity_check_detects_scope_mismatch():
|
||||
report = gates.CheckReport(mode="block")
|
||||
backend = {"/wip-overview": {"scope": "in-scope"}}
|
||||
frontend = {"/wip-overview": "deferred"}
|
||||
|
||||
gates._check_frontend_backend_route_contract_parity(backend, frontend, report)
|
||||
|
||||
assert any("scope mismatch" in error for error in report.errors)
|
||||
|
||||
|
||||
def test_style_governance_flags_shell_tokens_without_fallback(tmp_path):
|
||||
report = gates.CheckReport(mode="block")
|
||||
css_file = tmp_path / "route.css"
|
||||
css_file.write_text(
|
||||
".demo { background: linear-gradient(var(--portal-brand-start), #fff); }",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
original_route_css_targets = gates._route_css_targets
|
||||
try:
|
||||
gates._route_css_targets = lambda: {"/wip-overview": [css_file]} # type: ignore[assignment]
|
||||
gates._check_style_governance({"/wip-overview"}, {}, report)
|
||||
finally:
|
||||
gates._route_css_targets = original_route_css_targets # type: ignore[assignment]
|
||||
|
||||
assert any("without fallback" in error for error in report.errors)
|
||||
@@ -24,18 +24,15 @@ class TestHoldHistoryPageRoute(TestHoldHistoryRoutesBase):
|
||||
|
||||
@patch('mes_dashboard.routes.hold_history_routes.os.path.exists', return_value=False)
|
||||
def test_hold_history_page_includes_vite_entry(self, _mock_exists):
|
||||
with self.client.session_transaction() as sess:
|
||||
sess['admin'] = {'displayName': 'Test Admin', 'employeeNo': 'A001'}
|
||||
|
||||
response = self.client.get('/hold-history')
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'/static/dist/hold-history.js', response.data)
|
||||
response = self.client.get('/hold-history', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/hold-history'))
|
||||
|
||||
@patch('mes_dashboard.routes.hold_history_routes.os.path.exists', return_value=False)
|
||||
def test_hold_history_page_returns_403_without_admin(self, _mock_exists):
|
||||
response = self.client.get('/hold-history')
|
||||
self.assertEqual(response.status_code, 403)
|
||||
def test_hold_history_page_redirects_without_admin(self, _mock_exists):
|
||||
response = self.client.get('/hold-history', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/hold-history'))
|
||||
|
||||
|
||||
class TestHoldHistoryTrendRoute(TestHoldHistoryRoutesBase):
|
||||
|
||||
@@ -24,17 +24,15 @@ class TestHoldOverviewPageRoute(TestHoldOverviewRoutesBase):
|
||||
|
||||
@patch('mes_dashboard.routes.hold_overview_routes.os.path.exists', return_value=False)
|
||||
def test_hold_overview_page_includes_vite_entry(self, _mock_exists):
|
||||
# Page is registered as 'dev' status, requires admin session
|
||||
with self.client.session_transaction() as sess:
|
||||
sess['admin'] = {'displayName': 'Test Admin', 'employeeNo': 'A001'}
|
||||
response = self.client.get('/hold-overview')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'/static/dist/hold-overview.js', response.data)
|
||||
response = self.client.get('/hold-overview', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/hold-overview'))
|
||||
|
||||
@patch('mes_dashboard.routes.hold_overview_routes.os.path.exists', return_value=False)
|
||||
def test_hold_overview_page_returns_403_without_admin(self, _mock_exists):
|
||||
response = self.client.get('/hold-overview')
|
||||
self.assertEqual(response.status_code, 403)
|
||||
def test_hold_overview_page_redirects_without_admin(self, _mock_exists):
|
||||
response = self.client.get('/hold-overview', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/hold-overview'))
|
||||
|
||||
|
||||
class TestHoldOverviewSummaryRoute(TestHoldOverviewRoutesBase):
|
||||
@@ -225,4 +223,3 @@ class TestHoldOverviewLotsRoute(TestHoldOverviewRoutesBase):
|
||||
self.assertEqual(payload['error']['code'], 'TOO_MANY_REQUESTS')
|
||||
self.assertEqual(response.headers.get('Retry-After'), '4')
|
||||
mock_service.assert_not_called()
|
||||
|
||||
|
||||
@@ -23,25 +23,44 @@ class TestHoldRoutesBase(unittest.TestCase):
|
||||
self.client = self.app.test_client()
|
||||
|
||||
|
||||
class TestHoldDetailPageRoute(TestHoldRoutesBase):
|
||||
"""Test GET /hold-detail page route."""
|
||||
|
||||
def test_hold_detail_page_requires_reason(self):
|
||||
"""GET /hold-detail without reason should redirect to wip-overview."""
|
||||
response = self.client.get('/hold-detail')
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertIn('/wip-overview', response.location)
|
||||
|
||||
def test_hold_detail_page_with_reason(self):
|
||||
"""GET /hold-detail?reason=xxx should return 200."""
|
||||
response = self.client.get('/hold-detail?reason=YieldLimit')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_hold_detail_page_includes_vite_entry(self):
|
||||
"""Page should load the Hold Detail Vite module."""
|
||||
response = self.client.get('/hold-detail?reason=YieldLimit')
|
||||
class TestHoldDetailPageRoute(TestHoldRoutesBase):
|
||||
"""Test GET /hold-detail page route."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.app.config['PORTAL_SPA_ENABLED'] = True
|
||||
|
||||
def test_hold_detail_page_requires_reason(self):
|
||||
"""SPA mode should single-hop redirect missing reason to canonical shell overview."""
|
||||
response = self.client.get('/hold-detail', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/wip-overview'))
|
||||
|
||||
def test_hold_detail_page_requires_reason_non_spa_mode(self):
|
||||
"""Non-SPA mode should keep legacy overview redirect behavior."""
|
||||
self.app.config['PORTAL_SPA_ENABLED'] = False
|
||||
response = self.client.get('/hold-detail', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/wip-overview'))
|
||||
|
||||
def test_hold_detail_page_requires_reason_has_single_redirect_hop_in_spa_mode(self):
|
||||
"""Follow-redirect flow should complete with exactly one redirect hop."""
|
||||
response = self.client.get('/hold-detail', follow_redirects=True)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn(b'/static/dist/hold-detail.js', response.data)
|
||||
self.assertEqual(len(response.history), 1)
|
||||
self.assertTrue(response.history[0].location.endswith('/portal-shell/wip-overview'))
|
||||
|
||||
def test_hold_detail_page_with_reason(self):
|
||||
"""GET /hold-detail?reason=xxx should redirect to canonical shell route."""
|
||||
response = self.client.get('/hold-detail?reason=YieldLimit', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/hold-detail?reason=YieldLimit'))
|
||||
|
||||
def test_hold_detail_page_includes_vite_entry(self):
|
||||
"""Direct entry should be redirected to canonical shell host page."""
|
||||
response = self.client.get('/hold-detail?reason=YieldLimit', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertIn('/portal-shell/hold-detail?reason=YieldLimit', response.location)
|
||||
|
||||
|
||||
class TestHoldDetailSummaryRoute(TestHoldRoutesBase):
|
||||
|
||||
@@ -25,14 +25,14 @@ def client(app):
|
||||
return app.test_client()
|
||||
|
||||
|
||||
class TestJobQueryPage:
|
||||
"""Tests for /job-query page route."""
|
||||
|
||||
def test_page_returns_html(self, client):
|
||||
"""Should return the job query page."""
|
||||
response = client.get('/job-query')
|
||||
assert response.status_code == 200
|
||||
assert b'html' in response.data.lower()
|
||||
class TestJobQueryPage:
|
||||
"""Tests for /job-query page route."""
|
||||
|
||||
def test_page_returns_html(self, client):
|
||||
"""Should redirect direct entry to canonical shell page."""
|
||||
response = client.get('/job-query', follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert response.location.endswith('/portal-shell/job-query')
|
||||
|
||||
|
||||
class TestGetResources:
|
||||
|
||||
45
tests/test_modernization_policy_hardening.py
Normal file
45
tests/test_modernization_policy_hardening.py
Normal file
@@ -0,0 +1,45 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Hardening tests for modernization policy caching behavior."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from mes_dashboard.core import modernization_policy as policy
|
||||
|
||||
|
||||
def test_scope_matrix_loader_returns_defensive_copy(tmp_path, monkeypatch):
|
||||
scope_file = tmp_path / "route_scope_matrix.json"
|
||||
scope_file.write_text(
|
||||
json.dumps({"in_scope": [{"route": "/wip-overview"}], "deferred": []}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setattr(policy, "SCOPE_MATRIX_FILE", scope_file)
|
||||
policy.clear_modernization_policy_cache()
|
||||
|
||||
payload = policy.load_scope_matrix()
|
||||
payload["in_scope"].append({"route": "/mutated"})
|
||||
|
||||
fresh_payload = policy.load_scope_matrix()
|
||||
routes = [item["route"] for item in fresh_payload["in_scope"]]
|
||||
assert routes == ["/wip-overview"]
|
||||
|
||||
|
||||
def test_scope_matrix_cache_refresh_requires_explicit_clear(tmp_path, monkeypatch):
|
||||
scope_file = tmp_path / "route_scope_matrix.json"
|
||||
scope_file.write_text(
|
||||
json.dumps({"in_scope": [{"route": "/before"}], "deferred": []}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setattr(policy, "SCOPE_MATRIX_FILE", scope_file)
|
||||
policy.clear_modernization_policy_cache()
|
||||
assert [item["route"] for item in policy.load_scope_matrix()["in_scope"]] == ["/before"]
|
||||
|
||||
scope_file.write_text(
|
||||
json.dumps({"in_scope": [{"route": "/after"}], "deferred": []}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert [item["route"] for item in policy.load_scope_matrix()["in_scope"]] == ["/before"]
|
||||
|
||||
policy.clear_modernization_policy_cache()
|
||||
assert [item["route"] for item in policy.load_scope_matrix()["in_scope"]] == ["/after"]
|
||||
@@ -4,7 +4,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
from unittest.mock import ANY, MagicMock, patch
|
||||
|
||||
from mes_dashboard.app import create_app
|
||||
|
||||
@@ -19,7 +21,7 @@ def test_portal_shell_fallback_html_served_when_dist_missing(monkeypatch):
|
||||
app.config["TESTING"] = True
|
||||
|
||||
# Force fallback path by simulating missing dist file.
|
||||
monkeypatch.setattr("os.path.exists", lambda *_args, **_kwargs: False)
|
||||
monkeypatch.setattr("mes_dashboard.app.os.path.exists", lambda *_args, **_kwargs: False)
|
||||
|
||||
client = app.test_client()
|
||||
response = client.get("/portal-shell")
|
||||
@@ -42,7 +44,7 @@ def test_portal_shell_uses_nested_dist_html_when_top_level_missing(monkeypatch):
|
||||
return False
|
||||
return path.endswith("/dist/src/portal-shell/index.html")
|
||||
|
||||
monkeypatch.setattr("os.path.exists", fake_exists)
|
||||
monkeypatch.setattr("mes_dashboard.app.os.path.exists", fake_exists)
|
||||
|
||||
client = app.test_client()
|
||||
response = client.get("/portal-shell")
|
||||
@@ -133,7 +135,7 @@ def test_navigation_drawer_and_page_order_deterministic_non_admin():
|
||||
assert drawer_ids == ["reports", "drawer-2", "drawer"]
|
||||
|
||||
reports_routes = [page["route"] for page in payload["drawers"][0]["pages"]]
|
||||
assert reports_routes == ["/wip-overview", "/resource", "/qc-gate"]
|
||||
assert reports_routes == ["/wip-overview", "/hold-overview", "/resource", "/qc-gate"]
|
||||
|
||||
|
||||
def test_navigation_contract_page_metadata_fields_present_and_typed():
|
||||
@@ -199,29 +201,52 @@ def test_navigation_mixed_release_dev_visibility_admin_vs_non_admin():
|
||||
app.config["TESTING"] = True
|
||||
|
||||
non_admin_client = app.test_client()
|
||||
non_admin_resp = non_admin_client.get("/api/portal/navigation")
|
||||
assert non_admin_resp.status_code == 200
|
||||
non_admin_payload = json.loads(non_admin_resp.data.decode("utf-8"))
|
||||
non_admin_routes = {
|
||||
page["route"]
|
||||
for drawer in non_admin_payload["drawers"]
|
||||
for page in drawer["pages"]
|
||||
}
|
||||
assert "/hold-overview" not in non_admin_routes
|
||||
assert "/hold-history" not in non_admin_routes
|
||||
mixed_config = [
|
||||
{
|
||||
"id": "reports",
|
||||
"name": "Reports",
|
||||
"order": 1,
|
||||
"admin_only": False,
|
||||
"pages": [
|
||||
{"route": "/wip-overview", "name": "WIP", "status": "released", "order": 1},
|
||||
{"route": "/hold-overview", "name": "Hold", "status": "dev", "order": 2},
|
||||
],
|
||||
}
|
||||
]
|
||||
def fake_status(route: str):
|
||||
if route == "/hold-overview":
|
||||
return "dev"
|
||||
if route == "/wip-overview":
|
||||
return "released"
|
||||
return None
|
||||
|
||||
admin_client = app.test_client()
|
||||
_login_as_admin(admin_client)
|
||||
admin_resp = admin_client.get("/api/portal/navigation")
|
||||
assert admin_resp.status_code == 200
|
||||
admin_payload = json.loads(admin_resp.data.decode("utf-8"))
|
||||
admin_routes = {
|
||||
page["route"]
|
||||
for drawer in admin_payload["drawers"]
|
||||
for page in drawer["pages"]
|
||||
}
|
||||
assert "/hold-overview" in admin_routes
|
||||
assert "/hold-history" in admin_routes
|
||||
with (
|
||||
patch("mes_dashboard.app.get_navigation_config", return_value=mixed_config),
|
||||
patch("mes_dashboard.app.get_page_status", side_effect=fake_status),
|
||||
):
|
||||
non_admin_resp = non_admin_client.get("/api/portal/navigation")
|
||||
assert non_admin_resp.status_code == 200
|
||||
non_admin_payload = json.loads(non_admin_resp.data.decode("utf-8"))
|
||||
non_admin_routes = {
|
||||
page["route"]
|
||||
for drawer in non_admin_payload["drawers"]
|
||||
for page in drawer["pages"]
|
||||
}
|
||||
assert "/wip-overview" in non_admin_routes
|
||||
assert "/hold-overview" not in non_admin_routes
|
||||
|
||||
admin_client = app.test_client()
|
||||
_login_as_admin(admin_client)
|
||||
admin_resp = admin_client.get("/api/portal/navigation")
|
||||
assert admin_resp.status_code == 200
|
||||
admin_payload = json.loads(admin_resp.data.decode("utf-8"))
|
||||
admin_routes = {
|
||||
page["route"]
|
||||
for drawer in admin_payload["drawers"]
|
||||
for page in drawer["pages"]
|
||||
}
|
||||
assert "/wip-overview" in admin_routes
|
||||
assert "/hold-overview" in admin_routes
|
||||
|
||||
|
||||
def test_portal_navigation_includes_admin_links_by_auth_state():
|
||||
@@ -302,7 +327,8 @@ def test_portal_navigation_logs_contract_mismatch_route():
|
||||
assert payload["diagnostics"]["contract_mismatch_routes"] == ["/resource"]
|
||||
|
||||
|
||||
def test_wave_b_native_routes_are_reachable():
|
||||
def test_wave_b_native_routes_are_reachable(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "false")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
@@ -311,3 +337,96 @@ def test_wave_b_native_routes_are_reachable():
|
||||
for route in ["/job-query", "/excel-query", "/query-tool", "/tmtt-defect"]:
|
||||
response = client.get(route)
|
||||
assert response.status_code == 200, f"{route} should be reachable"
|
||||
|
||||
|
||||
def test_direct_entry_in_scope_report_routes_redirect_to_canonical_shell_when_spa_enabled(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
_login_as_admin(client)
|
||||
|
||||
cases = {
|
||||
"/wip-overview?status=queue": "/portal-shell/wip-overview?status=queue",
|
||||
"/resource-history?granularity=day": "/portal-shell/resource-history?granularity=day",
|
||||
"/job-query?start_date=2026-02-01&end_date=2026-02-02": "/portal-shell/job-query?start_date=2026-02-01&end_date=2026-02-02",
|
||||
"/tmtt-defect?start_date=2026-02-01&end_date=2026-02-02": "/portal-shell/tmtt-defect?start_date=2026-02-01&end_date=2026-02-02",
|
||||
"/hold-detail?reason=YieldLimit": "/portal-shell/hold-detail?reason=YieldLimit",
|
||||
}
|
||||
|
||||
for direct_url, canonical_url in cases.items():
|
||||
response = client.get(direct_url, follow_redirects=False)
|
||||
assert response.status_code == 302, direct_url
|
||||
assert response.location.endswith(canonical_url), response.location
|
||||
|
||||
|
||||
def test_direct_entry_redirect_preserves_non_ascii_query_params(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
|
||||
response = client.get("/wip-detail?workcenter=焊接_DB&status=queue", follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
|
||||
parsed = urlparse(response.location)
|
||||
assert parsed.path.endswith("/portal-shell/wip-detail")
|
||||
query = parse_qs(parsed.query)
|
||||
assert query.get("workcenter") == ["焊接_DB"]
|
||||
assert query.get("status") == ["queue"]
|
||||
|
||||
|
||||
def test_legacy_shell_contract_fallback_logs_warning(monkeypatch):
|
||||
from mes_dashboard import app as app_module
|
||||
|
||||
app_module._SHELL_ROUTE_CONTRACT_MAP = None
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
|
||||
primary_suffix = "/docs/migration/full-modernization-architecture-blueprint/route_contracts.json"
|
||||
legacy_suffix = "/docs/migration/portal-shell-route-view-integration/route_migration_contract.json"
|
||||
sample_payload = json.dumps({"routes": [{"route": "/wip-overview", "scope": "in-scope"}]})
|
||||
original_exists = Path.exists
|
||||
original_read_text = Path.read_text
|
||||
|
||||
def fake_exists(self):
|
||||
raw = str(self).replace("\\", "/")
|
||||
if raw.endswith(primary_suffix):
|
||||
return False
|
||||
if raw.endswith(legacy_suffix):
|
||||
return True
|
||||
return original_exists(self)
|
||||
|
||||
def fake_read_text(self, encoding="utf-8"):
|
||||
raw = str(self).replace("\\", "/")
|
||||
if raw.endswith(legacy_suffix):
|
||||
return sample_payload
|
||||
return original_read_text(self, encoding=encoding)
|
||||
|
||||
logger = MagicMock()
|
||||
with (
|
||||
patch("mes_dashboard.app.logging.getLogger", return_value=logger),
|
||||
patch.object(Path, "exists", fake_exists),
|
||||
patch.object(Path, "read_text", fake_read_text),
|
||||
):
|
||||
contract_map = app_module._load_shell_route_contract_map()
|
||||
|
||||
assert "/wip-overview" in contract_map
|
||||
logger.warning.assert_any_call(
|
||||
"Using legacy contract file fallback for shell route contracts: %s",
|
||||
ANY,
|
||||
)
|
||||
app_module._SHELL_ROUTE_CONTRACT_MAP = None
|
||||
|
||||
|
||||
def test_deferred_routes_keep_direct_entry_compatibility_when_spa_enabled(monkeypatch):
|
||||
monkeypatch.setenv("PORTAL_SPA_ENABLED", "true")
|
||||
app = create_app("testing")
|
||||
app.config["TESTING"] = True
|
||||
client = app.test_client()
|
||||
_login_as_admin(client)
|
||||
|
||||
for route in ["/excel-query", "/query-tool", "/tables"]:
|
||||
response = client.get(route, follow_redirects=False)
|
||||
# Deferred routes stay on direct-entry posture in this phase.
|
||||
assert response.status_code == 200, route
|
||||
|
||||
@@ -6,7 +6,6 @@ from __future__ import annotations
|
||||
from unittest.mock import patch
|
||||
|
||||
import mes_dashboard.core.database as db
|
||||
from flask import Response
|
||||
|
||||
from mes_dashboard.app import create_app
|
||||
|
||||
@@ -57,15 +56,7 @@ def test_qc_gate_summary_route_returns_500_on_failure(_mock_get_summary):
|
||||
assert 'error' in payload
|
||||
|
||||
|
||||
@patch('mes_dashboard.app.send_from_directory')
|
||||
def test_qc_gate_page_served_from_static_dist(mock_send_from_directory):
|
||||
mock_send_from_directory.return_value = Response('<html>ok</html>', mimetype='text/html')
|
||||
|
||||
response = _client().get('/qc-gate')
|
||||
|
||||
assert response.status_code == 200
|
||||
assert 'text/html' in response.content_type
|
||||
|
||||
call_args = mock_send_from_directory.call_args[0]
|
||||
assert call_args[0].endswith('/static/dist')
|
||||
assert call_args[1] == 'qc-gate.html'
|
||||
def test_qc_gate_page_redirects_to_canonical_shell_when_spa_enabled():
|
||||
response = _client().get('/qc-gate', follow_redirects=False)
|
||||
assert response.status_code == 302
|
||||
assert response.location.endswith('/portal-shell/qc-gate')
|
||||
|
||||
@@ -18,6 +18,14 @@ def _login_as_admin(client):
|
||||
sess['admin'] = {'displayName': 'Test Admin', 'employeeNo': 'A001'}
|
||||
|
||||
|
||||
def _get_response_and_html(client, endpoint):
|
||||
response = client.get(endpoint, follow_redirects=False)
|
||||
if response.status_code in {301, 302, 307, 308}:
|
||||
follow = client.get(response.location)
|
||||
return response, follow, follow.data.decode('utf-8')
|
||||
return response, response, response.data.decode('utf-8')
|
||||
|
||||
|
||||
class TestTemplateIntegration(unittest.TestCase):
|
||||
"""Test that all templates properly extend _base.html."""
|
||||
|
||||
@@ -273,18 +281,26 @@ class TestMesApiUsageInTemplates(unittest.TestCase):
|
||||
_login_as_admin(self.client)
|
||||
|
||||
def test_wip_overview_uses_mesapi(self):
|
||||
response = self.client.get('/wip-overview')
|
||||
html = response.data.decode('utf-8')
|
||||
response, final_response, html = _get_response_and_html(self.client, '/wip-overview')
|
||||
|
||||
self.assertTrue('MesApi.get' in html or '/static/dist/wip-overview.js' in html)
|
||||
self.assertNotIn('fetchWithTimeout', html)
|
||||
if response.status_code == 302:
|
||||
self.assertTrue(response.location.endswith('/portal-shell/wip-overview'))
|
||||
self.assertEqual(final_response.status_code, 200)
|
||||
self.assertIn('/static/dist/portal-shell.js', html)
|
||||
else:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue('MesApi.get' in html or '/static/dist/wip-overview.js' in html)
|
||||
|
||||
def test_wip_detail_uses_mesapi(self):
|
||||
response = self.client.get('/wip-detail')
|
||||
html = response.data.decode('utf-8')
|
||||
response, final_response, html = _get_response_and_html(self.client, '/wip-detail')
|
||||
|
||||
self.assertTrue('MesApi.get' in html or '/static/dist/wip-detail.js' in html)
|
||||
self.assertNotIn('fetchWithTimeout', html)
|
||||
if response.status_code == 302:
|
||||
self.assertTrue(response.location.endswith('/portal-shell/wip-detail'))
|
||||
self.assertEqual(final_response.status_code, 200)
|
||||
self.assertIn('/static/dist/portal-shell.js', html)
|
||||
else:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue('MesApi.get' in html or '/static/dist/wip-detail.js' in html)
|
||||
|
||||
def test_tables_page_uses_mesapi_or_vite_module(self):
|
||||
response = self.client.get('/tables')
|
||||
@@ -293,14 +309,19 @@ class TestMesApiUsageInTemplates(unittest.TestCase):
|
||||
self.assertTrue('MesApi.post' in html or '/static/dist/tables.js' in html)
|
||||
|
||||
def test_resource_page_uses_mesapi_or_vite_module(self):
|
||||
response = self.client.get('/resource')
|
||||
html = response.data.decode('utf-8')
|
||||
response, final_response, html = _get_response_and_html(self.client, '/resource')
|
||||
|
||||
self.assertTrue(
|
||||
'MesApi.post' in html or
|
||||
'MesApi.get' in html or
|
||||
'/static/dist/resource-status.js' in html
|
||||
)
|
||||
if response.status_code == 302:
|
||||
self.assertTrue(response.location.endswith('/portal-shell/resource'))
|
||||
self.assertEqual(final_response.status_code, 200)
|
||||
self.assertIn('/static/dist/portal-shell.js', html)
|
||||
else:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(
|
||||
'MesApi.post' in html or
|
||||
'MesApi.get' in html or
|
||||
'/static/dist/resource-status.js' in html
|
||||
)
|
||||
|
||||
def test_query_tool_page_uses_vite_module(self):
|
||||
response = self.client.get('/query-tool')
|
||||
@@ -310,11 +331,17 @@ class TestMesApiUsageInTemplates(unittest.TestCase):
|
||||
self.assertIn('type="module"', html)
|
||||
|
||||
def test_tmtt_defect_page_uses_vite_module(self):
|
||||
response = self.client.get('/tmtt-defect')
|
||||
html = response.data.decode('utf-8')
|
||||
response, final_response, html = _get_response_and_html(self.client, '/tmtt-defect')
|
||||
|
||||
self.assertIn('/static/dist/tmtt-defect.js', html)
|
||||
self.assertIn('type="module"', html)
|
||||
if response.status_code == 302:
|
||||
self.assertTrue(response.location.endswith('/portal-shell/tmtt-defect'))
|
||||
self.assertEqual(final_response.status_code, 200)
|
||||
self.assertIn('/static/dist/portal-shell.js', html)
|
||||
self.assertIn('type="module"', html)
|
||||
else:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn('/static/dist/tmtt-defect.js', html)
|
||||
self.assertIn('type="module"', html)
|
||||
|
||||
|
||||
class TestViteModuleIntegration(unittest.TestCase):
|
||||
@@ -341,13 +368,33 @@ class TestViteModuleIntegration(unittest.TestCase):
|
||||
('/query-tool', 'query-tool.js'),
|
||||
('/tmtt-defect', 'tmtt-defect.js'),
|
||||
]
|
||||
canonical_routes = {
|
||||
'/wip-overview': '/portal-shell/wip-overview',
|
||||
'/wip-detail': '/portal-shell/wip-detail',
|
||||
'/hold-overview': '/portal-shell/hold-overview',
|
||||
'/hold-detail?reason=test-reason': '/portal-shell/hold-detail?reason=test-reason',
|
||||
'/resource': '/portal-shell/resource',
|
||||
'/resource-history': '/portal-shell/resource-history',
|
||||
'/job-query': '/portal-shell/job-query',
|
||||
'/tmtt-defect': '/portal-shell/tmtt-defect',
|
||||
}
|
||||
|
||||
for endpoint, asset in endpoints_and_assets:
|
||||
with patch('mes_dashboard.app.os.path.exists', return_value=False):
|
||||
response = self.client.get(endpoint)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
html = response.data.decode('utf-8')
|
||||
self.assertIn(f'/static/dist/{asset}', html)
|
||||
self.assertIn('type="module"', html)
|
||||
response = self.client.get(endpoint, follow_redirects=False)
|
||||
|
||||
if endpoint in canonical_routes:
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith(canonical_routes[endpoint]))
|
||||
follow = self.client.get(response.location)
|
||||
self.assertEqual(follow.status_code, 200)
|
||||
html = follow.data.decode('utf-8')
|
||||
self.assertIn('/static/dist/portal-shell.js', html)
|
||||
self.assertIn('type="module"', html)
|
||||
else:
|
||||
self.assertEqual(response.status_code, 200)
|
||||
html = response.data.decode('utf-8')
|
||||
self.assertIn(f'/static/dist/{asset}', html)
|
||||
self.assertIn('type="module"', html)
|
||||
|
||||
|
||||
class TestStaticFilesServing(unittest.TestCase):
|
||||
|
||||
@@ -22,22 +22,18 @@ def client():
|
||||
|
||||
|
||||
def test_wip_pages_render_vite_assets(client):
|
||||
"""Core WIP/Hold pages should render Vite bundles."""
|
||||
overview = client.get("/wip-overview")
|
||||
detail = client.get("/wip-detail")
|
||||
hold = client.get("/hold-detail?reason=YieldLimit")
|
||||
"""Core WIP/Hold direct entries should redirect to canonical shell routes."""
|
||||
overview = client.get("/wip-overview", follow_redirects=False)
|
||||
detail = client.get("/wip-detail", follow_redirects=False)
|
||||
hold = client.get("/hold-detail?reason=YieldLimit", follow_redirects=False)
|
||||
|
||||
assert overview.status_code == 200
|
||||
assert detail.status_code == 200
|
||||
assert hold.status_code == 200
|
||||
assert overview.status_code == 302
|
||||
assert detail.status_code == 302
|
||||
assert hold.status_code == 302
|
||||
|
||||
overview_html = overview.data.decode("utf-8")
|
||||
detail_html = detail.data.decode("utf-8")
|
||||
hold_html = hold.data.decode("utf-8")
|
||||
|
||||
assert "/static/dist/wip-overview.js" in overview_html
|
||||
assert "/static/dist/wip-detail.js" in detail_html
|
||||
assert "/static/dist/hold-detail.js" in hold_html
|
||||
assert overview.location.endswith("/portal-shell/wip-overview")
|
||||
assert detail.location.endswith("/portal-shell/wip-detail")
|
||||
assert hold.location.endswith("/portal-shell/hold-detail?reason=YieldLimit")
|
||||
|
||||
|
||||
def test_wip_overview_and_detail_status_parameter_contract(client):
|
||||
|
||||
@@ -412,23 +412,26 @@ class TestMetaPackagesRoute(TestWipRoutesBase):
|
||||
self.assertFalse(data['success'])
|
||||
|
||||
|
||||
class TestPageRoutes(TestWipRoutesBase):
|
||||
"""Test page routes for WIP dashboards."""
|
||||
|
||||
def test_wip_overview_page_exists(self):
|
||||
"""GET /wip-overview should return 200."""
|
||||
response = self.client.get('/wip-overview')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_wip_detail_page_exists(self):
|
||||
"""GET /wip-detail should return 200."""
|
||||
response = self.client.get('/wip-detail')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_wip_detail_page_with_workcenter(self):
|
||||
"""GET /wip-detail?workcenter=xxx should return 200."""
|
||||
response = self.client.get('/wip-detail?workcenter=焊接_DB')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
class TestPageRoutes(TestWipRoutesBase):
|
||||
"""Test page routes for WIP dashboards."""
|
||||
|
||||
def test_wip_overview_page_exists(self):
|
||||
"""GET /wip-overview should redirect to canonical shell route."""
|
||||
response = self.client.get('/wip-overview', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/wip-overview'))
|
||||
|
||||
def test_wip_detail_page_exists(self):
|
||||
"""GET /wip-detail should redirect to canonical shell route."""
|
||||
response = self.client.get('/wip-detail', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertTrue(response.location.endswith('/portal-shell/wip-detail'))
|
||||
|
||||
def test_wip_detail_page_with_workcenter(self):
|
||||
"""GET /wip-detail?workcenter=xxx should preserve query during redirect."""
|
||||
response = self.client.get('/wip-detail?workcenter=焊接_DB', follow_redirects=False)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertIn('/portal-shell/wip-detail?workcenter=', response.location)
|
||||
|
||||
def test_old_wip_route_removed(self):
|
||||
"""GET /wip should return 404 (route removed)."""
|
||||
|
||||
Reference in New Issue
Block a user