79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
import os
|
|
import sys
|
|
|
|
import pytest
|
|
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 import altcoin_db
|
|
from app.web import web_server
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "altcoin_monitor.db"
|
|
monkeypatch.setattr(altcoin_db, "DB_PATH", str(db_path))
|
|
monkeypatch.setattr(web_server, "init_db", altcoin_db.init_db)
|
|
monkeypatch.setattr(web_server, "get_cron_run_logs", altcoin_db.get_cron_run_logs)
|
|
monkeypatch.setattr(web_server, "get_cron_run_summary", altcoin_db.get_cron_run_summary)
|
|
altcoin_db.init_db()
|
|
return db_path
|
|
|
|
|
|
def test_cron_run_log_roundtrip_and_summary(temp_db):
|
|
altcoin_db.log_cron_run(
|
|
job_name="粗筛",
|
|
script_name="altcoin_screener.py",
|
|
run_status="success",
|
|
result_status="screened",
|
|
duration_ms=1234,
|
|
summary={"total_candidates": 88, "total_qualified": 6, "alert_count": 2},
|
|
)
|
|
altcoin_db.log_cron_run(
|
|
job_name="跟踪",
|
|
script_name="price_tracker.py",
|
|
run_status="error",
|
|
result_status="exception",
|
|
duration_ms=222,
|
|
summary={"tracked_count": 0},
|
|
error_message="boom",
|
|
)
|
|
|
|
logs = altcoin_db.get_cron_run_logs(limit=10)
|
|
assert len(logs) == 2
|
|
assert logs[0]["job_name"] == "跟踪"
|
|
assert logs[0]["summary_json"]["tracked_count"] == 0
|
|
assert logs[1]["summary_json"]["total_qualified"] == 6
|
|
|
|
summary = altcoin_db.get_cron_run_summary(hours=24)
|
|
assert summary["overall"]["total_runs"] == 2
|
|
assert summary["overall"]["success_runs"] == 1
|
|
assert summary["overall"]["error_runs"] == 1
|
|
stats_by_job = {item["job_name"]: item for item in summary["job_stats"]}
|
|
assert stats_by_job["粗筛"]["last_result_status"] == "screened"
|
|
assert summary["recent_logs"][1]["summary_json"]["total_candidates"] == 88
|
|
assert stats_by_job["跟踪"]["last_error_message"] == "boom"
|
|
|
|
|
|
def test_cron_log_api_returns_summary_and_logs(temp_db):
|
|
altcoin_db.log_cron_run(
|
|
job_name="确认",
|
|
script_name="altcoin_confirm.py",
|
|
run_status="success",
|
|
result_status="confirmed",
|
|
duration_ms=456,
|
|
summary={"confirmed_count": 3, "unconfirmed_count": 1},
|
|
)
|
|
|
|
client = TestClient(web_server.app)
|
|
resp = client.get("/api/cron/summary?hours=24")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["overall"]["total_runs"] == 1
|
|
assert data["job_stats"][0]["last_result_status"] == "confirmed"
|
|
assert data["recent_logs"][0]["script_name"] == "altcoin_confirm.py"
|