238 lines
9.5 KiB
Python
238 lines
9.5 KiB
Python
"""
|
||
Интеграционные тесты сценариев с полигоном.
|
||
|
||
Требует polygon_server (conftest.py, session scope).
|
||
"""
|
||
import json
|
||
import time
|
||
import pytest
|
||
|
||
|
||
# ── Хелперы ──
|
||
|
||
def _create_scenario(app_client, name, steps):
|
||
resp = app_client.post("/api/scenario/definitions",
|
||
json={"name": name, "steps": steps},
|
||
headers={"Content-Type": "application/json"})
|
||
return resp.get_json()
|
||
|
||
|
||
def _delete_scenario(app_client, def_id):
|
||
app_client.delete(f"/api/scenario/definitions/{def_id}")
|
||
|
||
|
||
def _run_scenario(app_client, def_id):
|
||
return app_client.post("/api/scenario/run",
|
||
json={"definition_id": def_id},
|
||
headers={"Content-Type": "application/json"})
|
||
|
||
|
||
def _poll_scenario(app_client, run_id, timeout=15):
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
resp = app_client.get(f"/api/scenario/run/{run_id}")
|
||
data = resp.get_json()
|
||
if data and data.get("status") != "RUNNING":
|
||
return data
|
||
time.sleep(0.5)
|
||
return {"status": "TIMEOUT"}
|
||
|
||
|
||
# ── Тесты ──
|
||
|
||
class TestScenarioLifecycle:
|
||
"""Полный цикл: создать определение -> запустить -> проверить результат."""
|
||
|
||
def test_list_scenarios(self, app_client):
|
||
resp = app_client.get("/api/scenarios")
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert isinstance(data, list)
|
||
|
||
def test_manual_create_delete_uses_local_polygon(self, app_client, monkeypatch):
|
||
"""Ручной API приложения проводит create и delete через локальный Polygon."""
|
||
from api.http_client import HttpClient
|
||
from routes import api_test as api_test_routes
|
||
|
||
monkeypatch.setattr(api_test_routes, "_finish_op", lambda *args, **kwargs: None)
|
||
monkeypatch.setattr(
|
||
HttpClient,
|
||
"raw_delete",
|
||
lambda *args, **kwargs: pytest.fail("Polygon delete must not call external CMDB"),
|
||
)
|
||
|
||
operations_response = app_client.get("/api/operations/1")
|
||
assert operations_response.status_code == 200
|
||
operations = operations_response.get_json()["operations"]
|
||
create_id = next(op["svcOperationId"] for op in operations if op["operation"] == "create")
|
||
delete_id = next(op["svcOperationId"] for op in operations if op["operation"] == "delete")
|
||
|
||
created = app_client.post("/api/test", json={
|
||
"serviceId": 1,
|
||
"operation": "create",
|
||
"svcOperationId": create_id,
|
||
"params": {},
|
||
"displayName": "autotest-local-integration",
|
||
})
|
||
assert created.status_code == 200, created.get_json()
|
||
create_result = app_client.get(f"/api/test/status/{created.get_json()['opUid']}").get_json()
|
||
assert create_result["status"] == "OK"
|
||
|
||
deleted = app_client.post("/api/test", json={
|
||
"serviceId": 1,
|
||
"operation": "delete",
|
||
"svcOperationId": delete_id,
|
||
"instanceUid": created.get_json()["instanceUid"],
|
||
"params": {},
|
||
})
|
||
assert deleted.status_code == 200, deleted.get_json()
|
||
delete_result = app_client.get(f"/api/test/status/{deleted.get_json()['opUid']}").get_json()
|
||
assert delete_result["status"] == "OK"
|
||
|
||
def test_create_and_run_simple_scenario(self, app_client):
|
||
"""Сценарий create -> delete. Проверяем что оба шага OK."""
|
||
steps = [
|
||
{"service_id": 1, "operation": "create", "params": {}, "output": "d1"},
|
||
{"service_id": 1, "operation": "delete", "instance_ref": "d1", "params": {}},
|
||
]
|
||
name = f"test-simple-{int(time.time())}"
|
||
created = _create_scenario(app_client, name, steps)
|
||
assert "id" in created, f"Create failed: {created}"
|
||
|
||
resp = _run_scenario(app_client, created["id"])
|
||
assert resp.status_code == 202, f"Run failed: {resp.get_json()}"
|
||
run_data = resp.get_json()
|
||
assert run_data["status"] == "RUNNING"
|
||
run_id = run_data["run_id"]
|
||
|
||
result = _poll_scenario(app_client, run_id)
|
||
assert result["status"] == "OK", f"Scenario failed: {result}"
|
||
assert result["current_step"] == 2
|
||
assert result["total_steps"] == 2
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_scenario_404_unknown_id(self, app_client):
|
||
resp = _run_scenario(app_client, 99999)
|
||
assert resp.status_code == 404
|
||
|
||
def test_scenario_lock_prevents_second_run(self, app_client):
|
||
"""Запуск во время выполнения -> 409."""
|
||
steps = [
|
||
{"service_id": 1, "operation": "create", "params": {}, "output": "lock1"},
|
||
{"service_id": 1, "operation": "delete", "instance_ref": "lock1", "params": {}},
|
||
]
|
||
name = f"test-lock-{int(time.time())}"
|
||
created = _create_scenario(app_client, name, steps)
|
||
|
||
# Первый запуск
|
||
resp1 = _run_scenario(app_client, created["id"])
|
||
assert resp1.status_code == 202
|
||
|
||
# Второй запуск — должен получить 409
|
||
resp2 = _run_scenario(app_client, created["id"])
|
||
assert resp2.status_code == 409, f"Expected 409, got {resp2.status_code}: {resp2.get_json()}"
|
||
|
||
# Ждём завершения первого
|
||
run_id = resp1.get_json()["run_id"]
|
||
_poll_scenario(app_client, run_id)
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_scenario_can_run_after_completion(self, app_client):
|
||
"""После завершения можно запустить снова."""
|
||
steps = [
|
||
{"service_id": 1, "operation": "create", "params": {}, "output": "again"},
|
||
{"service_id": 1, "operation": "delete", "instance_ref": "again", "params": {}},
|
||
]
|
||
name = f"test-again-{int(time.time())}"
|
||
created = _create_scenario(app_client, name, steps)
|
||
|
||
# Первый прогон
|
||
resp1 = _run_scenario(app_client, created["id"])
|
||
run1_id = resp1.get_json()["run_id"]
|
||
result1 = _poll_scenario(app_client, run1_id)
|
||
assert result1["status"] == "OK"
|
||
|
||
# Второй прогон — должен работать
|
||
resp2 = _run_scenario(app_client, created["id"])
|
||
assert resp2.status_code == 202, f"Second run failed: {resp2.get_json()}"
|
||
run2_id = resp2.get_json()["run_id"]
|
||
result2 = _poll_scenario(app_client, run2_id)
|
||
assert result2["status"] == "OK"
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_scenario_status_endpoint(self, app_client):
|
||
"""GET /api/scenario/status возвращает список."""
|
||
resp = app_client.get("/api/scenario/status")
|
||
assert resp.status_code == 200
|
||
data = resp.get_json()
|
||
assert isinstance(data, list)
|
||
|
||
|
||
class TestScenarioCRUD:
|
||
"""CRUD определения сценариев."""
|
||
|
||
def test_create_and_get_definition(self, app_client):
|
||
name = f"test-crud-{int(time.time())}"
|
||
steps = [{"service_id": 1, "operation": "create", "params": {}}]
|
||
created = _create_scenario(app_client, name, steps)
|
||
assert "id" in created
|
||
|
||
resp = app_client.get(f"/api/scenario/definitions/{created['id']}")
|
||
data = resp.get_json()
|
||
assert data["name"] == name
|
||
assert data["version"] == 1
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_update_increments_version(self, app_client):
|
||
name = f"test-ver-{int(time.time())}"
|
||
created = _create_scenario(app_client, name,
|
||
[{"service_id": 1, "operation": "create", "params": {}}])
|
||
|
||
resp = app_client.put(f"/api/scenario/definitions/{created['id']}",
|
||
json={"name": name, "steps": [
|
||
{"service_id": 1, "operation": "create", "params": {}},
|
||
{"service_id": 1, "operation": "delete", "params": {}}
|
||
], "version": 1},
|
||
headers={"Content-Type": "application/json"})
|
||
data = resp.get_json()
|
||
assert data.get("ok"), f"Update failed: {data}"
|
||
assert data["version"] == 2
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_version_conflict_returns_409(self, app_client):
|
||
name = f"test-conflict-{int(time.time())}"
|
||
created = _create_scenario(app_client, name,
|
||
[{"service_id": 1, "operation": "create", "params": {}}])
|
||
|
||
# Обновляем с version=1 -> OK
|
||
app_client.put(f"/api/scenario/definitions/{created['id']}",
|
||
json={"name": name, "steps": [
|
||
{"service_id": 1, "operation": "create", "params": {}}
|
||
], "version": 1},
|
||
headers={"Content-Type": "application/json"})
|
||
|
||
# Ещё раз с version=1 -> conflict
|
||
resp = app_client.put(f"/api/scenario/definitions/{created['id']}",
|
||
json={"name": name, "steps": [
|
||
{"service_id": 1, "operation": "create", "params": {}}
|
||
], "version": 1},
|
||
headers={"Content-Type": "application/json"})
|
||
data = resp.get_json()
|
||
assert data.get("conflict"), f"Expected conflict, got: {data}"
|
||
|
||
_delete_scenario(app_client, created["id"])
|
||
|
||
def test_soft_delete(self, app_client):
|
||
name = f"test-del-{int(time.time())}"
|
||
created = _create_scenario(app_client, name,
|
||
[{"service_id": 1, "operation": "create", "params": {}}])
|
||
|
||
resp = app_client.delete(f"/api/scenario/definitions/{created['id']}")
|
||
assert resp.status_code == 200
|
||
assert resp.get_json().get("ok")
|