"""
polygon v0.2.0 — Mock Nubes API.
Эмулятор REST API облачной платформы Nubes для интеграционных тестов.
Все эндпоинты с префиксом /api/v1/svc притворяются реальным Nubes API.
Состояние инстансов/операций — в памяти (MockState).
Конфиги сервисов — из services/*.yaml (сгенерированы from_stands.py).
Деплой: Nubes pythonk8s (managed service) с gunicorn:
gunicorn app:app --workers 1 --bind 0.0.0.0:8000
⛔ РОВНО 1 ВОРКЕР — состояние в памяти, не shared между процессами.
Переменные окружения:
MOCK_OP_DELAY — задержка операции в секундах (по умолчанию 0.1)
WEB_CONCURRENCY — число gunicorn-воркеров (принудительно 1)
"""
import os
import time
from datetime import datetime, timezone
from flask import Flask, jsonify, make_response, request
# ⛔ Принудительно 1 воркер — состояние в памяти, иначе инстансы теряются.
# setdefault не перезапишет если Nubes/Procfile уже выставил WEB_CONCURRENCY=1.
os.environ.setdefault("WEB_CONCURRENCY", "1")
import config_loader
import mock_state
import state_machine
VERSION = "0.2.0"
# Задержка операции (синхронный sleep в /run)
MOCK_OP_DELAY = float(os.getenv("MOCK_OP_DELAY", "0.1"))
app = Flask(__name__, template_folder="templates", static_folder="static")
# Загружаем сервисы при старте
_SERVICES, _OPS_INDEX = config_loader.load_services("services")
_STATE = mock_state.state
def _now():
"""UTC-строка в ISO-формате с 'Z'."""
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
# ===========================================================================
# Health / Root
# ===========================================================================
@app.route("/health")
def health():
"""Healthcheck для Nubes — всегда возвращает 200 OK."""
return "OK"
@app.route("/")
def index():
"""Корневой эндпоинт — HTML с версией и статусом."""
svc_count = len(_SERVICES)
inst_count = len(_STATE.instances)
return f"""
polygon v{VERSION}
polygon v{VERSION}
● running
Сервисов: {svc_count} | Инстансов: {inst_count} | delay: {MOCK_OP_DELAY}s
Эндпоинты: /api/v1/svc/*
"""
# ===========================================================================
# Services (GET /services, GET /services/{id})
# ===========================================================================
@app.route("/api/v1/svc/services", methods=["GET"])
def list_services():
"""GET /services — список всех сервисов."""
results = []
for svc_id, svc_def in _SERVICES.items():
results.append({
"svcId": svc_id,
"svc": svc_def.get("service_display_name", ""),
"svcShort": svc_def.get("service_short_name", svc_def.get("name", "")),
})
return jsonify({"results": results})
@app.route("/api/v1/svc/services/", methods=["GET"])
def get_service(svc_id):
"""GET /services/{id} — детали сервиса (список операций)."""
svc_def = _SERVICES.get(svc_id)
if not svc_def:
return jsonify({"error": "service not found"}), 404
ops = []
for op in svc_def.get("operations", []):
ops.append({
"svcOperationId": op["svcOperationId"],
"operation": op["operation"],
})
return jsonify({
"svc": {
"svc": svc_def.get("service_display_name", ""),
"svcShort": svc_def.get("service_short_name", svc_def.get("name", "")),
"operations": ops,
}
})
# ===========================================================================
# Instances (GET list, GET detail, POST create)
# ===========================================================================
@app.route("/api/v1/svc/instances", methods=["GET"])
def list_instances():
"""GET /instances — список с пагинацией."""
page_size = request.args.get("pageSize", 200, type=int)
page = request.args.get("page", 1, type=int)
result = _STATE.list_instances(page_size=page_size, page=page)
return jsonify(result)
@app.route("/api/v1/svc/instances/", methods=["GET"])
def get_instance(uid):
"""GET /instances/{uid} — полные данные инстанса."""
inst = _STATE.get_instance(uid)
if not inst:
return jsonify({"error": "instance not found"}), 404
return jsonify({"instance": dict(inst)})
@app.route("/api/v1/svc/instances", methods=["POST"])
def create_instance():
"""POST /instances — создать shell инстанса.
Ожидает JSON: {serviceId, displayName}
Возвращает 201 + Location: ./{instanceUid} + {"instanceUid": uid}
"""
body = request.get_json(silent=True) or {}
service_id = body.get("serviceId")
display_name = body.get("displayName", "unnamed")
if service_id is None:
return jsonify({"error": "serviceId is required"}), 400
svc_def = _SERVICES.get(service_id)
if not svc_def:
return jsonify({"error": f"service {service_id} not found"}), 404
uid = _STATE.create_instance(service_id, display_name, svc_def)
resp = make_response(jsonify({"instanceUid": uid}), 201)
resp.headers["Location"] = f"./{uid}"
return resp
# ===========================================================================
# Instance Operations (create, get, params, validate, run)
# ===========================================================================
# ВАЖНО: default/ ДОЛЖЕН быть выше — иначе Flask
# распарсит "default" как uid и уйдёт в get_operation с op_id="default".
@app.route("/api/v1/svc/instanceOperations/default/", methods=["GET"])
def get_operation_default(op_id):
"""GET /instanceOperations/default/{id} — шаблон операции (cfsParams).
Это ключевой эндпоинт для app-autotest: он получает отсюда
список параметров с dataDescriptor, valueList, isModifiable.
"""
svc_def = _OPS_INDEX.get(op_id)
if not svc_def:
return jsonify({"error": "operation not found"}), 404
# Найти операцию по svcOperationId
target_op = None
for op in svc_def.get("operations", []):
if op["svcOperationId"] == op_id:
target_op = op
break
if not target_op:
return jsonify({"error": "operation not found"}), 404
# Собрать cfsParams для этой операции
cfs_params_by_op = svc_def.get("cfsParamsByOp", {})
cfs_params = {p["svcOperationCfsParamId"]: p for p in svc_def.get("cfsParams", [])}
param_ids = cfs_params_by_op.get(op_id, [])
cfs_list = []
for pid in param_ids:
p = cfs_params.get(pid)
if p:
cfs_list.append(dict(p))
return jsonify({
"svcOperation": {
"svcOperationId": op_id,
"operation": target_op.get("operation", ""),
"cfsParams": cfs_list,
}
})
@app.route("/api/v1/svc/instanceOperations", methods=["POST"])
def create_operation():
"""POST /instanceOperations — создать операцию.
Ожидает JSON: {instanceUid, svcOperationId?, operation}
Если operation="create" — svcOperationId может отсутствовать,
polygon находит его сам из service_def.
Возвращает 201 + Location: ./{opUid} + {"instanceOperationUid": opUid}
"""
body = request.get_json(silent=True) or {}
instance_uid = body.get("instanceUid")
operation = body.get("operation", "")
svc_op_id = body.get("svcOperationId")
if not instance_uid:
return jsonify({"error": "instanceUid is required"}), 400
inst = _STATE.get_instance(instance_uid)
if not inst:
return jsonify({"error": "instance not found"}), 404
# Если svcOperationId не передан — найти по operation name
if svc_op_id is None:
svc_id = inst.get("serviceId")
svc_def = _SERVICES.get(svc_id, {})
for op in svc_def.get("operations", []):
if op["operation"] == operation:
svc_op_id = op["svcOperationId"]
break
if svc_op_id is None:
return jsonify({"error": f"operation '{operation}' not found for service"}), 404
svc_def = _OPS_INDEX.get(svc_op_id, {})
kind = "instance"
action = operation
for op in svc_def.get("operations", []):
if op["svcOperationId"] == svc_op_id:
kind = op.get("kind", "instance")
action = op.get("action", operation)
break
op_uid = _STATE.create_operation(instance_uid, svc_op_id, operation, kind, action)
resp = make_response(jsonify({"instanceOperationUid": op_uid}), 201)
resp.headers["Location"] = f"./{op_uid}"
return resp
@app.route("/api/v1/svc/instanceOperations/", methods=["GET"])
def get_operation(uid):
"""GET /instanceOperations/{uid} — статус операции.
Query params: ?fields=cfsParams,... — app-autotest фильтрует поля.
Мок отдаёт всё, фильтрация на клиенте.
"""
op = _STATE.get_operation(uid)
if not op:
return jsonify({"error": "operation not found"}), 404
result = dict(op)
# Если запрошены cfsParams — добавить их
fields = request.args.get("fields", "")
if "cfsParams" in fields:
svc_op_id = op.get("svcOperationId")
svc_def = _OPS_INDEX.get(svc_op_id, {})
cfs_params_by_op = svc_def.get("cfsParamsByOp", {})
cfs_params = {p["svcOperationCfsParamId"]: p for p in svc_def.get("cfsParams", [])}
param_ids = cfs_params_by_op.get(svc_op_id, [])
op_params = _STATE.get_params(uid)
cfs_list = []
for pid in param_ids:
p = cfs_params.get(pid)
if p:
entry = dict(p)
entry["paramValue"] = op_params.get(pid, "")
cfs_list.append(entry)
result["cfsParams"] = cfs_list
return jsonify({"instanceOperation": result})
@app.route("/api/v1/svc/instanceOperationCfsParams", methods=["POST"])
def set_operation_param():
"""POST /instanceOperationCfsParams — установить значение параметра.
Ожидает JSON: {instanceOperationUid, svcOperationCfsParamId, paramValue}
"""
body = request.get_json(silent=True) or {}
op_uid = body.get("instanceOperationUid")
param_id = body.get("svcOperationCfsParamId")
value = body.get("paramValue", "")
if not op_uid or param_id is None:
return jsonify({"error": "instanceOperationUid and svcOperationCfsParamId required"}), 400
if not _STATE.get_operation(op_uid):
return jsonify({"error": "operation not found"}), 404
_STATE.set_param(op_uid, param_id, value)
return jsonify({})
@app.route("/api/v1/svc/instanceOperations//validate-cfs", methods=["GET"])
def validate_cfs(uid):
"""GET /instanceOperations/{uid}/validate-cfs — всегда OK.
⛔ КРИТИЧНО: возвращает пустое тело (200), НЕ jsonify.
app-autotest ловит JSONDecodeError и считает это успехом.
"""
if not _STATE.get_operation(uid):
return jsonify({"error": "operation not found"}), 404
return "", 200
@app.route("/api/v1/svc/instanceOperations//run", methods=["POST"])
def run_operation(uid):
"""POST /instanceOperations/{uid}/run — выполнить операцию.
Синхронный: sleep(MOCK_OP_DELAY) → apply_effect → dtFinish = now.
"""
op = _STATE.get_operation(uid)
if not op:
return jsonify({"error": "operation not found"}), 404
op["dtStart"] = _now()
# Синхронная задержка
if MOCK_OP_DELAY > 0:
time.sleep(MOCK_OP_DELAY)
# Применить эффект
state_machine.apply_effect(uid, _STATE, _SERVICES)
op["dtFinish"] = _now()
op["isSuccessful"] = True
return jsonify({"ok": True})
# ===========================================================================
# Mock control (reset, state, services, delay)
# ===========================================================================
@app.route("/api/v1/svc/_mock/reset", methods=["POST"])
def mock_reset():
"""POST /_mock/reset — полный сброс состояния."""
_STATE.reset()
return jsonify({"reset": "ok"})
@app.route("/api/v1/svc/_mock/state", methods=["GET"])
def mock_state_view():
"""GET /_mock/state — отладочный дамп текущего состояния."""
return jsonify({
"instances": {
uid: {
"instanceUid": inst["instanceUid"],
"displayName": inst["displayName"],
"status": inst["status"],
"serviceId": inst["serviceId"],
}
for uid, inst in _STATE.instances.items()
},
"operations": {
uid: {
"instanceOperationUid": op["instanceOperationUid"],
"instanceUid": op["instanceUid"],
"operation": op["operation"],
"dtFinish": op["dtFinish"],
"isSuccessful": op["isSuccessful"],
}
for uid, op in _STATE.operations.items()
},
})
@app.route("/api/v1/svc/_mock/services", methods=["GET"])
def mock_services():
"""GET /_mock/services — список загруженных сервисов."""
result = {}
for svc_id, svc_def in _SERVICES.items():
result[str(svc_id)] = {
"name": svc_def.get("name", ""),
"displayName": svc_def.get("service_display_name", ""),
"operations": len(svc_def.get("operations", [])),
"params": len(svc_def.get("cfsParams", [])),
}
return jsonify({"count": len(result), "services": result})
@app.route("/api/v1/svc/_mock/delay/", methods=["POST"])
def mock_set_delay(seconds):
"""POST /_mock/delay/{s} — изменить MOCK_OP_DELAY на лету."""
global MOCK_OP_DELAY
MOCK_OP_DELAY = seconds
return jsonify({"delay": MOCK_OP_DELAY})
# ===========================================================================
# Entry point
# ===========================================================================
# Локальный запуск (без gunicorn) — только для разработки
if __name__ == "__main__":
app.run(debug=True, host="0.0.0.0", port=5000)