131 lines
4.9 KiB
Python
131 lines
4.9 KiB
Python
"""
|
|
API для запуска сценариев автотестирования.
|
|
|
|
POST /api/scenario/run — запустить сценарий (фон)
|
|
GET /api/scenario/status/<id> — поллинг статуса сценария
|
|
GET /api/scenarios — список доступных сценариев
|
|
GET /api/scenario/history — история запусков сценариев
|
|
"""
|
|
|
|
from flask import Blueprint, current_app, jsonify, request
|
|
import threading
|
|
|
|
from api.auth import get_client, get_client_id, get_stand, get_token_info
|
|
from operations.scenario import load_scenarios, run_scenario, _create_scenario_run
|
|
from db.pool import get_conn, put_conn
|
|
|
|
bp = Blueprint("api_scenario", __name__)
|
|
|
|
|
|
@bp.route("/api/scenarios")
|
|
def api_scenarios():
|
|
"""Список доступных сценариев из config.yaml."""
|
|
try:
|
|
scenarios = load_scenarios()
|
|
result = [{"name": name, "steps": len(s.get("steps", []))} for name, s in scenarios.items()]
|
|
return jsonify(sorted(result, key=lambda x: x["name"]))
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@bp.route("/api/scenario/run", methods=["POST"])
|
|
def api_scenario_run():
|
|
"""Запустить сценарий. Создаёт запись ДО потока, возвращает run_id (202)."""
|
|
try:
|
|
data = request.get_json()
|
|
scenario_name = (data.get("scenario") or "").strip()
|
|
if not scenario_name:
|
|
return jsonify({"error": "scenario name is required"}), 400
|
|
|
|
client = get_client()
|
|
cid = get_client_id()
|
|
stand = get_stand()
|
|
user_email = get_token_info().get("email", "")
|
|
app_version = current_app.config.get("VERSION", "")
|
|
|
|
# Валидация: сценарий существует?
|
|
scenarios = load_scenarios()
|
|
if scenario_name not in scenarios:
|
|
return jsonify({"error": f"Scenario '{scenario_name}' not found"}), 404
|
|
|
|
total = len(scenarios[scenario_name].get("steps", []))
|
|
if total == 0:
|
|
return jsonify({"error": f"Scenario '{scenario_name}' has no steps"}), 400
|
|
|
|
# Создать запись ДО потока — вернуть run_id клиенту
|
|
run_id = _create_scenario_run(cid, stand, user_email, scenario_name, total, app_version)
|
|
if run_id is None:
|
|
return jsonify({"error": "Failed to create scenario_run"}), 500
|
|
|
|
# Запуск в фоне
|
|
threading.Thread(
|
|
target=run_scenario,
|
|
args=(client, scenario_name, cid, stand, user_email, app_version, run_id),
|
|
daemon=True,
|
|
).start()
|
|
|
|
return jsonify({"status": "RUNNING", "scenario": scenario_name, "run_id": run_id}), 202
|
|
except Exception as e:
|
|
import traceback
|
|
print(f"[SCENARIO API] run error: {traceback.format_exc()}", flush=True)
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@bp.route("/api/scenario/run/<int:run_id>")
|
|
def api_scenario_run_status(run_id):
|
|
"""Статус конкретного запуска сценария."""
|
|
conn = get_conn()
|
|
if not conn:
|
|
return jsonify({"error": "no db"}), 500
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT id, created_at, scenario_name, status, current_step, total_steps,
|
|
duration_sec, error_log
|
|
FROM scenario_runs
|
|
WHERE id = %s AND client_id = %s AND stand = %s
|
|
""", (run_id, get_client_id(), get_stand()))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return jsonify({"error": "not found"}), 404
|
|
cols = [d[0] for d in cur.description]
|
|
d = dict(zip(cols, row))
|
|
if d["created_at"]:
|
|
d["created_at"] = d["created_at"].isoformat()
|
|
return jsonify(d)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
finally:
|
|
put_conn(conn)
|
|
|
|
|
|
@bp.route("/api/scenario/status")
|
|
def api_scenario_status():
|
|
"""Последний запущенный сценарий (или несколько последних)."""
|
|
conn = get_conn()
|
|
if not conn:
|
|
return jsonify([])
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT id, created_at, scenario_name, status, current_step, total_steps,
|
|
duration_sec, error_log
|
|
FROM scenario_runs
|
|
WHERE client_id = %s AND stand = %s
|
|
ORDER BY created_at DESC
|
|
LIMIT 10
|
|
""", (get_client_id(), get_stand()))
|
|
rows = cur.fetchall()
|
|
cols = [d[0] for d in cur.description]
|
|
result = []
|
|
for r in rows:
|
|
d = dict(zip(cols, r))
|
|
if d["created_at"]:
|
|
d["created_at"] = d["created_at"].isoformat()
|
|
result.append(d)
|
|
return jsonify(result)
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
finally:
|
|
put_conn(conn)
|