""" LEGACY-раннер — автоматический прогон тестов по конфигурации из config.yaml. Этот модуль больше НЕ используется в основном потоке приложения. Оставлен для обратной совместимости: эндпоинты /api/run, /api/status, /api/config. Сейчас основной режим — ручной (через UI) и сценарный (через scenario.py). """ import threading import time import yaml import os from api.http_client import HttpClient # Путь к YAML-конфигу тестов CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml") # _current_run — глобальное состояние последнего запуска. # Защищено _lock для потокобезопасности. _current_run = None _lock = threading.Lock() def load_config(): """Загрузить YAML-конфиг тестов. Используется в /api/config (GET).""" with open(CONFIG_PATH) as f: return yaml.safe_load(f) def save_config(data): """Сохранить YAML-конфиг. Используется в /api/config (POST).""" with open(CONFIG_PATH, "w") as f: yaml.dump(data, f, allow_unicode=True, default_flow_style=False) def get_status(): """Текущий статус последнего запуска тестов → /api/status.""" return _current_run def _wait_for_ready(client, instance_uid, timeout=300): """Ждать пока инстанс перейдёт в стабильное состояние (не creating/modifying/deleting). Поллинг GET /instances с фильтрацией по instanceUid. Таймаут 5 минут — после этого возвращаем None. Args: client: HttpClient instance_uid: str — UUID инстанса timeout: int — макс. секунд ожидания Returns: dict — инстанс, или None если таймаут.""" deadline = time.time() + timeout while time.time() < deadline: data = client.get("/instances", params={"pageSize": 500}) for inst in data.get("results", []): if inst.get("instanceUid") == instance_uid: status = inst.get("explainedStatus", "") # Проверяем что операция не в процессе in_progress = inst.get("operationIsInProgress", False) pending = inst.get("operationIsPending", False) if not in_progress and not pending and status not in ("creating", "modifying", "deleting"): return inst time.sleep(10) return None def _find_svc_operation_id(client, service_id, operation_name): """Найти svcOperationId для операции в сервисе. GET /services/{id} → svc.operations[] → ищем по operation.name. Args: client: HttpClient service_id: int operation_name: str — "create"/"modify"/"delete"/... Returns: int или None — svcOperationId.""" data = client.get(f"/services/{service_id}") for op in data.get("svc", {}).get("operations", []): if op.get("operation") == operation_name: return op["svcOperationId"] return None def run_tests(endpoint, token): """Главная функция LEGACY-раннера — последовательный прогон тестов. Алгоритм: 1. Загрузить config.yaml 2. Для каждого сервиса (если enabled): a. Для каждой операции (если enabled): - create → POST /instances → ждать ready - другие → POST /instanceOperations → ждать ready b. Сохранить результат (OK/FAIL/SKIP) Вызывается в отдельном потоке из /api/run. Args: endpoint: str — URL API token: str — JWT-токен""" global _current_run client = HttpClient(endpoint, token) config = load_config() # Получить UUID организации (если не задан явно) org_uid = config.get("test_org_uid", "auto") if org_uid == "auto": from operations.get_instances import get_organization org = get_organization(client) org_uid = org["instanceUid"] if org else "" run_id = time.strftime("%Y%m%d-%H%M%S") with _lock: _current_run = { "run_id": run_id, "status": "running", "started_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "finished_at": None, "results": [], } for svc_cfg in config.get("services", []): if not svc_cfg.get("enabled"): continue service_id = svc_cfg["service_id"] display_name = svc_cfg.get("display_name", f"autotest-{svc_cfg['name']}") instance_uid = None # сбрасываем для каждого сервиса for op_cfg in svc_cfg.get("operations", []): if not op_cfg.get("enabled"): with _lock: _current_run["results"].append({ "service_id": service_id, "service_name": svc_cfg["name"], "operation": op_cfg["name"], "status": "SKIP", "instance_uid": None, "error": None, "duration_sec": 0, }) continue t0 = time.time() result = { "service_id": service_id, "service_name": svc_cfg["name"], "operation": op_cfg["name"], "status": "RUNNING", "instance_uid": None, "error": None, "duration_sec": 0, } with _lock: _current_run["results"].append(result) try: svc_op_id = _find_svc_operation_id(client, service_id, op_cfg["name"]) if not svc_op_id: raise Exception(f"Operation {op_cfg['name']} not found") if op_cfg["name"] == "create": payload = {"serviceId": service_id, "displayName": display_name} params = op_cfg.get("params", {}) if params: cfs_params = [{"svcOperationCfsParamId": int(k), "paramValue": str(v)} for k, v in params.items()] payload["cfsParams"] = cfs_params create_resp = client.post("/instances", payload) instance_uid = create_resp.get("instanceUid") or _extract_uid_from_response(create_resp) if not instance_uid: raise Exception("No instanceUid in create response") inst = _wait_for_ready(client, instance_uid) if not inst: raise Exception("Timeout waiting for create") else: if not instance_uid: raise Exception(f"No instance for {op_cfg['name']}") payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id} params = op_cfg.get("params", {}) if params: payload["cfsParams"] = [{"svcOperationCfsParamId": int(k), "paramValue": str(v)} for k, v in params.items()] client.post("/instanceOperations", payload) inst = _wait_for_ready(client, instance_uid) if not inst: raise Exception(f"Timeout waiting for {op_cfg['name']}") result["status"] = "OK" result["instance_uid"] = instance_uid except Exception as e: result["status"] = "FAIL" result["error"] = str(e) if op_cfg["name"] == "create": break # create не удался → остальные операции сервиса бессмысленны finally: result["duration_sec"] = round(time.time() - t0, 1) time.sleep(1) # пауза между сервисами with _lock: _current_run["status"] = "done" _current_run["finished_at"] = time.strftime("%Y-%m-%dT%H:%M:%S") def _extract_uid_from_response(resp): """Извлечь UUID из ответа API (legacy-метод, заменён на api.utils.find_uid). Простой перебор всех значений dict — если строка из 36 символов с 4 дефисами, считаем UUID'ом.""" if isinstance(resp, dict): for v in resp.values(): if isinstance(v, str) and len(v) == 36 and v.count("-") == 4: return v return None