147 lines
4.7 KiB
Python
147 lines
4.7 KiB
Python
"""
|
|
mock_state.py — состояние мок-полигона в памяти.
|
|
|
|
Синглтон state хранит:
|
|
instances — {instanceUid: instance_dict}
|
|
operations — {opUid: operation_dict}
|
|
op_params — {opUid: {paramId(int): paramValue(str)}}
|
|
|
|
Все методы — синхронные (один gunicorn-воркер на тестах, гонок нет).
|
|
"""
|
|
|
|
import uuid
|
|
|
|
from utils.now import now as _now
|
|
|
|
|
|
class MockState:
|
|
"""Хранилище состояния всех инстансов и операций."""
|
|
|
|
def __init__(self):
|
|
self.instances = {}
|
|
self.operations = {}
|
|
self.op_params = {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Инстансы
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_instance(self, service_id, display_name, svc_def):
|
|
"""Создать shell инстанса (статус 'creating').
|
|
|
|
Возвращает instanceUid (UUID v4).
|
|
"""
|
|
uid = str(uuid.uuid4())
|
|
self.instances[uid] = {
|
|
"instanceUid": uid,
|
|
"serviceId": service_id,
|
|
"displayName": display_name,
|
|
"descr": "",
|
|
"status": "creating",
|
|
"explainedStatus": "Создание",
|
|
"svc": svc_def.get("service_display_name", svc_def.get("name", "")),
|
|
"dtCreate": _now(),
|
|
"state": {
|
|
"params": {},
|
|
"out": {},
|
|
},
|
|
}
|
|
return uid
|
|
|
|
def get_instance(self, uid):
|
|
"""Получить инстанс по uid или None."""
|
|
return self.instances.get(uid)
|
|
|
|
def list_instances(self, page_size=200, page=1):
|
|
"""Список инстансов с пагинацией.
|
|
|
|
page_size capped at 200. Стоп по len(batch) < page_size.
|
|
"""
|
|
page_size = min(page_size, 200)
|
|
page = max(page, 1)
|
|
|
|
all_uids = list(self.instances.keys())
|
|
start = (page - 1) * page_size
|
|
batch = all_uids[start:start + page_size]
|
|
|
|
results = []
|
|
for uid in batch:
|
|
inst = self.instances[uid]
|
|
results.append({
|
|
"instanceUid": inst["instanceUid"],
|
|
"serviceId": inst["serviceId"],
|
|
"displayName": inst["displayName"],
|
|
"svc": inst["svc"],
|
|
"explainedStatus": inst["explainedStatus"],
|
|
})
|
|
|
|
return {
|
|
"results": results,
|
|
"pageSize": page_size,
|
|
"page": page,
|
|
"total": len(all_uids),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Операции
|
|
# ------------------------------------------------------------------
|
|
|
|
def create_operation(self, instance_uid, svc_operation_id, operation, kind, action):
|
|
"""Создать операцию (dtFinish=None, ещё не запущена).
|
|
|
|
Возвращает opUid (UUID v4).
|
|
"""
|
|
op_uid = str(uuid.uuid4())
|
|
inst = self.instances.get(instance_uid, {})
|
|
self.operations[op_uid] = {
|
|
"instanceOperationUid": op_uid,
|
|
"instanceUid": instance_uid,
|
|
"svcOperationId": svc_operation_id,
|
|
"operation": operation,
|
|
"kind": kind,
|
|
"action": action,
|
|
"dtStart": None,
|
|
"dtFinish": None,
|
|
"isSuccessful": None,
|
|
"errorLog": "",
|
|
"svc": inst.get("svc", ""),
|
|
"stages": [],
|
|
}
|
|
self.op_params[op_uid] = {}
|
|
return op_uid
|
|
|
|
def get_operation(self, op_uid):
|
|
"""Получить операцию по uid или None."""
|
|
return self.operations.get(op_uid)
|
|
|
|
def set_param(self, op_uid, param_id, value):
|
|
"""Установить значение параметра операции.
|
|
|
|
param_id — int (svcOperationCfsParamId).
|
|
value — str.
|
|
"""
|
|
if op_uid not in self.op_params:
|
|
self.op_params[op_uid] = {}
|
|
self.op_params[op_uid][int(param_id)] = value
|
|
|
|
def get_params(self, op_uid):
|
|
"""Получить все установленные параметры операции.
|
|
|
|
Возвращает {paramId: value}.
|
|
"""
|
|
return self.op_params.get(op_uid, {})
|
|
|
|
# ------------------------------------------------------------------
|
|
# Сброс
|
|
# ------------------------------------------------------------------
|
|
|
|
def reset(self):
|
|
"""Полный сброс состояния (для тестов)."""
|
|
self.instances.clear()
|
|
self.operations.clear()
|
|
self.op_params.clear()
|
|
|
|
|
|
# Глобальный синглтон
|
|
state = MockState()
|