""" 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))