Files
app-autotest/site/routes/api_test.py
T

558 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Основной модуль: запуск операций, параметры, поллинг статуса, логирование.
Эндпоинты:
GET /api/services — список сервисов
GET /api/instances/list — все инстансы
GET /api/operations/<svc_id> — операции + tracked-инстансы сервиса
GET /api/params/<op_id>[?instanceUid=xxx] — параметры операции (шаблон или текущие)
POST /api/test — запустить операцию (CREATE или non-CREATE)
GET /api/test/status/<op_uid> — поллинг статуса операции
GET /api/log — последние строки лога
Ключевые функции:
_client() — HttpClient с токеном из cookie/env
_client_id() — ClientID из JWT
_stand() — dev/test по токену
_with_prefix(name) — гарантирует префикс autotest-
_unique_display_name(client, name)— проверка на дубликат + суффикс
_find_uid(resp) — извлечение UUID из ответа API
_uid_from_location(loc) — извлечение UUID из Location-заголовка
_get_instance_display_name(...) — GET /instances/{uid} → displayName
_finish_op(...) — фоновый поллинг до dtFinish
_log(msg) — лог в stdout + /tmp/app-autotest.log
"""
from flask import Blueprint, current_app, jsonify, request
import uuid
import os
import json
import fcntl
from api.http_client import HttpClient, detect_endpoint, stand_name
from api.auth import get_token, get_client, get_client_id, get_stand, get_token_info
from operations.get_services import get_services, get_service_detail
from operations.get_instances import get_instances
from operations.get_params import get_params_with_current_values, _normalize_value_list
from operations.tracker import add as tracker_add, list_all as tracker_list
from operations.tracker import remove as tracker_remove
bp = Blueprint("api_test", __name__)
AUTOTEST_PREFIX = "autotest-"
LOG_FILE = "/tmp/app-autotest.log"
MAX_LOG_SIZE = 512 * 1024 # 512 KB — ротация
MAX_LOG_LINES = 200 # сколько отдавать в /api/log
def _log(msg):
"""Пишет в stdout (gunicorn) и в файл (UI-панель, общий для всех воркеров)."""
print(msg, flush=True)
try:
with open(LOG_FILE, "a") as f:
fcntl.flock(f, fcntl.LOCK_EX)
f.write(msg + "\n")
fcntl.flock(f, fcntl.LOCK_UN)
# Ротация если разросся
st = os.stat(LOG_FILE)
if st.st_size > MAX_LOG_SIZE:
with open(LOG_FILE, "r+") as f:
fcntl.flock(f, fcntl.LOCK_EX)
f.seek(st.st_size // 2)
f.readline() # доесть строку
rest = f.read()
f.seek(0)
f.truncate()
f.write(rest)
fcntl.flock(f, fcntl.LOCK_UN)
except Exception:
pass # молча — не ронять запрос из-за лога
def _with_prefix(name):
name = (name or "").strip() or "instance"
return name if name.startswith(AUTOTEST_PREFIX) else f"{AUTOTEST_PREFIX}{name}"
def _unique_display_name(client, requested_name):
base_name = _with_prefix(requested_name)
try:
existing = {
i.get("displayName", "")
for i in get_instances(client)
if str(i.get("displayName", "")).startswith(AUTOTEST_PREFIX)
}
except Exception:
existing = set()
if base_name not in existing:
return base_name
while True:
candidate = f"{base_name}-{uuid.uuid4().hex[:6]}"
if candidate not in existing:
return candidate
@bp.route("/api/services")
def api_services():
try:
raw = get_services(get_client())
svc_list = [{"svcId": s["svcId"], "svc": s["svc"], "svcExtendedName": s.get("svcExtendedName", "")} for s in raw]
svc_list.sort(key=lambda s: s["svcId"])
return jsonify(svc_list)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/api/instances/list")
def api_instances_list():
try:
raw = get_instances(get_client())
inst = [{"instanceUid": i["instanceUid"], "displayName": i["displayName"], "serviceId": i["serviceId"], "svc": i["svc"], "explainedStatus": i.get("explainedStatus", "?")} for i in raw]
return jsonify(inst)
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/api/params/<int:op_id>")
def api_params(op_id):
"""Параметры операции. Если передан instanceUid — с текущими значениями из API."""
try:
instance_uid = request.args.get("instanceUid")
_log(f"[api_params] op_id={op_id} instanceUid={instance_uid!r}")
if instance_uid:
# Текущие значения: GET /instances/{uid} → state.params + шаблон → слияние
result = get_params_with_current_values(get_client(), op_id, instance_uid)
_log(f"[api_params] MERGED returning {len(result)} params from state.params")
for p in result:
_log(f"[api_params] {p['name']}: defaultValue={p['defaultValue']!r}")
return jsonify({"params": result})
# Без instanceUid — шаблонные значения по умолчанию (CREATE)
_log(f"[api_params] DEFAULT branch: template defaults")
data = get_client().get(f"/instanceOperations/default/{op_id}")
params = data["svcOperation"]["cfsParams"]
result = []
for p in params:
dd = p.get("dataDescriptor")
result.append({
"svcOperationCfsParamId": p["svcOperationCfsParamId"],
"name": p.get("svcOperationCfsParam", ""),
"dataType": p.get("dataType", ""),
"isRequired": p.get("isRequired", False),
"defaultValue": p.get("defaultValue"),
"valueList": _normalize_value_list(p.get("valueList")),
"refSvcId": p.get("refSvcId"),
"dataDescriptor": {k: {"dataType": v.get("dataType",""), "valueList": _normalize_value_list(v.get("valueList","")), "defaultValue": v.get("defaultValue",""), "isRequired": v.get("isRequired", False)} for k, v in dd.items()} if isinstance(dd, dict) else None,
})
_log(f"[api_params] DEFAULT returning {len(result)} params")
return jsonify({"params": result})
except Exception as e:
import traceback
_log(f"[api_params] EXCEPTION: {e}")
traceback.print_exc()
return jsonify({"error": str(e)}), 500
@bp.route("/api/test", methods=["POST"])
def api_test():
"""Запустить операцию — возвращает opUid сразу, выполнение в фоне."""
import threading, time
data = request.get_json()
svc_id = data["serviceId"]
op_name = data["operation"]
svc_op_id = data["svcOperationId"]
params = data.get("params", {})
instance_uid = data.get("instanceUid")
display_name = data.get("displayName") or ""
client = get_client()
try:
if op_name == "create":
if not display_name:
display_name = f"autotest-{svc_id}"
display_name = _unique_display_name(client, display_name)
descr = f"created by autotest v{current_app.config.get('VERSION', '')}"
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:
return jsonify({"status": "FAIL", "error": "Не удалось получить instanceUid"}), 500
op_payload = {"instanceUid": instance_uid, "operation": "create"}
op_resp = client.post("/instanceOperations", op_payload)
op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
if not op_uid:
return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500
# Записать в трекер СРАЗУ после получения instanceUid, до params/run
# (если params упадут — инстанс уже отслежен, сирот не будет)
try:
tracker_add(get_client_id(), get_stand(), instance_uid, svc_id, display_name)
except Exception as e:
import traceback
print(f"[TRACKER ERROR] add failed: {e}", flush=True)
traceback.print_exc()
# Отправка параметров (шаги 3-7 Terraform)
_send_params_terraform(client, op_uid, params)
client.post(f"/instanceOperations/{op_uid}/run")
# фоном ждать завершения
user_email = get_token_info().get("email", "")
app_version = current_app.config.get("VERSION", "")
threading.Thread(target=_finish_op, args=(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, True, get_client_id(), get_stand(), False, params, user_email, app_version), daemon=True).start()
return jsonify({"status": "RUNNING", "opUid": op_uid, "instanceUid": instance_uid, "displayName": display_name})
else:
if not instance_uid:
return jsonify({"status": "FAIL", "error": "Нет instanceUid"}), 400
if op_name == "delete":
# Жёсткое удаление через CMDB (без авторизации) — для недосозданных инстансов
cmdb_url = f"https://cmdb-api.deck.nubes.ru/instances/{instance_uid}"
cmdb_ok, cmdb_code = client.raw_delete(cmdb_url)
_log(f"[CMDB] DELETE {instance_uid} → ok={cmdb_ok} code={cmdb_code}")
if cmdb_ok:
return jsonify({"status": "OK", "opUid": "cmdb-"+instance_uid[:8], "instanceUid": instance_uid, "displayName": display_name})
if op_name == "redeploy":
op_payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id, "operation": op_name}
op_resp = client.post("/instanceOperations", op_payload)
op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
if not op_uid:
return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500
client.post(f"/instanceOperations/{op_uid}/run")
else:
op_payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id, "operation": op_name}
op_resp = client.post("/instanceOperations", op_payload)
op_uid = _find_uid(op_resp) or _uid_from_location(op_resp.get("_location", ""))
if not op_uid:
return jsonify({"status": "FAIL", "error": "Не удалось получить opUid"}), 500
_send_params_terraform(client, op_uid, params)
client.post(f"/instanceOperations/{op_uid}/run")
# фоном ждать завершения
is_delete = (op_name == "delete")
if not display_name:
display_name = _get_instance_display_name(client, instance_uid)
user_email = get_token_info().get("email", "")
app_version = current_app.config.get("VERSION", "")
threading.Thread(target=_finish_op, args=(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, False, get_client_id(), get_stand(), is_delete, params, user_email, app_version), daemon=True).start()
return jsonify({"status": "RUNNING", "opUid": op_uid, "instanceUid": instance_uid, "displayName": display_name})
except Exception as e:
import traceback
return jsonify({"status": "FAIL", "error": str(e) + " | " + traceback.format_exc()[-200:]})
# Результаты фоновых операций: opUid → {status, error, stages, duration, _ts}
# _ts — timestamp добавления, для TTL-очистки (макс. 500 записей или старше 1 часа)
_op_results = {}
_MAX_OP_RESULTS = 500
def _normalize_value(val, data_type, data_descriptor):
"""normalizeUniversalValueV6 — Terraform-equivalent.
Пустые значения нормализуются по типу: integer→"0", boolean→"false",
array→"[]", map/json→"{}", map-fixed→сборка из sub-param defaults."""
val = (val or "").strip()
if val.lower() == "null" or val == '""':
val = ""
dt = (data_type or "").lower()
# map-fixed/map с DataDescriptor: пусто или {} → собрать из sub-param defaults
if ("map" in dt) and isinstance(data_descriptor, dict) and data_descriptor:
if not val or val == "{}":
sub_obj = {}
for sk, sv in data_descriptor.items():
sd = sv.get("defaultValue")
sub_obj[sk] = str(sd) if sd is not None else ""
return json.dumps(sub_obj)
return val
if val:
return val
# Пустое — нормализовать по типу
if "array" in dt:
return "[]"
if "map" in dt or "json" in dt:
return "{}"
if "integer" in dt or "int" in dt:
return "0"
if "boolean" in dt or "bool" in dt:
return "false"
return val # пустая строка — отправляем как есть (Terraform тоже шлёт)
def _resolve_ref_svc(client, cfs_params, params):
"""resolveRefSvcParamValues — Terraform step 4.
Для параметров с refSvcId в dataDescriptor: если значение пустое —
подставить UUID существующего инстанса этого сервиса."""
instances = None # lazy load
for p in cfs_params:
dd = p.get("dataDescriptor")
if not isinstance(dd, dict):
continue
pid = str(p["svcOperationCfsParamId"])
user_val = {}
if pid in params:
try:
user_val = json.loads(params[pid]) if params[pid] else {}
except (json.JSONDecodeError, ValueError):
user_val = {}
for sk, sv in dd.items():
ref_id = sv.get("refSvcId")
if not ref_id:
continue
cur = user_val.get(sk, "")
if cur and cur != '""':
continue # уже заполнено
if instances is None:
try:
instances = get_instances(client)
except Exception:
return
match = next((i for i in instances if i.get("serviceId") == ref_id and i.get("explainedStatus") not in ("deleted",)), None)
if match:
user_val[sk] = match["instanceUid"]
if user_val:
params[pid] = json.dumps(user_val)
def _send_params_terraform(client, op_uid, params):
"""Terraform steps 3-7: cfsParams → resolveRefSvc → send user → send all unsent → validate.
Работает для CREATE и non-CREATE одинаково."""
# Шаг 3: GET cfsParams реальной операции
op_det = client.get(f"/instanceOperations/{op_uid}?fields=cfsParams")
cfs_params = op_det.get("instanceOperation", {}).get("cfsParams", [])
# Шаг 4: resolveRefSvcParamValues
_resolve_ref_svc(client, cfs_params, params)
# Шаг 5: отправить ВСЕ пользовательские params (включая пустые!)
sent = set()
for pid, pval in params.items():
client.post("/instanceOperationCfsParams", {
"instanceOperationUid": op_uid,
"svcOperationCfsParamId": int(pid),
"paramValue": str(pval)
})
sent.add(int(pid))
# Шаг 6: дослать ВСЕ неотправленные params из cfsParams
for p in cfs_params:
pid = p["svcOperationCfsParamId"]
if pid in sent:
continue
# nil-check: paramValue может быть None (отличаем от "")
val = p.get("paramValue")
if val is None:
val = p.get("defaultValue")
if val is None:
val = ""
val = _normalize_value(str(val), p.get("dataType", ""), p.get("dataDescriptor"))
client.post("/instanceOperationCfsParams", {
"instanceOperationUid": op_uid,
"svcOperationCfsParamId": pid,
"paramValue": val
})
# Шаг 7: validate-cfs (пустой ответ = успех)
try:
client.get(f"/instanceOperations/{op_uid}/validate-cfs")
except Exception as e:
if "Expecting value" in str(e) or "JSON" in str(type(e).__name__):
pass # пустой ответ = OK
else:
raise
def _cleanup_op_results():
"""Удалить старые записи: старше 1 часа или сверх лимита."""
import time
now = time.time()
# Удалить старше 1 часа
stale = [k for k, v in _op_results.items() if now - v.get("_ts", 0) > 3600]
for k in stale:
del _op_results[k]
# Если всё ещё много — удалить самые старые
if len(_op_results) > _MAX_OP_RESULTS:
sorted_keys = sorted(_op_results.keys(), key=lambda k: _op_results[k].get("_ts", 0))
for k in sorted_keys[:len(_op_results) - _MAX_OP_RESULTS]:
del _op_results[k]
def _get_instance_display_name(client, instance_uid):
"""Получить displayName инстанса из GET /instances/{uid}."""
try:
data = client.get(f"/instances/{instance_uid}")
name = data.get("instance", {}).get("displayName", "")
return name or instance_uid # fallback — UUID если имя пустое
except Exception:
return instance_uid
def _finish_op(client, op_uid, instance_uid, svc_id, display_name, op_name, svc_op_id, is_create, client_id, stand, is_delete=False, params=None, user_email="", app_version=""):
"""Фоном ждать dtFinish и сохранить результат."""
import time
_cleanup_op_results() # очистить старые записи перед добавлением новой
t0 = time.time()
deadline = t0 + 1800
while time.time() < deadline:
try:
try:
data = client.get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,isInProgress,duration,stages,svc")
except Exception:
time.sleep(5)
continue
op = data.get("instanceOperation", {})
dt_finish = op.get("dtFinish")
# Обновляем stages по мере появления
_op_results[op_uid] = {
"status": "RUNNING",
"displayName": display_name,
"stages": op.get("stages", []),
"duration": round(time.time() - t0, 1),
"_ts": time.time(),
}
if dt_finish and str(dt_finish).strip():
is_ok = op.get("isSuccessful")
err = op.get("errorLog") or ""
print(f"[DEBUG] _finish_op op_uid={op_uid} isSuccessful={is_ok!r}", flush=True)
# tracker_add для create уже вызван синхронно в api_test()
if is_ok and is_delete:
tracker_remove(client_id, stand, instance_uid)
_op_results[op_uid] = {
"status": "OK" if is_ok else "FAIL",
"displayName": display_name,
"error": str(err) if err else "",
"stages": op.get("stages", []),
"duration": round(time.time() - t0, 1),
"_ts": time.time(),
}
# Сохранить в БД историю
try:
from db.save_run import save_run
save_run(client_id, stand,
user_email,
svc_id, op.get("svc", ""), op_name, svc_op_id,
op_uid, instance_uid, display_name,
"OK" if is_ok else "FAIL",
round(time.time() - t0, 1),
str(err) if err else "",
params or {},
op.get("stages", []),
app_version)
except Exception as e:
import traceback
print(f"[DB] save_run FAILED: {e}", flush=True)
traceback.print_exc()
return
time.sleep(5)
except Exception:
import traceback
print(f"[ERROR] _finish_op crashed: {traceback.format_exc()}", flush=True)
time.sleep(5)
_op_results[op_uid] = {"status": "TIMEOUT", "displayName": display_name, "duration": round(time.time() - t0, 1), "_ts": time.time()}
@bp.route("/api/test/status/<op_uid>")
def api_test_status(op_uid):
"""Получить текущий статус операции (поллинг с UI)."""
# сначала проверяем фоновый трекер
if op_uid in _op_results:
return jsonify(_op_results[op_uid])
# иначе спрашиваем API напрямую
try:
data = get_client().get(f"/instanceOperations/{op_uid}?fields=dtFinish,isSuccessful,errorLog,isInProgress,duration,stages")
op = data.get("instanceOperation", {})
dt_finish = op.get("dtFinish")
done = bool(dt_finish and str(dt_finish).strip())
return jsonify({
"status": "OK" if (done and op.get("isSuccessful")) else ("FAIL" if done else "RUNNING"),
"done": done,
"displayName": op.get("displayName"),
"isSuccessful": op.get("isSuccessful"),
"isInProgress": op.get("isInProgress"),
"duration": op.get("duration"),
"stages": op.get("stages", []),
"errorLog": op.get("errorLog"),
})
except Exception as e:
return jsonify({"error": str(e)}), 500
@bp.route("/api/log")
def api_log():
"""Отдать последние MAX_LOG_LINES строк лога для UI-панели."""
try:
with open(LOG_FILE, "r") as f:
fcntl.flock(f, fcntl.LOCK_SH)
lines = f.readlines()
fcntl.flock(f, fcntl.LOCK_UN)
return jsonify([l.rstrip("\n") for l in lines[-MAX_LOG_LINES:]])
except FileNotFoundError:
return jsonify([])
except Exception:
return jsonify([])
@bp.route("/api/history")
def api_history():
"""Последние 50 записей истории тестов (фильтр по client_id + stand)."""
try:
from db.pool import get_conn, put_conn
conn = get_conn()
if not conn:
return jsonify([])
cur = conn.cursor()
cur.execute("""
SELECT id, created_at, client_id, stand, user_email,
svc_id, svc_name, op_name, svc_op_id, op_uid,
instance_uid, display_name,
status, duration_sec, error_log, app_version
FROM runs
WHERE client_id = %s AND stand = %s
ORDER BY created_at DESC
LIMIT 50
""", (get_client_id(), get_stand()))
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
result = [dict(zip(cols, r)) for r in rows]
# Конвертировать datetime в строку
for r in result:
if r["created_at"]:
r["created_at"] = r["created_at"].isoformat()
cur.close()
put_conn(conn)
return jsonify(result)
except Exception as e:
try:
cur.close()
except Exception:
pass
try:
put_conn(conn)
except Exception:
pass
return jsonify({"error": str(e)}), 500
def _find_uid(resp):
"""Извлечь instanceUid или instanceOperationUid из ответа API."""
if isinstance(resp, dict):
for key in ("instanceOperationUid", "instanceUid", "uid", "Uid"):
v = resp.get(key)
if isinstance(v, str) and v:
return v
return None
def _uid_from_location(loc):
if loc:
parts = loc.rstrip("/").split("/")
if len(parts[-1]) == 36:
return parts[-1]
return None