""" 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 db.pool import get_conn, put_conn from db.save_run import save_run AUTOTEST_PREFIX = "autotest-scenario-" def _find_uid(resp): """Extract UUID from nested response (e.g. {instanceOperation: {instanceOperationUid: ...}}).""" if isinstance(resp, dict): for v in resp.values(): if isinstance(v, dict): uid = v.get("instanceOperationUid") or v.get("instanceUid") if uid: return uid return None def _uid_from_location(loc): """Extract UUID from Location header.""" if not loc: return None parts = str(loc).rstrip("/").split("/") return parts[-1] if parts[-1] else None 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") or _find_uid(resp) or _uid_from_location(resp.get("_location", "")) if not instance_uid: raise RuntimeError(f"CREATE: no instanceUid in response: {json.dumps(resp)[:200]}") 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 = _find_uid(op_resp) if not op_uid: op_uid = _uid_from_location(op_resp.get("_location", "")) if not op_uid: raise RuntimeError(f"No opUid in response: {json.dumps(op_resp)[:200]}") # Отправить параметры и запустить 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))