91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
import pytest
|
|
import logging
|
|
import json
|
|
from httpx import ASGITransport, AsyncClient
|
|
from pydantic import ValidationError
|
|
|
|
from .main import app
|
|
from .base_types import NotificationType, Notification
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
forward_notification_success_req = {
|
|
"warning": {
|
|
"Type": "Warning",
|
|
"Name": "Backup Failure",
|
|
"Description": "The backup failed due to a database problem",
|
|
},
|
|
"info": {
|
|
"Type": "Info",
|
|
"Name": "Quota Exceeded",
|
|
"Description": "Compute Quota exceeded",
|
|
},
|
|
}
|
|
|
|
forward_notification_fail_req = {
|
|
"type unknown": {
|
|
"Type": "Garbage",
|
|
"Name": "Quota Exceeded",
|
|
"Description": "Compute Quota exceeded",
|
|
},
|
|
"type lower case": {
|
|
"Type": "warning",
|
|
"Name": "Backup Failure",
|
|
"Description": "The backup failed due to a database problem",
|
|
},
|
|
"key lower cases": {
|
|
"type": "warning",
|
|
"name": "Backup Failure",
|
|
"description": "The backup failed due to a database problem",
|
|
},
|
|
"type missing": {
|
|
"Name": "Quota Exceeded",
|
|
"Description": "Compute Quota exceeded",
|
|
},
|
|
"name missing": {
|
|
"Type": "Info",
|
|
"Description": "Compute Quota exceeded",
|
|
},
|
|
"description missing": {
|
|
"Type": "Info",
|
|
"Name": "Quota Exceeded",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forward_notification_success(caplog):
|
|
for description, req in forward_notification_success_req.items():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
response = await ac.post("/notification", json=req)
|
|
assert response.status_code == 204, description
|
|
|
|
match req["Type"]:
|
|
case NotificationType.INFO:
|
|
pass
|
|
case NotificationType.WARNING:
|
|
# Check forwarding to logger
|
|
try:
|
|
res = Notification.model_validate_json(
|
|
json_data=json.dumps(req)
|
|
)
|
|
except ValidationError:
|
|
raise Exception(
|
|
"Model could not be serialized. Check dummy data"
|
|
)
|
|
assert str(res) in caplog.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_forward_notification_failure():
|
|
for description, req in forward_notification_fail_req.items():
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
response = await ac.post("/notification", json=req)
|
|
assert response.status_code >= 400 and response.status_code < 500, (
|
|
description
|
|
)
|