(feat) change directory structure and add test cases

app/base_types.py

@@ -0,0 +1,17 @@
+from enum import StrEnum
+from pydantic import BaseModel
+
+
+class NotificationType(StrEnum):
+    """Notification Types"""
+
+    INFO = "Info"
+    WARNING = "Warning"
+
+
+class Notification(BaseModel):
+    """Notification"""
+
+    Type: NotificationType
+    Name: str
+    Description: str

app/main.py

@@ -0,0 +1,37 @@
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from .base_types import Notification, NotificationType
+import logging
+
+app = FastAPI(
+    title="Cloud Accelerator Coding Challenge",
+    description="Simple Notification App for the Coding Challenge.",
+    version="0.0.1",
+    terms_of_service="TOS",
+    contact={
+        "name": "Nicolas Sebastian Schuler",
+        "url": "https://n-schuler.dev",
+        "email": "mail@n-schuler.dev",
+    },
+    license_info={"name": "MIT", "identifier": "MIT"},
+)
+
+
+def log_notification(ntfy: Notification):
+    """Log Notification."""
+    logging.warning(ntfy)
+
+
+@app.post("/notification", status_code=204)
+async def notification(ntfy: Notification, background_tasks: BackgroundTasks):
+    """Forward Notification
+
+    * Forwards Notification if `Type: Warning`.
+    """
+    match ntfy.Type:
+        case NotificationType.INFO:
+            pass
+        case NotificationType.WARNING:
+            background_tasks.add_task(log_notification, ntfy)
+        case _:
+            # Already catched by Pydantic
+            raise HTTPException(status_code=400, detail="Malformed Request Body")

app/test_main.py

@@ -0,0 +1,69 @@
+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:
+            print(req)
+            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 not response.status_code == 204, description