119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
"""IRM API с подменённым пулом БД (без PostgreSQL)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from starlette.requests import Request
|
||
|
||
from onguard24.deps import get_pool
|
||
from onguard24.main import app
|
||
|
||
from tests.irm_db_fake import IrmFakePool
|
||
|
||
|
||
@pytest.fixture
|
||
def irm_client() -> tuple[TestClient, IrmFakePool]:
|
||
pool = IrmFakePool()
|
||
|
||
def override_get_pool(_request: Request):
|
||
return pool
|
||
|
||
app.dependency_overrides[get_pool] = override_get_pool
|
||
with TestClient(app) as client:
|
||
yield client, pool
|
||
app.dependency_overrides.pop(get_pool, None)
|
||
|
||
|
||
def test_irm_incident_crud_and_tasks(irm_client: tuple[TestClient, IrmFakePool]) -> None:
|
||
client, _pool = irm_client
|
||
r = client.post(
|
||
"/api/v1/modules/incidents/",
|
||
json={"title": "Сбой API", "status": "open", "severity": "critical"},
|
||
)
|
||
assert r.status_code == 201
|
||
iid = r.json()["id"]
|
||
assert r.json()["source"] == "manual"
|
||
|
||
r = client.get(f"/api/v1/modules/incidents/{iid}")
|
||
assert r.status_code == 200
|
||
assert r.json()["title"] == "Сбой API"
|
||
|
||
r = client.patch(f"/api/v1/modules/incidents/{iid}", json={"status": "resolved"})
|
||
assert r.status_code == 200
|
||
assert r.json()["status"] == "resolved"
|
||
|
||
r = client.post(
|
||
"/api/v1/modules/tasks/",
|
||
json={"title": "Разбор логов", "incident_id": iid},
|
||
)
|
||
assert r.status_code == 201
|
||
tid = r.json()["id"]
|
||
|
||
r = client.get(f"/api/v1/modules/incidents/{iid}/tasks")
|
||
assert r.status_code == 200
|
||
assert len(r.json()["items"]) == 1
|
||
assert r.json()["items"][0]["id"] == tid
|
||
|
||
r = client.get(f"/api/v1/modules/tasks/{tid}")
|
||
assert r.status_code == 200
|
||
assert r.json()["status"] == "open"
|
||
|
||
r = client.patch(f"/api/v1/modules/tasks/{tid}", json={"status": "done"})
|
||
assert r.status_code == 200
|
||
assert r.json()["status"] == "done"
|
||
|
||
|
||
def test_irm_task_bad_incident(irm_client: tuple[TestClient, IrmFakePool]) -> None:
|
||
client, _ = irm_client
|
||
import uuid
|
||
|
||
r = client.post(
|
||
"/api/v1/modules/tasks/",
|
||
json={"title": "x", "incident_id": str(uuid.uuid4())},
|
||
)
|
||
assert r.status_code == 400
|
||
assert r.json()["detail"] == "incident not found"
|
||
|
||
|
||
def test_irm_incident_tasks_unknown(irm_client: tuple[TestClient, IrmFakePool]) -> None:
|
||
client, _ = irm_client
|
||
import uuid
|
||
|
||
rid = str(uuid.uuid4())
|
||
r = client.get(f"/api/v1/modules/incidents/{rid}/tasks")
|
||
assert r.status_code == 404
|
||
|
||
|
||
def test_irm_patch_validation(irm_client: tuple[TestClient, IrmFakePool]) -> None:
|
||
client, _ = irm_client
|
||
r = client.post("/api/v1/modules/incidents/", json={"title": "t"})
|
||
iid = r.json()["id"]
|
||
r = client.patch(f"/api/v1/modules/incidents/{iid}", json={})
|
||
assert r.status_code == 400
|
||
|
||
|
||
def test_irm_escalations_crud(irm_client: tuple[TestClient, IrmFakePool]) -> None:
|
||
client, _ = irm_client
|
||
r = client.post(
|
||
"/api/v1/modules/escalations/",
|
||
json={"name": "L1", "enabled": True, "steps": [{"after_min": 5, "channel": "slack"}]},
|
||
)
|
||
assert r.status_code == 201
|
||
pid = r.json()["id"]
|
||
assert r.json()["steps"][0]["channel"] == "slack"
|
||
|
||
r = client.get(f"/api/v1/modules/escalations/{pid}")
|
||
assert r.status_code == 200
|
||
assert r.json()["name"] == "L1"
|
||
|
||
r = client.patch(f"/api/v1/modules/escalations/{pid}", json={"enabled": False})
|
||
assert r.status_code == 200
|
||
assert r.json()["enabled"] is False
|
||
|
||
r = client.delete(f"/api/v1/modules/escalations/{pid}")
|
||
assert r.status_code == 204
|
||
|
||
r = client.get(f"/api/v1/modules/escalations/{pid}")
|
||
assert r.status_code == 404
|