125 lines
3.8 KiB
Python
125 lines
3.8 KiB
Python
import os
|
|
import sys
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
PROJECT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if PROJECT_DIR not in sys.path:
|
|
sys.path.insert(0, PROJECT_DIR)
|
|
|
|
from app.db.system_logs import get_system_error, list_system_errors, record_system_error
|
|
from app.web import web_server
|
|
|
|
|
|
def test_record_and_query_system_error():
|
|
log_id = record_system_error(
|
|
source="web",
|
|
error_type="RuntimeError",
|
|
message="boom",
|
|
stack_trace="Traceback\nRuntimeError: boom",
|
|
request_method="GET",
|
|
request_path="/api/test",
|
|
status_code=500,
|
|
context={"case": "unit"},
|
|
)
|
|
|
|
assert log_id > 0
|
|
detail = get_system_error(log_id)
|
|
assert detail["message"] == "boom"
|
|
assert detail["context"]["case"] == "unit"
|
|
|
|
rows = list_system_errors(search="boom", limit=10)
|
|
assert rows["total"] >= 1
|
|
assert rows["items"][0]["id"] == log_id
|
|
|
|
|
|
def test_record_system_error_sends_feishu_alert(monkeypatch):
|
|
pushed = []
|
|
monkeypatch.setattr("app.db.system_logs.push_system_error_alert", lambda item: pushed.append(item) or (True, {"StatusCode": 0}))
|
|
|
|
log_id = record_system_error(
|
|
source="web",
|
|
error_type="RuntimeError",
|
|
message="alert me",
|
|
stack_trace="Traceback\nRuntimeError: alert me",
|
|
request_path="/api/alert",
|
|
)
|
|
|
|
assert log_id > 0
|
|
assert pushed
|
|
assert pushed[0]["id"] == log_id
|
|
assert pushed[0]["message"] == "alert me"
|
|
|
|
|
|
def test_warning_system_error_does_not_send_feishu_alert(monkeypatch):
|
|
pushed = []
|
|
monkeypatch.setattr("app.db.system_logs.push_system_error_alert", lambda item: pushed.append(item) or (True, {"StatusCode": 0}))
|
|
|
|
log_id = record_system_error(
|
|
source="price_streamer",
|
|
level="warning",
|
|
error_type="ConnectionClosedError",
|
|
message="transient websocket disconnect",
|
|
status_code=0,
|
|
)
|
|
|
|
assert log_id > 0
|
|
assert pushed == []
|
|
assert get_system_error(log_id)["level"] == "warning"
|
|
|
|
|
|
def test_admin_system_error_api_uses_local_admin():
|
|
log_id = record_system_error(
|
|
source="scheduler",
|
|
error_type="job_exit_1",
|
|
message="job failed",
|
|
stack_trace="exit=1",
|
|
status_code=1,
|
|
)
|
|
client = TestClient(web_server.app)
|
|
|
|
listing = client.get("/api/admin/system-errors?search=job%20failed")
|
|
assert listing.status_code == 200
|
|
assert listing.json()["items"][0]["id"] == log_id
|
|
|
|
detail = client.get(f"/api/admin/system-errors/{log_id}")
|
|
assert detail.status_code == 200
|
|
assert detail.json()["stack_trace"] == "exit=1"
|
|
|
|
stats = client.get("/api/admin/system-errors/stats")
|
|
assert stats.status_code == 200
|
|
assert stats.json()["total"] >= 1
|
|
|
|
|
|
def test_system_logs_page_is_separate_from_admin_dashboard():
|
|
client = TestClient(web_server.app)
|
|
|
|
logs_page = client.get("/system-logs")
|
|
assert logs_page.status_code == 200
|
|
assert "系统日志" in logs_page.text
|
|
assert 'href="/system-logs"' in logs_page.text
|
|
assert 'active admin-link' in logs_page.text
|
|
assert "logTable" in logs_page.text
|
|
|
|
admin_page = client.get("/admin.html")
|
|
assert admin_page.status_code == 200
|
|
assert 'data-admin-tab="logs"' not in admin_page.text
|
|
assert 'id="logsPanel"' not in admin_page.text
|
|
assert "系统日志" not in admin_page.text
|
|
|
|
|
|
def test_sidebar_navigation_is_owned_by_base_template():
|
|
static_dir = os.path.join(PROJECT_DIR, "static")
|
|
page_files = [
|
|
name
|
|
for name in os.listdir(static_dir)
|
|
if name.endswith(".html") and name not in {"base.html", "auth.html", "index.html"}
|
|
]
|
|
offenders = []
|
|
for name in page_files:
|
|
with open(os.path.join(static_dir, name), "r", encoding="utf-8") as f:
|
|
if "{% block nav_links %}" in f.read():
|
|
offenders.append(name)
|
|
|
|
assert offenders == []
|