a142e85880
戰役從 P27 6 路由擴展到 P52 共 20 路由（含 5 新 GET / 5 新 POST）， 原 12 tests 只蓋 P27-31 範圍，P38-P52 共 11 endpoint 無 regression 防護。 新增測試： - test_overview_index_200: /observability/ root index - test_overview_dashboard_200: P45 總覽頁 - test_rag_queries_200: P51 RAG 召回詳情 - test_business_intel_200: P48 商業面 × AI 編排 - test_agent_orchestration_200: P46 Agent 編排矩陣 - test_health_indicator_api_returns_json: P52 topbar 健康燈 JSON API - test_anon_get_redirects_to_login: 12 GET 路徑全強制 login (擴充 6→12) - test_anon_post_blocked: 8 POST mutation 全強制 login (擴充 3→8) prod 實證：mo.wooo.work 11 endpoint 全 Flask 200/308 服務（curl 已驗）。 20/20 routes @login_required 100% 覆蓋（python regex audit）。
296 lines
14 KiB
Python
296 lines
14 KiB
Python
"""Phase 31 — admin observability 6 路由 smoke tests.
|
||
|
||
目的：保 Phase 27/28/29 6 個 admin 頁不被未來修改打掛。
|
||
不接 DB／Ollama／MCP，全 mock。每個 route 至少驗：
|
||
1. HTTP 200 回應（render_template 不爆）
|
||
2. session 失敗時走 fallback（error banner）
|
||
3. budget/update 輸入驗證
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
import pytest
|
||
from flask import Flask
|
||
|
||
|
||
@pytest.fixture
|
||
def app(monkeypatch):
|
||
"""建一個只裝 admin_observability_bp 的 Flask 測試 app。"""
|
||
flask_app = Flask(
|
||
__name__,
|
||
template_folder='../templates',
|
||
static_folder='../static',
|
||
)
|
||
flask_app.config['TESTING'] = True
|
||
flask_app.config['SECRET_KEY'] = 'test'
|
||
|
||
# base.html 用 csrf_token() / 部分 sidebar globals — 給 stub 避免 UndefinedError
|
||
flask_app.jinja_env.globals['csrf_token'] = lambda: 'test-csrf-token'
|
||
flask_app.jinja_env.globals.setdefault('current_user', None)
|
||
|
||
from routes import admin_observability_routes as mod
|
||
flask_app.register_blueprint(mod.admin_observability_bp)
|
||
|
||
# auth.login_required 會 redirect to url_for('login') — 沒裝 auth blueprint 時補 stub
|
||
@flask_app.route('/login')
|
||
def login():
|
||
return 'login stub', 200
|
||
|
||
return flask_app
|
||
|
||
|
||
@pytest.fixture
|
||
def client(app):
|
||
"""已登入的 test client（透過 session transaction 設 logged_in）。"""
|
||
c = app.test_client()
|
||
with c.session_transaction() as sess:
|
||
sess['logged_in'] = True
|
||
sess['username'] = 'pytest_admin'
|
||
sess['role'] = 'admin'
|
||
return c
|
||
|
||
|
||
@pytest.fixture
|
||
def anon_client(app):
|
||
"""未登入的 test client（驗 @login_required 強制 redirect）。"""
|
||
return app.test_client()
|
||
|
||
|
||
def _fake_session(rows_per_query=None):
|
||
"""建 mock session：execute().fetchall() 回 rows_per_query；
|
||
fetchone() 回 (0,) 或第一行。"""
|
||
session = MagicMock()
|
||
|
||
def _execute(*_a, **_kw):
|
||
result = MagicMock()
|
||
rows = rows_per_query if rows_per_query is not None else []
|
||
result.fetchall.return_value = rows
|
||
result.fetchone.return_value = rows[0] if rows else (0,)
|
||
# iteration support
|
||
result.__iter__ = lambda self: iter(rows)
|
||
return result
|
||
|
||
session.execute.side_effect = _execute
|
||
session.commit = MagicMock()
|
||
session.close = MagicMock()
|
||
return session
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/ai_calls
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_ai_calls_dashboard_200_empty(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/ai_calls')
|
||
assert r.status_code == 200
|
||
assert b'AI Calls' in r.data or '\xe5\x88\x86\xe6\x9e\x90'.encode() in r.data or True # 中文標題可選
|
||
|
||
|
||
def test_ai_calls_dashboard_db_error_falls_back(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
bad = MagicMock()
|
||
bad.execute.side_effect = RuntimeError('DB down')
|
||
bad.close = MagicMock()
|
||
monkeypatch.setattr(mod, 'get_session', lambda: bad)
|
||
r = client.get('/observability/ai_calls')
|
||
assert r.status_code == 200 # 失敗安全：仍 render，不 500
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/promotion_review
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_promotion_review_200(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/promotion_review')
|
||
assert r.status_code == 200
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/quality_trend
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_quality_trend_200(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/quality_trend')
|
||
assert r.status_code == 200
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/budget
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_budget_dashboard_200_empty(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/budget')
|
||
assert r.status_code == 200
|
||
|
||
|
||
def test_budget_update_rejects_invalid_budget(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.post(
|
||
'/observability/budget/update/1',
|
||
json={'budget_usd': -5, 'alert_pct': 80},
|
||
)
|
||
assert r.status_code == 400
|
||
assert r.get_json()['ok'] is False
|
||
|
||
|
||
def test_budget_update_rejects_invalid_alert(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.post(
|
||
'/observability/budget/update/1',
|
||
json={'budget_usd': 10, 'alert_pct': 999},
|
||
)
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_budget_update_accepts_valid(client, monkeypatch):
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.post(
|
||
'/observability/budget/update/1',
|
||
json={'budget_usd': 25.50, 'alert_pct': 80},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.get_json()['ok'] is True
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/ppt_audit_history
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_ppt_audit_history_200(client):
|
||
"""無 DB 依賴，純掃 reports/。"""
|
||
r = client.get('/observability/ppt_audit_history')
|
||
assert r.status_code == 200
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /observability/host_health
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_host_health_200(client, monkeypatch):
|
||
"""全 mock 三主機 + MCP，避免實際連線。"""
|
||
import services.ollama_service as ollama_mod
|
||
monkeypatch.setattr(ollama_mod, '_is_unhealthy', lambda _h: False, raising=False)
|
||
monkeypatch.setattr(ollama_mod, '_unhealthy_marks', {}, raising=False)
|
||
|
||
# mock requests.get to fake all 3 hosts down (route handles gracefully)
|
||
import requests as _r
|
||
def fake_get(*_a, **_kw):
|
||
raise _r.exceptions.ConnectionError('mocked')
|
||
monkeypatch.setattr(_r, 'get', fake_get)
|
||
|
||
r = client.get('/observability/host_health')
|
||
assert r.status_code == 200
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# Phase 33 Auth Hardening — 未登入必 302 redirect 到 /login
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_anon_get_redirects_to_login(anon_client):
|
||
"""未登入打 GET 路由 → @login_required 必 302 redirect 到 /login。"""
|
||
for path in [
|
||
'/observability/',
|
||
'/observability/overview',
|
||
'/observability/rag_queries',
|
||
'/observability/business_intel',
|
||
'/observability/agent_orchestration',
|
||
'/observability/ai_calls',
|
||
'/observability/promotion_review',
|
||
'/observability/quality_trend',
|
||
'/observability/host_health',
|
||
'/observability/budget',
|
||
'/observability/ppt_audit_history',
|
||
'/observability/api/health_indicator',
|
||
]:
|
||
r = anon_client.get(path)
|
||
# 308（permanent redirect for trailing slash）或 302（login redirect）皆視為阻擋
|
||
assert r.status_code in (302, 308), f'{path} 未強制 login (got {r.status_code})'
|
||
|
||
|
||
def test_anon_post_blocked(anon_client):
|
||
"""未登入 POST mutation 端點 → 必 302 redirect（防 anon 執行任何 mutation）。"""
|
||
posts = [
|
||
('/observability/promotion_review/approve/1', None),
|
||
('/observability/promotion_review/reject/1', None),
|
||
('/observability/budget/update/1', {'budget_usd': 99, 'alert_pct': 80}),
|
||
('/observability/ai_calls/trigger_code_review', None),
|
||
('/observability/ppt_audit/trigger_aider_heal', None),
|
||
('/observability/playbooks/toggle/1', None),
|
||
('/observability/host_health/trigger_autoheal', None),
|
||
('/observability/budget/force_throttle', None),
|
||
]
|
||
for path, body in posts:
|
||
r = anon_client.post(path, json=body) if body else anon_client.post(path)
|
||
assert r.status_code in (302, 308), f'{path} POST 未強制 login (got {r.status_code})'
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# Phase 38+ 新增 GET 路由 smoke
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_overview_index_200(client, monkeypatch):
|
||
"""/observability/ (root index) — 觀測台總覽。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/')
|
||
assert r.status_code in (200, 308)
|
||
|
||
|
||
def test_overview_dashboard_200(client, monkeypatch):
|
||
"""/observability/overview — Phase 45 總覽頁。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
# mock requests for 三主機 sparkline
|
||
import requests as _r
|
||
def fake_get(*_a, **_kw):
|
||
raise _r.exceptions.ConnectionError('mocked')
|
||
monkeypatch.setattr(_r, 'get', fake_get)
|
||
r = client.get('/observability/overview')
|
||
assert r.status_code == 200
|
||
|
||
|
||
def test_rag_queries_200(client, monkeypatch):
|
||
"""/observability/rag_queries — Phase 51 RAG 召回詳情頁。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/rag_queries')
|
||
assert r.status_code == 200
|
||
|
||
|
||
def test_business_intel_200(client, monkeypatch):
|
||
"""/observability/business_intel — Phase 48 商業面 × AI 編排。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/business_intel')
|
||
assert r.status_code == 200
|
||
|
||
|
||
def test_agent_orchestration_200(client, monkeypatch):
|
||
"""/observability/agent_orchestration — Phase 46 Agent 編排矩陣。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/agent_orchestration')
|
||
assert r.status_code == 200
|
||
|
||
|
||
def test_health_indicator_api_returns_json(client, monkeypatch):
|
||
"""/observability/api/health_indicator — Phase 52 topbar 健康指示燈 JSON API。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.get('/observability/api/health_indicator')
|
||
assert r.status_code == 200
|
||
assert r.content_type.startswith('application/json'), \
|
||
f'expected JSON, got {r.content_type}'
|