309 lines
15 KiB
Python
309 lines
15 KiB
Python
"""
|
||
Scenario runner — запуск сценария: шаги из БД → API Nubes → результат.
|
||
|
||
Сценарий — это последовательность операций над сервисами.
|
||
Пример: create Болванку → modify параметры → delete Болванку.
|
||
|
||
ШАГИ (новый формат с именованными ссылками):
|
||
{"service_id": 1, "operation": "create", "params": {...}, "output": "d1"}
|
||
— создать инстанс сервиса 1, дать ему имя "d1" для ссылок из следующих шагов
|
||
{"service_id": 1, "operation": "modify", "instance_ref": "d1", "params": {...}}
|
||
— модифицировать инстанс, созданный в шаге с output="d1"
|
||
{"service_id": 1, "operation": "delete", "instance_uid": "UUID", "params": {}}
|
||
— удалить конкретный инстанс по UUID
|
||
|
||
Старый формат (без output/instance_ref/instance_uid):
|
||
{"service_id": 1, "operation": "create", "params": {...}}
|
||
{"service_id": 1, "operation": "modify", "params": {...}}
|
||
— fallback на service_id (instance_map[service_id] → последний созданный)
|
||
|
||
РЕЗОЛВИНГ instance_uid (приоритет):
|
||
1. Явный instance_uid в шаге
|
||
2. instance_ref → поиск в bindings (output предыдущего create-шага)
|
||
3. instance_map[service_id] → последний инстанс этого сервиса (fallback)
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import uuid
|
||
|
||
from operations.get_services import get_service_detail
|
||
from operations.executor import execute_operation
|
||
from operations.poll import poll_until_done
|
||
from db.pool import get_conn, put_conn
|
||
from db.save_run import save_run
|
||
|
||
# Все инстансы созданные сценарием имеют такой префикс в displayName
|
||
AUTOTEST_PREFIX = "autotest-scenario-"
|
||
|
||
|
||
def _resolve_params(cfs_params, symbolic_params):
|
||
"""Конвертировать символические имена параметров → {numeric_id: value}.
|
||
|
||
В БД сценарии хранят параметры с СИМВОЛИЧЕСКИМИ именами (напр. "durationMs": "0").
|
||
Но executor.execute_operation требует ЧИСЛОВЫЕ ID (напр. "198": "0").
|
||
|
||
Эта функция делает обратный маппинг: code → svcOperationCfsParamId.
|
||
|
||
Args:
|
||
cfs_params: list — cfsParams из шаблона операции (содержит svcOperationCfsParam)
|
||
symbolic_params: dict — {code: value} из БД
|
||
|
||
Returns:
|
||
dict — {numeric_id: value}
|
||
|
||
Raises:
|
||
ValueError если код параметра не найден в шаблоне (опечатка в сценарии)."""
|
||
# Строим словарь: code → numeric_id
|
||
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 в БД.
|
||
|
||
Вызывается после КАЖДОГО шага чтобы UI видел прогресс.
|
||
При ошибке — пишем в stderr и продолжаем (не роняем сценарий из-за БД)."""
|
||
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 _update_bindings(scenario_run_id, bindings):
|
||
"""Сохранить output→instance_uid в scenario_runs.instance_bindings (JSONB).
|
||
|
||
bindings — это словарь: {"d1": "uuid-1", "d2": "uuid-2"}
|
||
Нужен для отладки: видеть какие инстансы были созданы на каждом шаге."""
|
||
conn = get_conn()
|
||
if not conn:
|
||
return
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("""
|
||
UPDATE scenario_runs SET instance_bindings = %s WHERE id = %s
|
||
""", (json.dumps(bindings), scenario_run_id))
|
||
conn.commit()
|
||
cur.close()
|
||
except Exception as e:
|
||
print(f"[SCENARIO] update_bindings error: {e}", flush=True)
|
||
conn.rollback()
|
||
finally:
|
||
put_conn(conn)
|
||
|
||
|
||
def _resolve_instance_uid(step, bindings, instance_map):
|
||
"""Резолвинг instance_uid по приоритету: явный uid > instance_ref > service_id.
|
||
|
||
Args:
|
||
step: dict — шаг сценария
|
||
bindings: dict — {output_name: instance_uid} (накопленные за сценарий)
|
||
instance_map: dict — {service_id: instance_uid} (fallback, последний инстанс сервиса)
|
||
|
||
Returns:
|
||
str — instance_uid
|
||
|
||
Raises:
|
||
ValueError если instance_ref не найден в bindings (шаги не в том порядке)."""
|
||
# Приоритет 1: явный instance_uid в шаге
|
||
uid = (step.get("instance_uid") or "").strip()
|
||
if uid:
|
||
return uid
|
||
|
||
# Приоритет 2: instance_ref → поиск в bindings
|
||
ref = (step.get("instance_ref") or "").strip()
|
||
if ref:
|
||
if ref in bindings:
|
||
return bindings[ref]
|
||
raise ValueError(f"instance_ref '{ref}' not found in bindings (step order issue?)")
|
||
|
||
# Приоритет 3: fallback — последний инстанс этого сервиса
|
||
svc_id = step.get("service_id")
|
||
return instance_map.get(svc_id)
|
||
|
||
|
||
def run_scenario(client, steps, client_id, stand, user_email, app_version, scenario_run_id, scenario_name):
|
||
"""Главный исполнитель сценария — последовательно выполняет шаги.
|
||
|
||
Алгоритм для КАЖДОГО шага:
|
||
1. Получить детали сервиса → найти svcOperationId для операции
|
||
2. Получить шаблон параметров → резолвить символические имена в числовые ID
|
||
3. Резолвить instance_uid (create → None, не-create → из bindings/instance_map)
|
||
4. Вызвать execute_operation (единый executor)
|
||
5. Сохранить RUNNING в БД
|
||
6. Поллить через poll_until_done
|
||
7. Сохранить финальный результат (OK/FAIL/TIMEOUT) + instance_meta
|
||
8. Обновить scenario_runs (прогресс для UI)
|
||
|
||
Если любой шаг FAIL — сценарий останавливается.
|
||
Если все шаги OK — сценарий завершается успешно.
|
||
|
||
Args:
|
||
client: HttpClient
|
||
steps: list[dict] — шаги из БД (scenario_definitions.steps)
|
||
client_id: str — ClientID
|
||
stand: str — "dev"/"test"
|
||
user_email: str — email пользователя
|
||
app_version: str — версия приложения
|
||
scenario_run_id: int — ID записи в scenario_runs
|
||
scenario_name: str — имя сценария
|
||
"""
|
||
total = len(steps)
|
||
t0 = time.time()
|
||
|
||
# Две карты для резолвинга instance_uid:
|
||
bindings = {} # output_name → instance_uid (новый формат, именованные ссылки)
|
||
instance_map = {} # service_id → instance_uid (fallback, старый формат)
|
||
|
||
for i, step in enumerate(steps):
|
||
step_num = i + 1 # шаги нумеруются с 1 (для пользователя)
|
||
svc_id = step.get("service_id")
|
||
op_name = (step.get("operation") or "").strip()
|
||
symbolic_params = step.get("params", {})
|
||
|
||
try:
|
||
# Валидация базовых полей шага
|
||
if not isinstance(svc_id, int) or svc_id <= 0:
|
||
raise ValueError(f"Invalid service_id: {svc_id!r}")
|
||
if not op_name:
|
||
raise ValueError("Empty operation name")
|
||
|
||
# ═══ Шаг 1: получить svcOperationId для операции ═══
|
||
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_id}")
|
||
svc_op_id = op["svcOperationId"]
|
||
|
||
# ═══ Шаг 2: резолвить параметры (символ. имена → числовые ID) ═══
|
||
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)
|
||
|
||
# ═══ Шаг 3: резолвинг instance_uid ═══
|
||
is_create = (op_name == "create")
|
||
if is_create:
|
||
# Для create: instance_uid=None (создаём новый)
|
||
# displayName генерируем уникальный: autotest-scenario-{name}-{random6}
|
||
instance_uid = None
|
||
display_name = f"{AUTOTEST_PREFIX}{scenario_name}-{uuid.uuid4().hex[:6]}"
|
||
descr = f"scenario {scenario_name} step {step_num}"
|
||
else:
|
||
# Для не-create: резолвим instance_uid
|
||
instance_uid = _resolve_instance_uid(step, bindings, instance_map)
|
||
if not instance_uid:
|
||
raise ValueError(
|
||
f"No instance for step {step_num} — need CREATE before non-create operations")
|
||
display_name = None
|
||
descr = None
|
||
|
||
# ═══ Шаг 4: запуск через ЕДИНЫЙ executor ═══
|
||
# Это гарантирует идентичное поведение с ручным режимом
|
||
result = execute_operation(
|
||
client, svc_id, op_name, instance_uid, resolved_params,
|
||
svc_op_id=svc_op_id, display_name=display_name, descr=descr,
|
||
client_id=client_id, stand=stand
|
||
)
|
||
if not result["ok"]:
|
||
raise RuntimeError(
|
||
f"{op_name}: {result['error']} (step: {result['failed_step']})")
|
||
|
||
instance_uid = result["instance_uid"]
|
||
op_uid = result["op_uid"]
|
||
step_display = result["display_name"]
|
||
|
||
# Обновить карты для резолвинга следующих шагов
|
||
instance_map[svc_id] = instance_uid # fallback: последний инстанс сервиса
|
||
output_name = (step.get("output") or "").strip()
|
||
if is_create and output_name:
|
||
bindings[output_name] = instance_uid # именованная ссылка
|
||
_update_bindings(scenario_run_id, bindings)
|
||
|
||
# ═══ Шаг 5: сохранить RUNNING в БД (до поллинга) ═══
|
||
try:
|
||
save_run(client_id, stand, user_email,
|
||
svc_id, "", op_name, svc_op_id,
|
||
op_uid, instance_uid, step_display,
|
||
"RUNNING", 0, "",
|
||
resolved_params, [], app_version,
|
||
scenario_run_id, step_num)
|
||
except Exception as e:
|
||
print(f"[SCENARIO] save_run(RUNNING) failed for step {step_num}: {e}", flush=True)
|
||
|
||
# ═══ Шаг 6: поллинг до завершения ═══
|
||
poll_result = poll_until_done(client, op_uid)
|
||
step_status = poll_result["status"] # OK / FAIL / TIMEOUT
|
||
step_error = poll_result["error_log"]
|
||
step_duration = poll_result["duration"]
|
||
step_stages = poll_result["stages"]
|
||
svc_final = poll_result["svc"]
|
||
|
||
# ═══ Шаг 7: сохранить финальный результат + instance_meta ═══
|
||
try:
|
||
# Получить полную информацию об инстансе для анализа
|
||
instance_meta = None
|
||
try:
|
||
inst_data = client.get(f"/instances/{instance_uid}")
|
||
inst = inst_data.get("instance", {}) if isinstance(inst_data, dict) else {}
|
||
if inst:
|
||
# Сохраняем ВСЁ кроме очевидных/избыточных полей
|
||
instance_meta = {k: v for k, v in inst.items()
|
||
if k not in ('instanceUid', 'displayName', 'svc', 'serviceId', 'explainedStatus')}
|
||
except Exception:
|
||
pass # не смогли получить meta — не критично
|
||
save_run(client_id, stand, user_email,
|
||
svc_id, svc_final, op_name, svc_op_id,
|
||
op_uid, instance_uid, step_display,
|
||
step_status, step_duration, step_error,
|
||
resolved_params, step_stages, app_version,
|
||
scenario_run_id, step_num,
|
||
instance_meta=instance_meta)
|
||
except Exception as e:
|
||
print(f"[SCENARIO] save_run failed for step {step_num}: {e}", flush=True)
|
||
|
||
# ═══ Шаг 8: обновить scenario_runs (прогресс для UI) ═══
|
||
# Пока не последний шаг — статус RUNNING
|
||
_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)
|
||
|
||
# Если шаг FAIL — останавливаем сценарий
|
||
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:
|
||
# Любая ошибка → FAIL, сценарий остановлен
|
||
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
|
||
|
||
# Все шаги пройдены успешно
|
||
_save_scenario_run(scenario_run_id, "OK", total, round(time.time() - t0, 1))
|