app-autotest/site/routes/api_test.py

from flask import Blueprint, current_app, jsonify, request

from api.http_client import HttpClient
from operations.get_services import get_services, get_service_detail
from operations.get_instances import get_instances

bp = Blueprint("api_test", __name__)


def _client():
    token = request.cookies.get("token") or current_app.config["NUBES_API_TOKEN"]
    return HttpClient(current_app.config["NUBES_API_ENDPOINT"], token)


@bp.route("/api/services")
def api_services():
    try:
        raw = get_services(_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(_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/operations/<int:svc_id>")
def api_operations(svc_id):
    try:
        detail = get_service_detail(_client(), svc_id)
        ops = detail.get("operations", [])
        instances = get_instances(_client())
        svc_instances = [i for i in instances if i.get("serviceId") == svc_id and i.get("explainedStatus") not in ("deleted", "not created")]
        return jsonify({
            "svc": detail.get("svc", ""),
            "operations": [{"svcOperationId": o["svcOperationId"], "operation": o["operation"]} for o in ops],
            "instances": svc_instances,
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/params/<int:op_id>")
def api_params(op_id):
    try:
        data = _client().get(f"/instanceOperations/default/{op_id}")
        params = data["svcOperation"]["cfsParams"]
        result = []
        for p in params:
            result.append({
                "svcOperationCfsParamId": p["svcOperationCfsParamId"],
                "name": p.get("svcOperationCfsParam", ""),
                "dataType": p.get("dataType", ""),
                "isRequired": p.get("isRequired", False),
                "defaultValue": p.get("defaultValue"),
                "valueList": p.get("valueList"),
                "dataDescriptor": {k: {"dataType": v.get("dataType",""), "valueList": v.get("valueList",""), "isRequired": v.get("isRequired", False)} for k, v in p.get("dataDescriptor", {}).items()} if p.get("dataDescriptor") else None,
            })
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 500


@bp.route("/api/test", methods=["POST"])
def api_test():
    """Запустить одну операцию."""
    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", f"autotest-{svc_id}")

    client = _client()

    result = {"status": "RUNNING", "error": None, "instanceUid": None, "duration": 0}
    t0 = time.time()

    try:
        if op_name == "create":
            payload = {"serviceId": svc_id, "displayName": display_name}
            if params:
                payload["cfsParams"] = [{"svcOperationCfsParamId": int(k), "paramValue": str(v)} for k, v in params.items()]
            resp = client.post("/instances", payload)
            instance_uid = resp.get("instanceUid") or _find_uid(resp)
            # Location header: /instances/{uid}
            loc = resp.get("_location", "")
            if not instance_uid and loc:
                parts = loc.rstrip("/").split("/")
                instance_uid = parts[-1] if len(parts[-1]) == 36 else None

            # wait
            deadline = time.time() + 300
            while time.time() < deadline:
                insts = get_instances(client)
                for i in insts:
                    if i.get("instanceUid") == instance_uid:
                        if not i.get("operationIsInProgress") and i.get("explainedStatus") not in ("creating",):
                            result["status"] = "OK"
                            result["instanceUid"] = instance_uid
                            result["duration"] = round(time.time() - t0, 1)
                            return jsonify(result)
                time.sleep(5)
            result["status"] = "TIMEOUT"
        else:
            if not instance_uid:
                return jsonify({"status": "FAIL", "error": "Нет instanceUid"}), 400
            payload = {"instanceUid": instance_uid, "svcOperationId": svc_op_id}
            if params:
                payload["cfsParams"] = [{"svcOperationCfsParamId": int(k), "paramValue": str(v)} for k, v in params.items()]
            client.post("/instanceOperations", payload)

            deadline = time.time() + 300
            while time.time() < deadline:
                insts = get_instances(client)
                for i in insts:
                    if i.get("instanceUid") == instance_uid:
                        if not i.get("operationIsInProgress"):
                            result["status"] = "OK"
                            result["instanceUid"] = instance_uid
                            result["duration"] = round(time.time() - t0, 1)
                            return jsonify(result)
                time.sleep(5)
            result["status"] = "TIMEOUT"
    except Exception as e:
        result["status"] = "FAIL"
        result["error"] = str(e)

    result["instanceUid"] = instance_uid
    result["duration"] = round(time.time() - t0, 1)
    return jsonify(result)


def _find_uid(resp):
    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