app-autotest/site/runner.py

import threading
import time
import yaml
import os

from api.http_client import HttpClient


CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")

_current_run = None
_lock = threading.Lock()


def load_config():
    with open(CONFIG_PATH) as f:
        return yaml.safe_load(f)


def save_config(data):
    with open(CONFIG_PATH, "w") as f:
        yaml.dump(data, f, allow_unicode=True, default_flow_style=False)


def get_status():
    return _current_run


def _wait_for_ready(client, instance_uid, timeout=300):
    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):
    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):
    global _current_run
    client = HttpClient(endpoint, token)
    config = load_config()

    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  # дальше операции этого сервиса бессмысленны
            finally:
                result["duration_sec"] = round(time.time() - t0, 1)

        # wait between services
        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):
    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