v1.1.46: phase2 — scenario runner MVP (YAML config, DB schema, API endpoints, UI section)
This commit is contained in:
@@ -0,0 +1,248 @@
|
||||
"""
|
||||
Scenario runner — символические коды → API → шаги → результат.
|
||||
|
||||
Формат сценария в config.yaml:
|
||||
scenarios:
|
||||
dummy_test:
|
||||
steps:
|
||||
- svc: dummy # символическое имя (из services: в config.yaml)
|
||||
op: create
|
||||
params:
|
||||
durationMs: "5000"
|
||||
- svc: dummy
|
||||
op: delete
|
||||
|
||||
Для каждого шага:
|
||||
1. service имя → svc_id (из config.yaml services)
|
||||
2. operation имя → svcOperationId (из GET /services/{id})
|
||||
3. param коды → numeric IDs (из GET /instanceOperations/default/{opId})
|
||||
4. create: новый инстанс; остальные: переиспользовать инстанс этого сервиса
|
||||
5. Выполнить через API (как в api_test)
|
||||
6. Сохранить результат в runs + scenario_runs
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from runner import load_config
|
||||
from operations.get_services import get_service_detail
|
||||
from operations.get_instances import get_instances
|
||||
from db.pool import get_conn, put_conn
|
||||
from db.save_run import save_run
|
||||
|
||||
AUTOTEST_PREFIX = "autotest-scenario-"
|
||||
|
||||
|
||||
def load_scenarios():
|
||||
"""Загрузить словарь сценариев из config.yaml."""
|
||||
cfg = load_config() or {}
|
||||
return cfg.get("scenarios", {})
|
||||
|
||||
|
||||
def _resolve_service(config, svc_name):
|
||||
"""Найти svc_id по символическому имени сервиса из config.yaml services."""
|
||||
for s in config.get("services", []):
|
||||
if s.get("name") == svc_name:
|
||||
return s
|
||||
raise ValueError(f"Service '{svc_name}' not found in config.yaml services")
|
||||
|
||||
|
||||
def _resolve_params(cfs_params, symbolic_params):
|
||||
"""Символические коды → {numeric_id: value}.
|
||||
cfs_params — список из GET /instanceOperations/default/{opId}
|
||||
symbolic_params — dict {"durationMs": "5000"}
|
||||
"""
|
||||
code_to_id = {p.get("svcOperationCfsParam", ""): p["svcOperationCfsParamId"] for p in cfs_params}
|
||||
result = {}
|
||||
for code, val in symbolic_params.items():
|
||||
pid = code_to_id.get(code)
|
||||
if pid is None:
|
||||
raise ValueError(f"Param code '{code}' not found in operation template")
|
||||
result[str(pid)] = str(val)
|
||||
return result
|
||||
|
||||
|
||||
def _save_scenario_run(scenario_run_id, status, current_step=0, duration_sec=None, error_log=None):
|
||||
"""Обновить запись scenario_runs."""
|
||||
conn = get_conn()
|
||||
if not conn:
|
||||
return
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
UPDATE scenario_runs
|
||||
SET status = %s, current_step = %s, duration_sec = %s, error_log = %s
|
||||
WHERE id = %s
|
||||
""", (status, current_step, duration_sec, error_log, scenario_run_id))
|
||||
conn.commit()
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
print(f"[SCENARIO] save_scenario_run error: {e}", flush=True)
|
||||
conn.rollback()
|
||||
finally:
|
||||
put_conn(conn)
|
||||
|
||||
|
||||
def _create_scenario_run(client_id, stand, user_email, scenario_name, total_steps, app_version):
|
||||
"""Создать запись в scenario_runs, вернуть ID."""
|
||||
conn = get_conn()
|
||||
if not conn:
|
||||
return None
|
||||
rid = None
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
INSERT INTO scenario_runs (client_id, stand, user_email, scenario_name,
|
||||
status, current_step, total_steps, app_version)
|
||||
VALUES (%s, %s, %s, %s, 'RUNNING', 0, %s, %s)
|
||||
RETURNING id
|
||||
""", (client_id, stand, user_email, scenario_name, total_steps, app_version))
|
||||
rid = cur.fetchone()[0]
|
||||
conn.commit()
|
||||
cur.close()
|
||||
except Exception as e:
|
||||
print(f"[SCENARIO] create_scenario_run error: {e}", flush=True)
|
||||
conn.rollback()
|
||||
finally:
|
||||
put_conn(conn)
|
||||
return rid
|
||||
|
||||
|
||||
def run_scenario(client, scenario_name, client_id, stand, user_email, app_version):
|
||||
"""Главный исполнитель сценария. Вызывается в фоновом потоке.
|
||||
Возвращает scenario_run_id для поллинга."""
|
||||
config = load_config() or {}
|
||||
scenarios = config.get("scenarios", {})
|
||||
scenario = scenarios.get(scenario_name)
|
||||
if not scenario:
|
||||
raise ValueError(f"Scenario '{scenario_name}' not found")
|
||||
|
||||
steps = scenario.get("steps", [])
|
||||
if not steps:
|
||||
raise ValueError(f"Scenario '{scenario_name}' has no steps")
|
||||
|
||||
total = len(steps)
|
||||
|
||||
# Создать запись в scenario_runs
|
||||
scenario_run_id = _create_scenario_run(client_id, stand, user_email, scenario_name, total, app_version)
|
||||
if scenario_run_id is None:
|
||||
raise RuntimeError("Failed to create scenario_run record")
|
||||
|
||||
t0 = time.time()
|
||||
instance_map = {} # svc_name → instanceUid (переиспользование внутри сценария)
|
||||
|
||||
for i, step in enumerate(steps):
|
||||
step_num = i + 1
|
||||
svc_name = step.get("svc", "")
|
||||
op_name = step.get("op", "")
|
||||
symbolic_params = step.get("params", {})
|
||||
|
||||
try:
|
||||
# 1. Сервис
|
||||
svc_info = _resolve_service(config, svc_name)
|
||||
svc_id = svc_info["service_id"]
|
||||
|
||||
# 2. Операция
|
||||
detail = get_service_detail(client, svc_id)
|
||||
ops = detail.get("operations", [])
|
||||
op = next((o for o in ops if o.get("operation") == op_name), None)
|
||||
if not op:
|
||||
raise ValueError(f"Operation '{op_name}' not found for service {svc_name}")
|
||||
svc_op_id = op["svcOperationId"]
|
||||
|
||||
# 3. Параметры
|
||||
tmpl_data = client.get(f"/instanceOperations/default/{svc_op_id}")
|
||||
cfs_params = tmpl_data.get("svcOperation", {}).get("cfsParams", [])
|
||||
resolved_params = _resolve_params(cfs_params, symbolic_params)
|
||||
|
||||
# 4. Инстанс
|
||||
if op_name == "create":
|
||||
display_name = f"{AUTOTEST_PREFIX}{scenario_name}-{uuid.uuid4().hex[:6]}"
|
||||
descr = f"scenario {scenario_name} step {step_num}"
|
||||
payload = {"serviceId": svc_id, "displayName": display_name, "descr": descr}
|
||||
resp = client.post("/instances", payload)
|
||||
instance_uid = resp.get("instanceUid")
|
||||
if not instance_uid:
|
||||
raise RuntimeError("CREATE: no instanceUid in response")
|
||||
instance_map[svc_name] = instance_uid
|
||||
else:
|
||||
instance_uid = instance_map.get(svc_name)
|
||||
if not instance_uid:
|
||||
raise RuntimeError(f"No instance for service '{svc_name}' — need CREATE first")
|
||||
|
||||
# 5. Запуск операции
|
||||
op_payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id, "operation": op_name}
|
||||
op_resp = client.post("/instanceOperations", op_payload)
|
||||
op_uid = op_resp.get("instanceOperationUid") or op_resp.get("instanceOperation", {}).get("instanceOperationUid")
|
||||
if not op_uid:
|
||||
op_uid = next((v for k, v in op_resp.items() if "Uid" in k or "uid" in k), None)
|
||||
if not op_uid and "_location" in op_resp:
|
||||
loc = op_resp["_location"]
|
||||
op_uid = loc.rsplit("/", 1)[-1] if "/" in loc else loc
|
||||
if not op_uid:
|
||||
raise RuntimeError("No opUid in response")
|
||||
|
||||
# Отправить параметры и запустить
|
||||
from routes.api_test import _send_params_terraform
|
||||
_send_params_terraform(client, op_uid, resolved_params)
|
||||
client.post(f"/instanceOperations/{op_uid}/run")
|
||||
|
||||
# 6. Поллинг завершения
|
||||
step_t0 = time.time()
|
||||
deadline = step_t0 + 1800
|
||||
step_status = "TIMEOUT"
|
||||
step_error = ""
|
||||
step_duration = 0
|
||||
step_stages = []
|
||||
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
data = client.get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,duration,stages,svc")
|
||||
except Exception:
|
||||
time.sleep(5)
|
||||
continue
|
||||
op_data = data.get("instanceOperation", {})
|
||||
dt_finish = op_data.get("dtFinish")
|
||||
if dt_finish and str(dt_finish).strip():
|
||||
is_ok = op_data.get("isSuccessful")
|
||||
err = op_data.get("errorLog") or ""
|
||||
step_status = "OK" if is_ok else "FAIL"
|
||||
step_error = str(err) if err else ""
|
||||
step_duration = round(time.time() - step_t0, 1)
|
||||
step_stages = op_data.get("stages", [])
|
||||
break
|
||||
time.sleep(5)
|
||||
|
||||
# 7. Сохранить шаг в runs
|
||||
try:
|
||||
save_run(client_id, stand, user_email,
|
||||
svc_id, op_data.get("svc", svc_name), op_name, svc_op_id,
|
||||
op_uid, instance_uid, display_name if op_name == "create" else instance_uid,
|
||||
step_status, step_duration, step_error,
|
||||
resolved_params, step_stages, app_version)
|
||||
except Exception as e:
|
||||
print(f"[SCENARIO] save_run failed for step {step_num}: {e}", flush=True)
|
||||
|
||||
# Обновить scenario_runs
|
||||
_save_scenario_run(scenario_run_id, "RUNNING" if step_num < total else step_status,
|
||||
step_num, round(time.time() - t0, 1),
|
||||
step_error if step_status != "OK" else None)
|
||||
|
||||
if step_status != "OK":
|
||||
# Прервать сценарий при ошибке
|
||||
_save_scenario_run(scenario_run_id, step_status, step_num,
|
||||
round(time.time() - t0, 1), step_error)
|
||||
return
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
err = f"Step {step_num}: {e}"
|
||||
print(f"[SCENARIO] {err}", flush=True)
|
||||
traceback.print_exc()
|
||||
_save_scenario_run(scenario_run_id, "FAIL", step_num,
|
||||
round(time.time() - t0, 1), err)
|
||||
return
|
||||
|
||||
# Все шаги OK
|
||||
_save_scenario_run(scenario_run_id, "OK", total, round(time.time() - t0, 1))
|
||||
Reference in New Issue
Block a user