1418 lines
54 KiB
Python
1418 lines
54 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 datetime import datetime
|
||
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
|
||
|
||
|
||
def test_ai_calls_recent_row_marks_legacy_code_review_gemini_as_backup():
|
||
from routes.admin_observability_routes import _build_ai_call_recent_row
|
||
|
||
row = (
|
||
1452,
|
||
datetime(2026, 5, 19, 12, 2, 49),
|
||
'code_review_openclaw',
|
||
'gemini',
|
||
'gemini-2.5-flash',
|
||
181,
|
||
56,
|
||
9158,
|
||
'ok',
|
||
0,
|
||
False,
|
||
False,
|
||
)
|
||
|
||
data = _build_ai_call_recent_row(row)
|
||
|
||
assert data['caller'] == 'code_review_openclaw'
|
||
assert data['caller_display'] == 'code_review_openclaw_gemini'
|
||
assert data['provider'] == 'gemini'
|
||
assert 'Gemini 備援' in data['route_badges']
|
||
assert '舊 caller' in data['route_badges']
|
||
|
||
|
||
def test_ai_calls_recent_row_marks_legacy_openclaw_qa_gemini_as_backup():
|
||
from routes.admin_observability_routes import _build_ai_call_recent_row
|
||
|
||
row = (
|
||
1460,
|
||
datetime(2026, 5, 19, 12, 12, 49),
|
||
'openclaw_qa',
|
||
'gemini',
|
||
'gemini-2.5-flash',
|
||
181,
|
||
56,
|
||
9158,
|
||
'ok',
|
||
0,
|
||
False,
|
||
False,
|
||
)
|
||
|
||
data = _build_ai_call_recent_row(row)
|
||
|
||
assert data['caller'] == 'openclaw_qa'
|
||
assert data['caller_display'] == 'openclaw_qa_gemini_fallback'
|
||
assert data['provider'] == 'gemini'
|
||
assert 'Gemini 備援' in data['route_badges']
|
||
assert '舊 caller' in data['route_badges']
|
||
|
||
|
||
def test_ai_calls_recent_row_marks_new_openclaw_qa_gemini_fallback_as_backup():
|
||
from routes.admin_observability_routes import _build_ai_call_recent_row
|
||
|
||
row = (
|
||
1461,
|
||
datetime(2026, 5, 19, 12, 13, 49),
|
||
'openclaw_qa_gemini_fallback',
|
||
'gemini',
|
||
'gemini-2.5-flash',
|
||
181,
|
||
56,
|
||
9158,
|
||
'ok',
|
||
0,
|
||
False,
|
||
False,
|
||
)
|
||
|
||
data = _build_ai_call_recent_row(row)
|
||
|
||
assert data['caller_display'] == 'openclaw_qa_gemini_fallback'
|
||
assert 'Gemini 備援' in data['route_badges']
|
||
assert '舊 caller' not in data['route_badges']
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /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
|
||
|
||
|
||
def test_ppt_vision_runtime_status_explains_blockers(monkeypatch):
|
||
"""視覺 QA runtime 狀態要能直接說明停用原因與下一步。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
monkeypatch.delenv('PPT_VISION_ENABLED', raising=False)
|
||
monkeypatch.setenv('PPT_VISION_MODEL', 'minicpm-v:test')
|
||
monkeypatch.setattr(svc.shutil, 'which', lambda _name: None)
|
||
|
||
status = svc.get_ppt_vision_runtime_status()
|
||
|
||
assert status['ready'] is False
|
||
assert status['status_label'] == '環境未就緒'
|
||
assert status['ready_count'] == 1
|
||
assert status['check_count'] == 3
|
||
assert 'PPT_VISION_ENABLED 未設定為 true' in status['blockers']
|
||
assert '容器內缺少 LibreOffice' in status['blockers'][1]
|
||
assert status['readiness_checks'][0]['key'] == 'feature_flag'
|
||
assert status['readiness_checks'][0]['status'] == 'error'
|
||
assert status['readiness_checks'][1]['key'] == 'converter'
|
||
assert status['readiness_checks'][1]['status'] == 'error'
|
||
assert status['readiness_checks'][2]['label'] == '視覺模型'
|
||
assert status['readiness_checks'][2]['value'] == 'minicpm-v:test'
|
||
assert any('PPT_VISION_ENABLED=true' in action for action in status['next_actions'])
|
||
|
||
|
||
def test_ppt_audit_history_renders_vision_runtime_checklist(client, monkeypatch):
|
||
"""PPT 產線頁不應只顯示停用，要呈現 runtime checklist 與修復動作。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
monkeypatch.setattr(svc, 'get_ppt_vision_runtime_status', lambda: {
|
||
'enabled': False,
|
||
'env_value': '未設定（預設 false）',
|
||
'model': 'minicpm-v:test',
|
||
'converter': None,
|
||
'converter_ready': False,
|
||
'blockers': ['PPT_VISION_ENABLED 未設定為 true', '容器內缺少 LibreOffice'],
|
||
'ready': False,
|
||
'ready_count': 1,
|
||
'check_count': 3,
|
||
'status_label': '環境未就緒',
|
||
'summary': '視覺 QA runtime 仍有必要條件未通過。',
|
||
'checked_at': '2026-05-19 10:00:00',
|
||
'readiness_checks': [
|
||
{
|
||
'key': 'feature_flag',
|
||
'label': '功能開關',
|
||
'value': 'PPT_VISION_ENABLED=未設定',
|
||
'status': 'error',
|
||
'detail': '目前會阻擋立即視覺 QA 按鈕。',
|
||
},
|
||
{
|
||
'key': 'converter',
|
||
'label': '轉檔器',
|
||
'value': 'not found',
|
||
'status': 'error',
|
||
'detail': '缺轉檔器時無法建立視覺模型輸入。',
|
||
},
|
||
{
|
||
'key': 'vision_model',
|
||
'label': '視覺模型',
|
||
'value': 'minicpm-v:test',
|
||
'status': 'ready',
|
||
'detail': '沿用 Ollama-first 三主機路由。',
|
||
},
|
||
],
|
||
'next_actions': [
|
||
'在 momo-app / scheduler 環境設定 PPT_VISION_ENABLED=true，重新 recreate 相關 app 容器。',
|
||
'確認映像已安裝 LibreOffice Impress，完成 rebuild 後再重啟 momo-app / scheduler。',
|
||
],
|
||
})
|
||
|
||
r = client.get('/observability/ppt_audit_history')
|
||
html = r.get_data(as_text=True)
|
||
|
||
assert r.status_code == 200
|
||
assert '視覺 QA runtime checklist' in html
|
||
assert 'PPT_VISION_ENABLED=未設定' in html
|
||
assert 'LibreOffice' in html
|
||
assert 'minicpm-v:test' in html
|
||
assert '1/3 通過' in html
|
||
assert '重新 recreate 相關 app 容器' in html
|
||
assert '阻擋：PPT_VISION_ENABLED 未設定為 true；容器內缺少 LibreOffice' in html
|
||
|
||
|
||
def test_ppt_audit_history_shows_ppt_schedule_and_db_runs(client, monkeypatch):
|
||
"""PPT 產線頁必須呈現六種固定週期與 DB 寫入紀錄。"""
|
||
from services import ppt_auto_generation_service as svc
|
||
|
||
cadences = svc.get_schedule_cadence_status([
|
||
{'key': key, 'ready': True}
|
||
for key in svc.DEFINED_REPORT_TYPES
|
||
])
|
||
monkeypatch.setattr(svc, 'get_defined_report_coverage', lambda **_kw: {
|
||
'enabled': True,
|
||
'items': [{
|
||
'key': 'daily',
|
||
'label': '每日日報',
|
||
'target_label': '2026/05/17',
|
||
'ready': False,
|
||
'status': 'missing',
|
||
'status_label': '待排程補齊',
|
||
'status_hint': '尚未找到符合定義的檔案或 DB 紀錄。',
|
||
}],
|
||
'missing_report_types': ['daily'],
|
||
'missing_count': 1,
|
||
'ready_count': len(svc.DEFINED_REPORT_TYPES) - 1,
|
||
'total': len(svc.DEFINED_REPORT_TYPES),
|
||
'last_run': None,
|
||
'cadences': cadences,
|
||
'cadence_summary': '、'.join(c['schedule_text'] for c in cadences),
|
||
})
|
||
monkeypatch.setattr(svc, 'get_generation_run_history', lambda **_kw: [
|
||
{
|
||
'schedule_kind': 'daily',
|
||
'schedule_label': '每日',
|
||
'report_type': 'daily',
|
||
'report_label': '每日日報',
|
||
'target_label': '2026/05/17',
|
||
'status': 'error',
|
||
'status_label': '失敗',
|
||
'file_name': '',
|
||
'file_size_kb': None,
|
||
'error_msg': '產線錯誤：缺少資料來源',
|
||
'started_at': '2026-05-17 20:29',
|
||
'finished_at': '2026-05-17 20:30',
|
||
},
|
||
{
|
||
'schedule_kind': 'daily',
|
||
'schedule_label': '每日',
|
||
'report_type': 'daily',
|
||
'report_label': '每日日報',
|
||
'target_label': '2026/05/17',
|
||
'status': 'ready',
|
||
'status_label': '已產出',
|
||
'file_name': 'ocbot_daily_20260517.pptx',
|
||
'file_size_kb': 1024,
|
||
'error_msg': '',
|
||
'started_at': '2026-05-17 20:30',
|
||
'finished_at': '2026-05-17 20:31',
|
||
},
|
||
])
|
||
|
||
r = client.get('/observability/ppt_audit_history')
|
||
html = r.data.decode('utf-8')
|
||
|
||
for text in ['每日 20:30', '每週一 20:40', '每月 1 日 20:50', '每季首日 21:00', '每半年首日 21:10', '每年 1/1 21:20']:
|
||
assert text in html
|
||
assert 'ppt_generation_runs' in html
|
||
assert '每日日報' in html
|
||
assert '產線健康度' in html
|
||
assert '排程節奏' in html
|
||
assert 'DB 寫入' in html
|
||
assert '線上預覽' in html
|
||
assert '工作隊列' in html
|
||
assert '接下來要處理的事' in html
|
||
assert '異常優先' in html
|
||
assert '產線錯誤：缺少資料來源' in html
|
||
assert 'data-ppt-generate-one' in html
|
||
assert 'data-report-type="daily"' in html
|
||
assert '重跑' in html
|
||
assert '待排程補齊' in html
|
||
|
||
|
||
def test_ppt_audit_history_shows_recent_preview_workbench(client, monkeypatch, tmp_path):
|
||
"""有檔案時，頁面上方要先給可預覽簡報入口。"""
|
||
import zipfile
|
||
from pathlib import Path
|
||
from services import ppt_preview_service as preview_svc
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
cache_dir = tmp_path / 'preview-cache'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260517.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
monkeypatch.setenv('PPT_PREVIEW_CACHE_DIR', str(cache_dir))
|
||
cache_info = preview_svc.get_ppt_preview_cache_info(pptx)
|
||
Path(cache_info.pdf_path).parent.mkdir(parents=True, exist_ok=True)
|
||
Path(cache_info.pdf_path).write_bytes(b'%PDF-1.4\n')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
|
||
r = client.get('/observability/ppt_audit_history?month=2026-05')
|
||
html = r.data.decode('utf-8')
|
||
|
||
assert r.status_code == 200
|
||
assert '最近可預覽簡報' in html
|
||
assert 'ocbot_daily_20260517.pptx' in html
|
||
assert '線上預覽' in html
|
||
assert 'data-ppt-open-preview' in html
|
||
assert 'data-ppt-preview-modal' in html
|
||
assert 'data-ppt-preview-modal-title' in html
|
||
assert 'data-ppt-preview-pdf' in html
|
||
assert 'PDF 快取' in html
|
||
assert '1</strong> 份 PDF 快取' in html
|
||
|
||
|
||
def test_ppt_audit_history_coverage_matrix_joins_db_preview_qa(client, monkeypatch, tmp_path):
|
||
"""定義簡報覆蓋區要同列呈現 DB、預覽、QA 與交付狀態。"""
|
||
import zipfile
|
||
from datetime import datetime
|
||
from unittest.mock import MagicMock
|
||
from routes import admin_observability_routes as mod
|
||
from services import ppt_auto_generation_service as svc
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260517.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
monkeypatch.setenv('PPT_PREVIEW_CACHE_DIR', str(tmp_path / 'preview-cache'))
|
||
|
||
coverage_items = [{
|
||
'key': 'daily',
|
||
'label': '每日日報',
|
||
'target_label': '2026/05/17',
|
||
'ready': True,
|
||
'status': 'ready',
|
||
'status_label': '已產出',
|
||
'status_hint': '檔案參數與本期定義相符。',
|
||
'sources': ['database', 'filesystem'],
|
||
'latest_generated_at': '2026-05-17 20:31',
|
||
'latest_file_path': str(pptx),
|
||
'latest_file_name': pptx.name,
|
||
}]
|
||
monkeypatch.setattr(svc, 'get_defined_report_coverage', lambda **_kw: {
|
||
'enabled': True,
|
||
'items': coverage_items,
|
||
'missing_report_types': [],
|
||
'missing_count': 0,
|
||
'ready_count': 1,
|
||
'total': 1,
|
||
'last_run': None,
|
||
'can_auto_start': False,
|
||
'cadences': svc.get_schedule_cadence_status(coverage_items),
|
||
'cadence_summary': '每日 20:30',
|
||
})
|
||
monkeypatch.setattr(svc, 'get_generation_run_history', lambda **_kw: [{
|
||
'schedule_kind': 'daily',
|
||
'schedule_label': '每日',
|
||
'report_type': 'daily',
|
||
'report_label': '每日日報',
|
||
'target_label': '2026/05/17',
|
||
'status': 'ready',
|
||
'status_label': '已產出',
|
||
'file_name': pptx.name,
|
||
'file_size_kb': 1024,
|
||
'error_msg': '',
|
||
'started_at': '2026-05-17 20:30',
|
||
'finished_at': '2026-05-17 20:31',
|
||
}])
|
||
|
||
class FakeSession:
|
||
def execute(self, statement, _params=None):
|
||
sql = str(statement)
|
||
result = MagicMock()
|
||
if 'FROM ppt_reports' in sql:
|
||
result.fetchall.return_value = []
|
||
elif 'SELECT audited_at, pptx_filename' in sql:
|
||
result.fetchall.return_value = [
|
||
(datetime(2026, 5, 17, 22, 5), pptx.name, 'passed', 0, 0.95, 900, '', [])
|
||
]
|
||
elif 'COALESCE(AVG(confidence)' in sql:
|
||
result.fetchone.return_value = (1, 1, 0, 0, 0, 0.95, 0)
|
||
elif 'GROUP BY pptx_filename' in sql:
|
||
result.fetchall.return_value = []
|
||
else:
|
||
result.fetchall.return_value = []
|
||
result.fetchone.return_value = (0,)
|
||
return result
|
||
|
||
def close(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(mod, 'get_session', lambda: FakeSession())
|
||
|
||
r = client.get('/observability/ppt_audit_history?month=2026-05')
|
||
html = r.get_data(as_text=True)
|
||
|
||
assert r.status_code == 200
|
||
assert '報表覆蓋矩陣' in html
|
||
assert 'DB / 預覽 / 視覺 QA / 交付' in html
|
||
assert 'DB 已寫入' in html
|
||
assert '可預覽' in html
|
||
assert 'QA 通過' in html
|
||
assert '可交付' in html
|
||
assert 'data-ppt-open-preview' in html
|
||
assert 'ocbot_daily_20260517.pptx' in html
|
||
|
||
|
||
def test_ppt_audit_file_view_renders_online_preview(client, monkeypatch, tmp_path):
|
||
"""PPTX view 入口應回站內預覽頁，而不是把 PPTX 直接丟給瀏覽器。"""
|
||
import zipfile
|
||
from services import ppt_preview_service as preview_svc
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260517.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
monkeypatch.setattr(
|
||
preview_svc,
|
||
'build_ppt_preview',
|
||
lambda *_args, **_kwargs: preview_svc.PPTPreviewResult(
|
||
ok=True,
|
||
pdf_path=str(reports_dir / 'preview.pdf'),
|
||
cache_hit=True,
|
||
converter='/usr/bin/libreoffice',
|
||
),
|
||
)
|
||
|
||
r = client.get('/observability/ppt_audit_file/ocbot_daily_20260517.pptx')
|
||
html = r.data.decode('utf-8')
|
||
|
||
assert r.status_code == 200
|
||
assert 'PPT 線上預覽' in html
|
||
assert 'action=pdf' in html
|
||
assert '下載 PPTX' in html
|
||
|
||
|
||
def test_ppt_audit_history_audit_rows_include_inline_replay(client, monkeypatch, tmp_path):
|
||
"""審核歷史每筆紀錄都應能直接開同頁 PDF 回放。"""
|
||
import zipfile
|
||
from datetime import datetime
|
||
from unittest.mock import MagicMock
|
||
from routes import admin_observability_routes as mod
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260517.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
|
||
class FakeSession:
|
||
def execute(self, statement, _params=None):
|
||
sql = str(statement)
|
||
result = MagicMock()
|
||
if 'FROM ppt_reports' in sql:
|
||
rows = []
|
||
result.fetchall.return_value = rows
|
||
elif 'SELECT audited_at, pptx_filename' in sql:
|
||
rows = [
|
||
(
|
||
datetime(2026, 5, 17, 22, 5),
|
||
'ocbot_daily_20260517.pptx',
|
||
'failed',
|
||
1,
|
||
0.93,
|
||
1200,
|
||
'',
|
||
[{'slide': 1, 'issues': ['⚠️ 圖表被切掉：右側圖例超出邊界']}],
|
||
)
|
||
]
|
||
result.fetchall.return_value = rows
|
||
elif 'COALESCE(AVG(confidence)' in sql:
|
||
row = (1, 1, 0, 0, 0, 0.93, 0)
|
||
result.fetchone.return_value = row
|
||
elif 'GROUP BY pptx_filename' in sql:
|
||
rows = []
|
||
result.fetchall.return_value = rows
|
||
else:
|
||
rows = []
|
||
result.fetchall.return_value = rows
|
||
result.fetchone.return_value = (0,)
|
||
return result
|
||
|
||
def close(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(mod, 'get_session', lambda: FakeSession())
|
||
|
||
r = client.get('/observability/ppt_audit_history?month=2026-05')
|
||
html = r.data.decode('utf-8')
|
||
|
||
assert r.status_code == 200
|
||
assert '審核回放 · ocbot_daily_20260517.pptx' in html
|
||
assert 'data-ppt-open-preview' in html
|
||
assert 'ocbot_daily_20260517.pptx?action=pdf' in html
|
||
assert '圖表被切掉' in html
|
||
assert '視覺問題追蹤' in html
|
||
assert '版面越界' in html
|
||
assert '問題回放 · ocbot_daily_20260517.pptx' in html
|
||
assert '回放' in html
|
||
|
||
|
||
def test_ppt_audit_history_weekly_rows_include_visual_audit(client, monkeypatch, tmp_path):
|
||
"""非 daily 簡報也應顯示自己的視覺 QA 歷史，不再要求切回 daily。"""
|
||
import zipfile
|
||
from datetime import datetime
|
||
from unittest.mock import MagicMock
|
||
from routes import admin_observability_routes as mod
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_weekly_20260518.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
|
||
class FakeSession:
|
||
def execute(self, statement, _params=None):
|
||
sql = str(statement)
|
||
result = MagicMock()
|
||
if 'FROM ppt_reports' in sql:
|
||
result.fetchall.return_value = []
|
||
elif 'SELECT audited_at, pptx_filename' in sql:
|
||
result.fetchall.return_value = [
|
||
(datetime(2026, 5, 18, 22, 7), 'ocbot_weekly_20260518.pptx', 'passed', 0, 0.91, 1800, '', [])
|
||
]
|
||
elif 'COALESCE(AVG(confidence)' in sql:
|
||
result.fetchone.return_value = (1, 1, 0, 0, 0, 0.91, 0)
|
||
elif 'GROUP BY pptx_filename' in sql:
|
||
result.fetchall.return_value = []
|
||
else:
|
||
result.fetchall.return_value = []
|
||
result.fetchone.return_value = (0,)
|
||
return result
|
||
|
||
def close(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(mod, 'get_session', lambda: FakeSession())
|
||
|
||
r = client.get('/observability/ppt_audit_history?month=2026-05&report_type=weekly')
|
||
html = r.data.decode('utf-8')
|
||
|
||
assert r.status_code == 200
|
||
assert 'ocbot_weekly_20260518.pptx' in html
|
||
assert '審核回放 · ocbot_weekly_20260518.pptx' in html
|
||
assert '只有「每日日報」會進入視覺審核流程' not in html
|
||
assert 'data-ppt-run-vision' in html
|
||
|
||
|
||
def test_ppt_audit_history_shows_preview_prewarm_action(client, monkeypatch, tmp_path):
|
||
"""未快取 PDF 的 PPTX 要能在產線清單直接預熱預覽。"""
|
||
import zipfile
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260518.pptx'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
monkeypatch.setenv('PPT_PREVIEW_CACHE_DIR', str(tmp_path / 'preview-cache'))
|
||
|
||
r = client.get('/observability/ppt_audit_history?month=2026-05')
|
||
html = r.data.decode('utf-8')
|
||
|
||
assert r.status_code == 200
|
||
assert '首次轉檔' in html
|
||
assert 'data-ppt-prewarm-all' in html
|
||
assert 'data-ppt-prewarm-count' in html
|
||
assert 'data-ppt-prewarm-preview' in html
|
||
assert 'data-ppt-filename="ocbot_daily_20260518.pptx"' in html
|
||
assert '預熱本頁 PDF' in html
|
||
assert '預熱 PDF' in html
|
||
|
||
|
||
def test_ppt_audit_file_prewarm_builds_preview_cache(client, monkeypatch, tmp_path):
|
||
"""預熱端點應回 JSON，不直接把 PDF body 丟給前端。"""
|
||
import zipfile
|
||
from services import ppt_preview_service as preview_svc
|
||
|
||
reports_dir = tmp_path / 'reports'
|
||
reports_dir.mkdir()
|
||
pptx = reports_dir / 'ocbot_daily_20260518.pptx'
|
||
pdf = reports_dir / 'preview.pdf'
|
||
with zipfile.ZipFile(pptx, 'w') as zf:
|
||
zf.writestr('[Content_Types].xml', '<Types></Types>')
|
||
|
||
monkeypatch.setenv('REPORTS_DIR', str(reports_dir))
|
||
monkeypatch.setattr(
|
||
preview_svc,
|
||
'build_ppt_preview',
|
||
lambda *_args, **_kwargs: preview_svc.PPTPreviewResult(
|
||
ok=True,
|
||
pdf_path=str(pdf),
|
||
cache_hit=False,
|
||
converter='/usr/bin/libreoffice',
|
||
),
|
||
)
|
||
|
||
r = client.post('/observability/ppt_audit_file/ocbot_daily_20260518.pptx/prewarm')
|
||
data = r.get_json()
|
||
|
||
assert r.status_code == 200
|
||
assert data['ok'] is True
|
||
assert data['filename'] == 'ocbot_daily_20260518.pptx'
|
||
assert data['message'] == 'PDF 預覽快取已建立'
|
||
|
||
|
||
def test_ppt_audit_run_vision_queues_background_audit(client, monkeypatch):
|
||
"""立即視覺 QA 端點只排入背景任務，不讓瀏覽器等待模型跑完。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
captured = {}
|
||
|
||
def fake_start(**kwargs):
|
||
captured.update(kwargs)
|
||
return {'ok': True, 'status': 'queued', 'message': 'PPT vision audit queued.'}
|
||
|
||
monkeypatch.setattr(svc, 'start_ppt_vision_audit_background', fake_start)
|
||
|
||
r = client.post(
|
||
'/observability/ppt_audit/run_vision',
|
||
json={'filenames': ['ocbot_daily_20260518.pptx'], 'max_files': 1},
|
||
)
|
||
data = r.get_json()
|
||
|
||
assert r.status_code == 202
|
||
assert data['ok'] is True
|
||
assert captured['filenames'] == ['ocbot_daily_20260518.pptx']
|
||
assert captured['max_files'] == 1
|
||
|
||
|
||
def test_ppt_vision_audit_status_sanitizes_last_run(monkeypatch, tmp_path):
|
||
"""背景視覺 QA 狀態只回檔名與摘要，不把 reports_dir 絕對路徑曝露到頁面。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
monkeypatch.setenv('PPT_VISION_STATE_PATH', str(tmp_path / 'vision_state.json'))
|
||
monkeypatch.setattr(svc, '_LAST_AUDIT_RUN', {
|
||
'ok': True,
|
||
'status': 'completed',
|
||
'queued_at': '2026-05-19 12:00:00',
|
||
'started_at': '2026-05-19 12:00:01',
|
||
'finished_at': '2026-05-19 12:00:05',
|
||
'filenames': ['/app/data/reports/ocbot_daily_20260518.pptx'],
|
||
'max_files': 1,
|
||
'summary': {
|
||
'audited_files': [{
|
||
'path': '/app/data/reports/ocbot_daily_20260518.pptx',
|
||
'slides_checked': 1,
|
||
'issues': 0,
|
||
'error': None,
|
||
}],
|
||
'total_issues': 0,
|
||
'errors': [],
|
||
},
|
||
})
|
||
|
||
status = svc.get_ppt_vision_audit_status()
|
||
|
||
assert status['ok'] is True
|
||
assert status['status'] == 'completed'
|
||
assert status['status_label'] == '已完成'
|
||
assert status['last_run']['filenames'] == ['ocbot_daily_20260518.pptx']
|
||
assert status['last_run']['summary']['audited_count'] == 1
|
||
assert status['last_run']['summary']['files'][0]['filename'] == 'ocbot_daily_20260518.pptx'
|
||
assert '/app/data/reports' not in str(status)
|
||
|
||
|
||
def test_ppt_vision_audit_status_reads_persisted_state(monkeypatch, tmp_path):
|
||
"""多 worker 下狀態需從 runtime state 檔讀取，不能只靠單一 worker 記憶體。"""
|
||
import json
|
||
from services import ppt_vision_service as svc
|
||
|
||
state_path = tmp_path / 'vision_state.json'
|
||
monkeypatch.setenv('PPT_VISION_STATE_PATH', str(state_path))
|
||
monkeypatch.setattr(svc, '_LAST_AUDIT_RUN', None)
|
||
state_path.write_text(json.dumps({
|
||
'ok': True,
|
||
'status': 'completed',
|
||
'queued_at': '2026-05-19 12:00:00',
|
||
'started_at': '2026-05-19 12:00:01',
|
||
'finished_at': '2026-05-19 12:00:05',
|
||
'updated_at': '2026-05-19 12:00:05',
|
||
'filenames': ['/app/data/reports/ocbot_daily_20260518.pptx'],
|
||
'max_files': 1,
|
||
'summary': {
|
||
'audited_files': [{
|
||
'path': '/app/data/reports/ocbot_daily_20260518.pptx',
|
||
'slides_checked': 1,
|
||
'issues': 0,
|
||
'error': None,
|
||
}],
|
||
'total_issues': 0,
|
||
'errors': [],
|
||
},
|
||
}), encoding='utf-8')
|
||
|
||
status = svc.get_ppt_vision_audit_status()
|
||
|
||
assert status['status'] == 'completed'
|
||
assert status['last_run']['updated_at'] == '2026-05-19 12:00:05'
|
||
assert status['last_run']['filenames'] == ['ocbot_daily_20260518.pptx']
|
||
assert '/app/data/reports' not in str(status)
|
||
|
||
|
||
def test_ppt_vision_audit_background_respects_persisted_running(monkeypatch, tmp_path):
|
||
"""另一個 worker 已在跑時，新 request 應回 already_running，避免重複開模型任務。"""
|
||
import json
|
||
from services import ppt_vision_service as svc
|
||
|
||
state_path = tmp_path / 'vision_state.json'
|
||
monkeypatch.setenv('PPT_VISION_STATE_PATH', str(state_path))
|
||
monkeypatch.setattr(svc, '_LAST_AUDIT_RUN', None)
|
||
state_path.write_text(json.dumps({
|
||
'ok': True,
|
||
'status': 'running',
|
||
'queued_at': '2999-05-19 12:00:00',
|
||
'started_at': '2999-05-19 12:00:01',
|
||
'updated_at': '2999-05-19 12:00:01',
|
||
'filenames': ['/app/data/reports/ocbot_daily_20260518.pptx'],
|
||
'max_files': 1,
|
||
}), encoding='utf-8')
|
||
|
||
result = svc.start_ppt_vision_audit_background(
|
||
filenames=['ocbot_daily_20260518.pptx'],
|
||
max_files=1,
|
||
)
|
||
|
||
assert result['ok'] is True
|
||
assert result['status'] == 'already_running'
|
||
assert result['last_run']['filenames'] == ['ocbot_daily_20260518.pptx']
|
||
|
||
|
||
def test_ppt_vision_audit_status_marks_dead_worker_stale(monkeypatch, tmp_path):
|
||
"""部署 reload 後若舊 worker 已不存在，running state 要自動轉為可診斷錯誤。"""
|
||
import json
|
||
from services import ppt_vision_service as svc
|
||
|
||
state_path = tmp_path / 'vision_state.json'
|
||
monkeypatch.setenv('PPT_VISION_STATE_PATH', str(state_path))
|
||
monkeypatch.setattr(svc, '_LAST_AUDIT_RUN', None)
|
||
monkeypatch.setattr(svc, '_pid_exists', lambda _pid: False)
|
||
state_path.write_text(json.dumps({
|
||
'ok': True,
|
||
'status': 'running',
|
||
'queued_at': '2999-05-19 12:00:00',
|
||
'started_at': '2999-05-19 12:00:01',
|
||
'updated_at': '2999-05-19 12:00:01',
|
||
'pid': 999999,
|
||
'filenames': ['/app/data/reports/ocbot_daily_20260518.pptx'],
|
||
'max_files': 1,
|
||
}), encoding='utf-8')
|
||
|
||
status = svc.get_ppt_vision_audit_status()
|
||
|
||
assert status['running'] is False
|
||
assert status['status'] == 'error'
|
||
assert 'worker no longer running' in status['last_run']['error']
|
||
assert status['last_run']['filenames'] == ['ocbot_daily_20260518.pptx']
|
||
|
||
|
||
def test_ppt_vision_audit_background_allows_retry_after_dead_worker(monkeypatch, tmp_path):
|
||
"""dead PID 的 running state 不應阻擋下一次手動補跑。"""
|
||
import json
|
||
from services import ppt_vision_service as svc
|
||
|
||
state_path = tmp_path / 'vision_state.json'
|
||
monkeypatch.setenv('PPT_VISION_STATE_PATH', str(state_path))
|
||
monkeypatch.setattr(svc, '_LAST_AUDIT_RUN', None)
|
||
monkeypatch.setattr(svc, '_pid_exists', lambda _pid: False)
|
||
monkeypatch.setattr(svc, 'audit_recent_ppts', lambda **_kwargs: {
|
||
'audited_files': [],
|
||
'total_issues': 0,
|
||
'errors': [],
|
||
})
|
||
state_path.write_text(json.dumps({
|
||
'ok': True,
|
||
'status': 'running',
|
||
'queued_at': '2999-05-19 12:00:00',
|
||
'started_at': '2999-05-19 12:00:01',
|
||
'updated_at': '2999-05-19 12:00:01',
|
||
'pid': 999999,
|
||
'filenames': ['/app/data/reports/ocbot_daily_20260518.pptx'],
|
||
'max_files': 1,
|
||
}), encoding='utf-8')
|
||
|
||
result = svc.start_ppt_vision_audit_background(
|
||
filenames=['ocbot_daily_20260518.pptx'],
|
||
max_files=1,
|
||
)
|
||
|
||
assert result['ok'] is True
|
||
assert result['status'] == 'queued'
|
||
|
||
|
||
def test_ppt_audit_vision_status_route_returns_json(client, monkeypatch):
|
||
"""頁面輪詢用 status endpoint 要能回最近一次背景視覺 QA 狀態。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
monkeypatch.setattr(svc, 'get_ppt_vision_audit_status', lambda: {
|
||
'ok': True,
|
||
'running': False,
|
||
'status': 'completed',
|
||
'status_label': '已完成',
|
||
'message': '最近一次視覺 QA 已完成。',
|
||
'last_run': {
|
||
'summary': {'audited_count': 2, 'total_issues': 1, 'error_count': 0, 'errors': [], 'files': []},
|
||
},
|
||
})
|
||
|
||
r = client.get('/observability/ppt_audit/vision_status')
|
||
data = r.get_json()
|
||
|
||
assert r.status_code == 200
|
||
assert data['ok'] is True
|
||
assert data['status'] == 'completed'
|
||
assert data['last_run']['summary']['audited_count'] == 2
|
||
|
||
|
||
def test_ppt_audit_history_renders_last_vision_status(client, monkeypatch):
|
||
"""PPT 產線頁要在按下立即 QA 前後都看得到背景狀態卡。"""
|
||
from services import ppt_vision_service as svc
|
||
|
||
monkeypatch.setattr(svc, 'get_ppt_vision_audit_status', lambda: {
|
||
'ok': True,
|
||
'running': False,
|
||
'status': 'completed',
|
||
'status_label': '已完成',
|
||
'message': '最近一次視覺 QA 已完成。',
|
||
'last_run': {
|
||
'queued_at': '2026-05-19 12:00:00',
|
||
'started_at': '2026-05-19 12:00:01',
|
||
'finished_at': '2026-05-19 12:00:05',
|
||
'summary': {'audited_count': 2, 'total_issues': 1, 'error_count': 0, 'errors': [], 'files': []},
|
||
},
|
||
})
|
||
|
||
r = client.get('/observability/ppt_audit_history')
|
||
html = r.get_data(as_text=True)
|
||
|
||
assert r.status_code == 200
|
||
assert 'data-ppt-vision-status' in html
|
||
assert 'data-ppt-vision-status-title' in html
|
||
assert '最近一次視覺 QA 已完成。' in html
|
||
assert '2 份 / 1 問題' in html
|
||
assert '執行環境' in html
|
||
|
||
|
||
def test_ppt_audit_trigger_aider_heal_accepts_issue_summary(client, monkeypatch):
|
||
"""視覺 QA failed 常只有 issues_found；AiderHeal 應可吃診斷摘要派工。"""
|
||
from routes import admin_observability_routes as mod
|
||
from services import aider_heal_executor as svc
|
||
|
||
captured = {}
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
def fake_execute_code_fix(**kwargs):
|
||
captured.update(kwargs)
|
||
return {
|
||
'success': True,
|
||
'action': 'CODE_FIX',
|
||
'message': '已派出 AiderHeal',
|
||
'commit_sha': None,
|
||
'reverted': False,
|
||
}
|
||
|
||
monkeypatch.setattr(svc, 'execute_code_fix', fake_execute_code_fix)
|
||
|
||
class ImmediateThread:
|
||
def __init__(self, target, **_kwargs):
|
||
self.target = target
|
||
|
||
def start(self):
|
||
self.target()
|
||
|
||
monkeypatch.setattr(mod.threading, 'Thread', ImmediateThread)
|
||
|
||
r = client.post(
|
||
'/observability/ppt_audit/trigger_aider_heal',
|
||
json={
|
||
'pptx_filename': 'ocbot_daily_20260517.pptx',
|
||
'issue_summary': 'S1: 圖表被切掉：右側圖例超出邊界',
|
||
},
|
||
)
|
||
data = r.get_json()
|
||
|
||
assert r.status_code == 202
|
||
assert data['ok'] is True
|
||
assert data['status'] == 'queued'
|
||
assert captured['error_type'] == 'ppt_vision_audit_failure'
|
||
assert captured['target_file'] == 'services/ppt_generator.py'
|
||
assert 'ocbot_daily_20260517.pptx' in captured['error_message']
|
||
assert '圖表被切掉' in captured['error_message']
|
||
assert captured['context']['issue_summary'] == 'S1: 圖表被切掉：右側圖例超出邊界'
|
||
assert mod._PPT_AIDER_HEAL_ACTIVE == {}
|
||
|
||
|
||
def test_ppt_audit_trigger_aider_heal_dedupes_same_file(client, monkeypatch):
|
||
"""同一份 PPT 已在背景修復時，重複按鈕不應重開第二條 AiderHeal。"""
|
||
from routes import admin_observability_routes as mod
|
||
from services import aider_heal_executor as svc
|
||
|
||
calls = []
|
||
|
||
def fake_execute_code_fix(**kwargs):
|
||
calls.append(kwargs)
|
||
return {
|
||
'success': True,
|
||
'action': 'CODE_FIX',
|
||
'message': '不應在此測試執行',
|
||
'commit_sha': None,
|
||
'reverted': False,
|
||
}
|
||
|
||
class HoldingThread:
|
||
def __init__(self, target, **_kwargs):
|
||
self.target = target
|
||
|
||
def start(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(svc, 'execute_code_fix', fake_execute_code_fix)
|
||
monkeypatch.setattr(mod.threading, 'Thread', HoldingThread)
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
payload = {
|
||
'pptx_filename': 'ocbot_daily_20260517.pptx',
|
||
'issue_summary': 'S1: 圖表被切掉：右側圖例超出邊界',
|
||
}
|
||
first = client.post('/observability/ppt_audit/trigger_aider_heal', json=payload)
|
||
second = client.post('/observability/ppt_audit/trigger_aider_heal', json=payload)
|
||
|
||
try:
|
||
assert first.status_code == 202
|
||
assert first.get_json()['status'] == 'queued'
|
||
assert second.status_code == 202
|
||
assert second.get_json()['status'] == 'already_running'
|
||
assert calls == []
|
||
finally:
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
|
||
def test_ppt_audit_aider_heal_status_reports_active_jobs(client, monkeypatch):
|
||
"""背景 AiderHeal 已派工時，狀態端點要能讓頁面重新整理後看見執行中。"""
|
||
from routes import admin_observability_routes as mod
|
||
from services import aider_heal_executor as svc
|
||
|
||
def fake_execute_code_fix(**_kwargs):
|
||
return {'success': True, 'message': '不應在此測試執行'}
|
||
|
||
class HoldingThread:
|
||
def __init__(self, target, **_kwargs):
|
||
self.target = target
|
||
|
||
def start(self):
|
||
return None
|
||
|
||
monkeypatch.setattr(svc, 'execute_code_fix', fake_execute_code_fix)
|
||
monkeypatch.setattr(mod.threading, 'Thread', HoldingThread)
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
payload = {
|
||
'pptx_filename': 'ocbot_daily_20260517.pptx',
|
||
'issue_summary': 'S1: 圖表被切掉：右側圖例超出邊界',
|
||
}
|
||
first = client.post('/observability/ppt_audit/trigger_aider_heal', json=payload)
|
||
status = client.get('/observability/ppt_audit/aider_heal_status')
|
||
|
||
try:
|
||
assert first.status_code == 202
|
||
data = status.get_json()
|
||
assert status.status_code == 200
|
||
assert data['ok'] is True
|
||
assert data['active_count'] == 1
|
||
assert data['jobs'][0]['pptx_filename'] == 'ocbot_daily_20260517.pptx'
|
||
assert data['jobs'][0]['target_file'] == 'services/ppt_generator.py'
|
||
assert '圖表被切掉' in data['jobs'][0]['diagnosis']
|
||
finally:
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
|
||
def test_ppt_audit_history_shows_active_aider_heal_jobs(client):
|
||
"""PPT 產線頁要直接顯示正在背景修復的檔案。"""
|
||
from routes import admin_observability_routes as mod
|
||
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
mod._PPT_AIDER_HEAL_ACTIVE['ocbot_daily_20260517.pptx'] = {
|
||
'key': 'ocbot_daily_20260517.pptx',
|
||
'pptx_filename': 'ocbot_daily_20260517.pptx',
|
||
'target_file': 'services/ppt_generator.py',
|
||
'queued_at': '2026-05-18 14:42:00',
|
||
'diagnosis': 'S1: 圖表被切掉：右側圖例超出邊界',
|
||
}
|
||
try:
|
||
r = client.get('/observability/ppt_audit_history')
|
||
html = r.get_data(as_text=True)
|
||
assert r.status_code == 200
|
||
assert 'AiderHeal 執行中' in html
|
||
assert 'ocbot_daily_20260517.pptx' in html
|
||
assert '圖表被切掉' in html
|
||
finally:
|
||
mod._PPT_AIDER_HEAL_ACTIVE.clear()
|
||
|
||
|
||
def test_ppt_pipeline_view_regenerates_audit_failure_with_inferred_report_type():
|
||
"""QA 失敗不能硬編 daily；重跑要回到原本的簡報類型。"""
|
||
from routes import admin_observability_routes as mod
|
||
|
||
view = mod._build_ppt_pipeline_view(
|
||
files=[],
|
||
auto_generation={'items': [], 'ready_count': 0, 'total': 0, 'missing_count': 0},
|
||
audit_stats={'total': 1, 'total_issues': 2, 'pass_rate': 0},
|
||
generation_runs=[],
|
||
vision_status={'ready': True},
|
||
audit_records=[{
|
||
'pptx_filename': 'ocbot_ttm_8fd2ae22.pptx',
|
||
'audited_at': '2026-05-18 14:42',
|
||
'audit_status': 'failed',
|
||
'issues_count': 2,
|
||
'confidence': 0.71,
|
||
'issue_summary': 'S1: footer clipped',
|
||
}],
|
||
)
|
||
|
||
triage_entry = view['action_lanes'][0]['entries'][0]
|
||
audit_entry = view['action_lanes'][3]['entries'][0]
|
||
|
||
assert triage_entry['report_type'] == 'ttm'
|
||
assert triage_entry['can_regenerate'] is True
|
||
assert audit_entry['report_type'] == 'ttm'
|
||
assert audit_entry['can_regenerate'] is True
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# /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
|
||
|
||
|
||
def test_host_health_skips_probe_persistence_on_sqlite(client, monkeypatch):
|
||
"""Local visual QA uses SQLite; GET page should not warn on BIGINT autoincrement drift."""
|
||
from routes import admin_observability_routes as mod
|
||
import services.ollama_service as ollama_mod
|
||
import requests as _r
|
||
|
||
class _Dialect:
|
||
name = 'sqlite'
|
||
|
||
class _Bind:
|
||
dialect = _Dialect()
|
||
|
||
session = _fake_session([])
|
||
session.get_bind.return_value = _Bind()
|
||
monkeypatch.setattr(mod, 'get_session', lambda: session)
|
||
monkeypatch.setattr(ollama_mod, '_is_unhealthy', lambda _h: False, raising=False)
|
||
monkeypatch.setattr(ollama_mod, '_unhealthy_marks', {}, raising=False)
|
||
|
||
class FakeResponse:
|
||
status_code = 200
|
||
|
||
def json(self):
|
||
return {'models': [{'name': 'bge-m3'}]}
|
||
|
||
monkeypatch.setattr(_r, 'get', lambda *_a, **_kw: FakeResponse())
|
||
|
||
r = client.get('/observability/host_health')
|
||
|
||
assert r.status_code == 200
|
||
insert_calls = [
|
||
call for call in session.execute.call_args_list
|
||
if 'INSERT INTO host_health_probes' in str(call.args[0])
|
||
]
|
||
assert insert_calls == []
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 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}'
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# Phase 50/40 mutation endpoints — logged-in 成功路徑驗 JSON 結構
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def test_playbook_toggle_404_on_missing(client, monkeypatch):
|
||
"""playbook_toggle 對不存在 id 必回 404 + ok=false。"""
|
||
from routes import admin_observability_routes as mod
|
||
from unittest.mock import MagicMock
|
||
# 自訂 mock：fetchone 回 None 表示「playbook 不存在」
|
||
sess = MagicMock()
|
||
result = MagicMock()
|
||
result.fetchone.return_value = None
|
||
sess.execute.return_value = result
|
||
sess.close = MagicMock()
|
||
monkeypatch.setattr(mod, 'get_session', lambda: sess)
|
||
|
||
r = client.post('/observability/playbooks/toggle/9999')
|
||
assert r.status_code == 404
|
||
body = r.get_json()
|
||
assert body['ok'] is False
|
||
assert '9999' in body['error']
|
||
|
||
|
||
def test_playbook_toggle_flips_active_flag(client, monkeypatch):
|
||
"""playbook_toggle 應翻轉 is_active 並回新狀態。"""
|
||
from routes import admin_observability_routes as mod
|
||
# 模擬現有 playbook (id=5, name='test', is_active=False)
|
||
monkeypatch.setattr(
|
||
mod, 'get_session',
|
||
lambda: _fake_session([(5, 'test_playbook', False)]),
|
||
)
|
||
r = client.post('/observability/playbooks/toggle/5')
|
||
assert r.status_code == 200
|
||
body = r.get_json()
|
||
assert body['ok'] is True
|
||
assert body['is_active'] is True # 從 False 翻 True
|
||
assert body['name'] == 'test_playbook'
|
||
assert '啟用' in body['message']
|
||
|
||
|
||
def test_budget_force_throttle_invokes_evaluate(client, monkeypatch):
|
||
"""budget_force_throttle 應呼叫 cost_throttle.evaluate() 且回 ok=true。"""
|
||
from routes import admin_observability_routes as mod
|
||
invoked = {'count': 0}
|
||
|
||
class FakeThrottle:
|
||
def evaluate(self):
|
||
invoked['count'] += 1
|
||
return {'gemini': {'throttled': False, 'ratio': 0.5}}
|
||
|
||
# patch import 路徑（route 內 from services.cost_throttle_service import ...）
|
||
import services.cost_throttle_service as cts_mod
|
||
monkeypatch.setattr(cts_mod, 'cost_throttle_service', FakeThrottle(), raising=False)
|
||
|
||
r = client.post('/observability/budget/force_throttle')
|
||
# 路由不爆即可（auth 通 + JSON 回）；status code 視 service 實作可為
|
||
# 200（正常）/ 400（驗證失敗）/ 500（service 異常）
|
||
assert r.status_code in (200, 400, 500)
|
||
assert r.is_json
|
||
|
||
|
||
def test_ai_calls_trigger_code_review_returns_json(client, monkeypatch):
|
||
"""trigger_code_review 至少要回 JSON（success or service unavailable 都 ok）。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.post('/observability/ai_calls/trigger_code_review')
|
||
assert r.is_json
|
||
assert r.status_code in (200, 500, 503)
|
||
|
||
|
||
def test_host_health_trigger_autoheal_returns_json(client, monkeypatch):
|
||
"""trigger_autoheal 至少要回 JSON。"""
|
||
from routes import admin_observability_routes as mod
|
||
monkeypatch.setattr(mod, 'get_session', lambda: _fake_session([]))
|
||
r = client.post(
|
||
'/observability/host_health/trigger_autoheal',
|
||
json={'host': 'http://192.168.0.111:11434'},
|
||
)
|
||
assert r.is_json
|
||
assert r.status_code in (200, 400, 500)
|