71 lines
2.0 KiB
Python
71 lines
2.0 KiB
Python
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from .main import app
|
|
|
|
|
|
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():
|
|
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
|
|
|
|
|
|
@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
|
|
)
|