Added endpoints

Author: joshvote

pyproject.toml

@@ -4,6 +4,18 @@ line-length = 120
 [tool.pytest.ini_options]
 pythonpath = ["src/"]
 testpaths = "tests"
+asyncio_mode = "auto"
+markers = [
+    "APP_PORT: Sets the APP_PORT env variable variable at test startup",
+    "SERVER_URL: Sets the SERVER_URL env variable variable at test startup",
+    "MOUNT_POINT: Sets the MOUNT_POINT env variable variable at test startup",
+    "MAX_IDLE_DURATION_SECONDS: Sets the MAX_IDLE_DURATION_SECONDS env variable variable at test startup",
+    "MAX_DURATION_SECONDS: Sets the MAX_DURATION_SECONDS env variable variable at test startup",
+    "MAX_ACTIVE_ENDPOINTS: Sets the MAX_ACTIVE_ENDPOINTS env variable variable at test startup",
+    "MAX_ENDPOINT_NOTIFICATIONS: Sets the MAX_ENDPOINT_NOTIFICATIONS env variable variable at test startup",
+    "CLEANUP_FREQUENCY_SECONDS: Sets the CLEANUP_FREQUENCY_SECONDS env variable variable at test startup",
+]
+
 
 [tool.isort]
 profile = "black"
@@ -39,7 +51,7 @@ dependencies = [
 [project.optional-dependencies]
 server = ["aiohttp>=3.11.12,<4"]
 dev = ["bandit", "black", "coverage", "flake8", "flake8-bugbear", "mypy"]
-test = ["assertical", "pytest", "pytest-aiohttp", "freezegun"]
+test = ["assertical", "pytest", "pytest-asyncio", "pytest-aiohttp", "freezegun"]
 
 [tool.setuptools.package-data]
 "cactus_client_notifications" = ["py.typed", "**/*.xsd"]

src/cactus_client_notifications/server/handler.py

@@ -10,7 +10,10 @@
     CreateEndpointRequest,
     CreateEndpointResponse,
 )
-from cactus_client_notifications.server.endpoint_store import NotificationException
+from cactus_client_notifications.server.endpoint_store import (
+    NotificationException,
+    generate_collected_notification,
+)
 from cactus_client_notifications.server.settings import ServerSettings
 from cactus_client_notifications.server.shared import (
     APPKEY_NOTIFICATION_STORE,
@@ -86,6 +89,7 @@ async def post_manage_endpoint_list(request: web.Request) -> web.Response:
     try:
         endpoint_id = await request.app[APPKEY_NOTIFICATION_STORE].create_endpoint()
     except NotificationException as exc:
+        logger.error("Error creating endpoint", exc_info=exc)
         return web.Response(status=exc.status_code, text=str(exc))
 
     create_response = CreateEndpointResponse(
@@ -119,6 +123,7 @@ async def get_manage_endpoint(request: web.Request) -> web.Response:
     try:
         collected_notifications = await request.app[APPKEY_NOTIFICATION_STORE].collect_notifications(endpoint_id)
     except NotificationException as exc:
+        logger.error(f"Error updating config for {endpoint_id}", exc_info=exc)
         return web.Response(status=exc.status_code, text=str(exc))
 
     collect_response = CollectEndpointResponse(
@@ -159,6 +164,75 @@ async def put_manage_endpoint(request: web.Request) -> web.Response:
     try:
         await request.app[APPKEY_NOTIFICATION_STORE].update_endpoint(endpoint_id, enabled=configure_request.enabled)
     except NotificationException as exc:
+        logger.error(f"Error configuring {endpoint_id} with {configure_request}", exc_info=exc)
+        return web.Response(status=exc.status_code, text=str(exc))
+
+    return web.Response(status=http.HTTPStatus.NO_CONTENT)
+
+
+async def delete_manage_endpoint(request: web.Request) -> web.Response:
+    """Deletes an existing endpoint based on the endpoint_id in the path. All uncollected notifications will be lost.
+
+    Args:
+        request: An aiohttp.web.Request instance.
+
+    Returns:
+        aiohttp.web.Response: Encodes a CreateEndpointResponse on success
+
+        a 204 (NO_CONTENT) on success.
+        a 404 (NOT_FOUND) if the endpoint has been deleted or the endpoint_id is invalid
+    """
+
+    endpoint_id = request.match_info.get("endpoint_id")
+    if not endpoint_id:
+        return web.Response(status=http.HTTPStatus.BAD_REQUEST, text="endpoint_id couldn't be extracted from the path.")
+
+    logger.info(f"Deleting endpoint {endpoint_id} for {request.remote}")
+
+    try:
+        await request.app[APPKEY_NOTIFICATION_STORE].try_delete_endpoint(endpoint_id)
+    except NotificationException as exc:
+        logger.error(f"Error deleting {endpoint_id}", exc_info=exc)
         return web.Response(status=exc.status_code, text=str(exc))
 
     return web.Response(status=http.HTTPStatus.NO_CONTENT)
+
+
+async def webhook_endpoint(request: web.Request) -> web.Response:
+    """This is the endpoint that will handle ALL incoming 2030.5 notifications from the utility server. It will try
+    to report success for everything and log the contents of the incoming request.
+
+    Args:
+        request: An aiohttp.web.Request instance.
+
+    Returns:
+        aiohttp.web.Response: Encodes a CreateEndpointResponse on success
+
+        a 200 (OK) on success.
+        a 404 (NOT_FOUND) if the endpoint has been deleted or the endpoint_id is invalid
+        a 500 (INTERNAL_SERVER_ERROR) if the endpoint has been disabled
+        a 507 (INSUFFICIENT_STORAGE) if the endpoint has too many uncollected notifications
+    """
+
+    endpoint_id = request.match_info.get("endpoint_id")
+    if not endpoint_id:
+        return web.Response(status=http.HTTPStatus.NOT_FOUND)
+
+    try:
+        collected_notification = await generate_collected_notification(request)
+    except Exception as exc:
+        logger.error(f"Error parsing incoming webhook request for {endpoint_id} from {request.remote}", exc_info=exc)
+        return web.Response(status=http.HTTPStatus.BAD_REQUEST)
+
+    logger.info(
+        f"{collected_notification.method} notification ({len(collected_notification.body)}) at {endpoint_id}"
+        + f"from {request.remote}."
+    )
+
+    try:
+        await request.app[APPKEY_NOTIFICATION_STORE].add_notification(endpoint_id, collected_notification)
+    except NotificationException as exc:
+        logger.error(f"Error adding notification to {endpoint_id}", exc_info=exc)
+        return web.Response(status=exc.status_code, text=str(exc))
+
+    return web.Response(status=http.HTTPStatus.OK)

src/cactus_client_notifications/server/main.py

@@ -28,15 +28,17 @@ async def periodic_task(app: web.Application) -> None:
     store = app[shared.APPKEY_NOTIFICATION_STORE]
 
     while True:
+
+        # Sleep first - we don't need to initiate a cleanup immediately
+        await asyncio.sleep(server_settings.cleanup_frequency.total_seconds())
+
         try:
             await store.cleanup_expired_endpoints(
                 utc_now(), server_settings.max_endpoint_idle_duration, server_settings.max_endpoint_duration
             )
-        except Exception as e:
+        except Exception as exc:
             # Catch and log uncaught exceptions to prevent periodic task from hanging
-            logger.error(f"Uncaught exception in periodic task: {repr(e)}")
-
-        await asyncio.sleep(server_settings.cleanup_frequency.total_seconds())
+            logger.error("Uncaught exception in periodic task", exc_info=exc)
 
 
 async def setup_periodic_task(app: web.Application) -> AsyncGenerator:
@@ -62,11 +64,11 @@ def create_app() -> web.Application:
     ENV_APP_PORT = int(os.getenv("APP_PORT", 8080))
     ENV_PUBLIC_SERVER_URL = os.getenv("SERVER_URL", f"http://localhost:{ENV_APP_PORT}")
     ENV_MOUNT_POINT = os.getenv("MOUNT_POINT", "/")
-    ENV_MAX_IDLE_DURATION_SECONDS = int(os.getenv("MAX_IDLE_DURATION_SECONDS", 3600))
-    ENV_MAX_DURATION_SECONDS = int(os.getenv("MAX_DURATION_SECONDS", (3600 * 24 * 3) + 3600))
+    ENV_MAX_IDLE_DURATION_SECONDS = float(os.getenv("MAX_IDLE_DURATION_SECONDS", 3600))
+    ENV_MAX_DURATION_SECONDS = float(os.getenv("MAX_DURATION_SECONDS", (3600 * 24 * 3) + 3600))
     ENV_MAX_ACTIVE_ENDPOINTS = int(os.getenv("MAX_ACTIVE_ENDPOINTS", 1024))
     ENV_MAX_ENDPOINT_NOTIFICATIONS = int(os.getenv("MAX_ENDPOINT_NOTIFICATIONS", 100))
-    ENV_CLEANUP_FREQUENCY_SECONDS = int(os.getenv("CLEANUP_FREQUENCY_SECONDS", 120))
+    ENV_CLEANUP_FREQUENCY_SECONDS = float(os.getenv("CLEANUP_FREQUENCY_SECONDS", 120))
     server_settings = ServerSettings(
         port=ENV_APP_PORT,
         public_server_url=ENV_PUBLIC_SERVER_URL,
@@ -92,6 +94,8 @@ def create_app() -> web.Application:
     )
     app.router.add_route("GET", handler.path_join(mount, schema.URI_MANAGE_ENDPOINT), handler.get_manage_endpoint)
     app.router.add_route("PUT", handler.path_join(mount, schema.URI_MANAGE_ENDPOINT), handler.put_manage_endpoint)
+    app.router.add_route("DELETE", handler.path_join(mount, schema.URI_MANAGE_ENDPOINT), handler.delete_manage_endpoint)
+    app.router.add_route("*", handler.path_join(mount, schema.URI_ENDPOINT), handler.webhook_endpoint)
 
     # Start the periodic task
     app.cleanup_ctx.append(setup_periodic_task)

tests/conftest.py

@@ -0,0 +1,30 @@
+import os
+
+import pytest
+from aiohttp import ClientSession, ClientTimeout
+from assertical.fixtures.environment import environment_snapshot
+
+from cactus_client_notifications.server.main import create_app
+
+
+def marker_to_env(request: pytest.FixtureRequest, var_name: str) -> None:
+    marker = request.node.get_closest_marker(var_name)
+    if marker is not None:
+        os.environ[var_name] = str(marker.args[0])
+
+
+@pytest.fixture
+async def client_session(aiohttp_client, request: pytest.FixtureRequest):
+    with environment_snapshot():
+        marker_to_env(request, "APP_PORT")
+        marker_to_env(request, "SERVER_URL")
+        marker_to_env(request, "MOUNT_POINT")
+        marker_to_env(request, "MAX_IDLE_DURATION_SECONDS")
+        marker_to_env(request, "MAX_DURATION_SECONDS")
+        marker_to_env(request, "MAX_ACTIVE_ENDPOINTS")
+        marker_to_env(request, "MAX_ENDPOINT_NOTIFICATIONS")
+        marker_to_env(request, "CLEANUP_FREQUENCY_SECONDS")
+
+        async with await aiohttp_client(create_app()) as app:
+            async with ClientSession(base_url=app.make_url("/"), timeout=ClientTimeout(30)) as session:
+                yield session