228 lines
9.0 KiB
Python
228 lines
9.0 KiB
Python
"""
|
||
Интеграционные тесты app-autotest ↔ polygon.
|
||
|
||
Каждый тест требует запущенного polygon (фикстура polygon_server).
|
||
Перед каждым тестом состояние polygon сбрасывается (autouse reset_polygon).
|
||
|
||
Все запросы — через requests (HTTP), имитируя работу HttpClient из app-autotest.
|
||
"""
|
||
|
||
import pytest
|
||
import requests
|
||
|
||
API = "/api/v1/svc"
|
||
|
||
|
||
def _create_instance(url, service_id=1, display_name="test"):
|
||
"""POST /instances → извлечь instanceUid из Location."""
|
||
r = requests.post(
|
||
f"{url}{API}/instances",
|
||
json={"serviceId": service_id, "displayName": display_name},
|
||
allow_redirects=False,
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 201
|
||
location = r.headers.get("Location", "")
|
||
uid = location.rsplit("/", 1)[-1]
|
||
assert len(uid) >= 32 # UUID без дефисов
|
||
return uid
|
||
|
||
|
||
def _create_operation(url, instance_uid, operation, svc_op_id=None):
|
||
"""POST /instanceOperations → извлечь opUid из Location."""
|
||
body = {"instanceUid": instance_uid, "operation": operation}
|
||
if svc_op_id is not None:
|
||
body["svcOperationId"] = svc_op_id
|
||
r = requests.post(
|
||
f"{url}{API}/instanceOperations",
|
||
json=body,
|
||
allow_redirects=False,
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 201
|
||
location = r.headers.get("Location", "")
|
||
uid = location.rsplit("/", 1)[-1]
|
||
assert len(uid) >= 32
|
||
return uid
|
||
|
||
|
||
def _run(url, op_uid):
|
||
"""POST /run → дождаться dtFinish."""
|
||
r = requests.post(f"{url}{API}/instanceOperations/{op_uid}/run", timeout=10)
|
||
assert r.status_code == 200
|
||
# Проверить что операция завершена
|
||
r2 = requests.get(f"{url}{API}/instanceOperations/{op_uid}", timeout=10)
|
||
op = r2.json()["instanceOperation"]
|
||
assert op["dtFinish"] is not None
|
||
assert op["isSuccessful"] is True
|
||
|
||
|
||
class TestServices:
|
||
"""Тесты эндпоинтов /services."""
|
||
|
||
def test_list_services(self, polygon_server):
|
||
"""GET /services возвращает все сервисы."""
|
||
r = requests.get(f"{polygon_server}{API}/services", timeout=10)
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert len(data["results"]) >= 30 # минимум 30 сервисов
|
||
|
||
def test_service_detail(self, polygon_server):
|
||
"""GET /services/1 возвращает операции Болванки."""
|
||
r = requests.get(f"{polygon_server}{API}/services/1", timeout=10)
|
||
assert r.status_code == 200
|
||
ops = r.json()["svc"]["operations"]
|
||
assert any(op["operation"] == "create" for op in ops)
|
||
|
||
def test_service_404(self, polygon_server):
|
||
"""GET /services/99999 → 404."""
|
||
r = requests.get(f"{polygon_server}{API}/services/99999", timeout=10)
|
||
assert r.status_code == 404
|
||
|
||
|
||
class TestInstanceCreate:
|
||
"""Тесты создания инстанса."""
|
||
|
||
def test_create_dummy(self, polygon_server):
|
||
"""POST /instances → 201 + Location → GET → status=creating."""
|
||
uid = _create_instance(polygon_server)
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
assert r.status_code == 200
|
||
inst = r.json()["instance"]
|
||
assert inst["status"] == "creating"
|
||
assert inst["serviceId"] == 1
|
||
|
||
def test_create_unknown_service(self, polygon_server):
|
||
"""POST /instances с несуществующим serviceId → 404."""
|
||
r = requests.post(
|
||
f"{polygon_server}{API}/instances",
|
||
json={"serviceId": 99999, "displayName": "bad"},
|
||
timeout=10,
|
||
)
|
||
assert r.status_code == 404
|
||
|
||
|
||
class TestFullCreateFlow:
|
||
"""Полный цикл create → run → проверить params."""
|
||
|
||
def test_create_and_run(self, polygon_server):
|
||
"""create → run → статус running + params из шаблона."""
|
||
uid = _create_instance(polygon_server)
|
||
op_uid = _create_operation(polygon_server, uid, "create")
|
||
_run(polygon_server, op_uid)
|
||
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
inst = r.json()["instance"]
|
||
assert inst["status"] == "running"
|
||
assert "resourceRealm" in inst["state"]["params"]
|
||
assert inst["state"]["params"]["resourceRealm"] == "dummy"
|
||
|
||
def test_create_postgres_state_out(self, polygon_server):
|
||
"""create PG → state.out содержит users, databases."""
|
||
uid = _create_instance(polygon_server, service_id=90, display_name="pg")
|
||
op_uid = _create_operation(polygon_server, uid, "create")
|
||
_run(polygon_server, op_uid)
|
||
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
out = r.json()["instance"]["state"]["out"]
|
||
assert "users" in out
|
||
assert "databases" in out
|
||
|
||
def test_run_twice_409(self, polygon_server):
|
||
"""Повторный POST /run → 409."""
|
||
uid = _create_instance(polygon_server)
|
||
op_uid = _create_operation(polygon_server, uid, "create")
|
||
_run(polygon_server, op_uid)
|
||
|
||
r = requests.post(
|
||
f"{polygon_server}{API}/instanceOperations/{op_uid}/run", timeout=10
|
||
)
|
||
assert r.status_code == 409
|
||
|
||
|
||
class TestOperations:
|
||
"""Тесты операций: modify, suspend, resume, delete."""
|
||
|
||
def test_modify_merges_params(self, polygon_server):
|
||
"""modify → новый paramValue в state.params."""
|
||
uid = _create_instance(polygon_server)
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "create"))
|
||
|
||
# Установить новое значение durationMs (modify-версия: paramId=287, create-версия: 198)
|
||
op_uid = _create_operation(polygon_server, uid, "modify")
|
||
requests.post(
|
||
f"{polygon_server}{API}/instanceOperationCfsParams",
|
||
json={"instanceOperationUid": op_uid, "svcOperationCfsParamId": 287, "paramValue": "999"},
|
||
timeout=10,
|
||
)
|
||
_run(polygon_server, op_uid)
|
||
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
assert r.json()["instance"]["state"]["params"]["durationMs"] == "999"
|
||
|
||
def test_suspend_resume(self, polygon_server):
|
||
"""suspend → status=suspended, resume → status=running."""
|
||
uid = _create_instance(polygon_server)
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "create"))
|
||
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "suspend"))
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
assert r.json()["instance"]["status"] == "suspended"
|
||
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "resume"))
|
||
r = requests.get(f"{polygon_server}{API}/instances/{uid}", timeout=10)
|
||
assert r.json()["instance"]["status"] == "running"
|
||
|
||
def test_delete_removes_instance(self, polygon_server):
|
||
"""delete → инстанс удалён из списка."""
|
||
uid = _create_instance(polygon_server)
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "create"))
|
||
_run(polygon_server, _create_operation(polygon_server, uid, "delete"))
|
||
|
||
r = requests.get(f"{polygon_server}{API}/instances", timeout=10)
|
||
uids = [i["instanceUid"] for i in r.json()["results"]]
|
||
assert uid not in uids
|
||
|
||
|
||
class TestPagination:
|
||
"""Тесты пагинации."""
|
||
|
||
def test_pagination(self, polygon_server):
|
||
"""3 инстанса, pageSize=2 → 2 страницы."""
|
||
for i in range(3):
|
||
_create_instance(polygon_server, display_name=f"p{i}")
|
||
|
||
r1 = requests.get(
|
||
f"{polygon_server}{API}/instances", params={"pageSize": 2, "page": 1}, timeout=10
|
||
)
|
||
assert len(r1.json()["results"]) == 2
|
||
|
||
r2 = requests.get(
|
||
f"{polygon_server}{API}/instances", params={"pageSize": 2, "page": 2}, timeout=10
|
||
)
|
||
assert len(r2.json()["results"]) == 1 # последний
|
||
|
||
|
||
class TestMockEndpoints:
|
||
"""Тесты /_mock/* эндпоинтов."""
|
||
|
||
def test_mock_state(self, polygon_server):
|
||
"""GET /_mock/state — отладочный дамп."""
|
||
_create_instance(polygon_server)
|
||
r = requests.get(f"{polygon_server}{API}/_mock/state", timeout=10)
|
||
assert r.status_code == 200
|
||
assert len(r.json()["instances"]) == 1
|
||
|
||
def test_mock_services(self, polygon_server):
|
||
"""GET /_mock/services — список конфигов."""
|
||
r = requests.get(f"{polygon_server}{API}/_mock/services", timeout=10)
|
||
assert r.status_code == 200
|
||
assert r.json()["count"] >= 30
|
||
|
||
def test_mock_delay(self, polygon_server):
|
||
"""POST /_mock/delay — изменение задержки."""
|
||
r = requests.post(f"{polygon_server}{API}/_mock/delay/2", timeout=10)
|
||
assert r.json()["delay"] == 2.0
|
||
# Сбросить
|
||
requests.post(f"{polygon_server}{API}/_mock/delay/0.1", timeout=10)
|