diff --git a/tests/test_admin_observability_routes.py b/tests/test_admin_observability_routes.py new file mode 100644 index 0000000..0c4e012 --- /dev/null +++ b/tests/test_admin_observability_routes.py @@ -0,0 +1,176 @@ +"""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) + return flask_app + + +@pytest.fixture +def client(app): + 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 + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/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('/admin/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('/admin/ai_calls') + assert r.status_code == 200 # 失敗安全:仍 render,不 500 + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/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('/admin/promotion_review') + assert r.status_code == 200 + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/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('/admin/quality_trend') + assert r.status_code == 200 + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/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('/admin/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( + '/admin/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( + '/admin/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( + '/admin/budget/update/1', + json={'budget_usd': 25.50, 'alert_pct': 80}, + ) + assert r.status_code == 200 + assert r.get_json()['ok'] is True + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/ppt_audit_history +# ────────────────────────────────────────────────────────────────────────── + +def test_ppt_audit_history_200(client): + """無 DB 依賴,純掃 reports/。""" + r = client.get('/admin/ppt_audit_history') + assert r.status_code == 200 + + +# ────────────────────────────────────────────────────────────────────────── +# /admin/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('/admin/host_health') + assert r.status_code == 200