Verify Pyctuator can register when SBA requires basic-auth

Add tests for verifying Pyctuator's support for basic-auth registration
with Spring Boot Admin.

Solves #28

Author: michael.yak

tests/conftest.py

@@ -1,4 +1,5 @@
 import logging
+import secrets
 import threading
 import time
 from abc import ABC, abstractmethod
@@ -8,9 +9,10 @@
 
 import pytest
 import requests
-from fastapi import FastAPI
+from fastapi import FastAPI, Depends, HTTPException
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
 from pydantic import BaseModel
-from starlette.requests import Request
+from starlette import status
 from uvicorn.config import Config
 from uvicorn.main import Server
 
@@ -65,9 +67,21 @@ def boot_admin_server(registration_tracker: RegistrationTrackerFixture) -> Gener
         docs_url="/api",
     )
 
-    # pylint: disable=unused-argument,unused-variable
+    security = HTTPBasic()
+
+    def get_current_username(credentials: HTTPBasicCredentials = Depends(security)) -> str:
+        correct_username = secrets.compare_digest(credentials.username, "moo")
+        correct_password = secrets.compare_digest(credentials.password, "haha")
+        if not (correct_username and correct_password):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Moo haha",
+            )
+        return credentials.username
+
+    # pylint: disable=unused-variable
     @boot_admin_app.post("/register", tags=["admin-server"])
-    def register(request: Request, registration: RegistrationRequest) -> Dict[str, str]:
+    def register(registration: RegistrationRequest) -> Dict[str, str]:
         logging.debug("Got registration post %s, %d registrations since %s",
                       registration, registration_tracker.count, registration_tracker.start_time)
         registration_tracker.registration = registration
@@ -76,6 +90,19 @@ def register(request: Request, registration: RegistrationRequest) -> Dict[str, s
             registration_tracker.start_time = registration.metadata["startup"]
         return {"id": "JB007"}
 
+    # pylint: disable=unused-variable
+    @boot_admin_app.post("/register-with-basic-auth", tags=["admin-server"])
+    def register_with_basic_auth(
+            registration: RegistrationRequest,
+            username: str = Depends(get_current_username)) -> Dict[str, str]:
+        logging.debug("Got registration post %s from %s, %d registrations since %s",
+                      registration, username, registration_tracker.count, registration_tracker.start_time)
+        registration_tracker.registration = registration
+        registration_tracker.count += 1
+        if registration_tracker.start_time is None:
+            registration_tracker.start_time = registration.metadata["startup"]
+        return {"id": "JB007"}
+
     # pylint: disable=unused-argument,unused-variable
     @boot_admin_app.delete("/register/{registration_id}", tags=["admin-server"])
     def deregister(registration_id: str) -> None:

tests/test_spring_boot_admin_registration.py

@@ -0,0 +1,86 @@
+from datetime import datetime
+from typing import Optional, Any
+
+import pytest
+
+from pyctuator.auth import Auth, BasicAuth
+from pyctuator.impl.spring_boot_admin_registration import BootAdminRegistrationHandler
+from tests.conftest import RegistrationTrackerFixture
+
+
+@pytest.mark.usefixtures("boot_admin_server")
+def test_registration_no_auth(registration_tracker: RegistrationTrackerFixture) -> None:
+    registration_handler = get_registration_handler("http://localhost:8001/register", None)
+
+    try:
+        registration_handler.start()
+        assert registration_tracker.count == 1
+
+    finally:
+        registration_handler.stop()
+
+
+@pytest.mark.usefixtures("boot_admin_server")
+def test_registration_basic_auth_no_creds(registration_tracker: RegistrationTrackerFixture, caplog: Any) -> None:
+    registration_handler = get_registration_handler("http://localhost:8001/register-with-basic-auth", None)
+
+    try:
+        registration_handler.start()
+        assert registration_tracker.count == 0
+
+        error_message = "Failed registering with boot-admin, got %s - %s"
+        assert error_message in [record.msg for record in caplog.records]
+
+        error_args = (401, b'{"detail":"Not authenticated"}')
+        assert error_args in [record.args for record in caplog.records if record.msg == error_message]
+
+    finally:
+        registration_handler.stop()
+
+
+@pytest.mark.usefixtures("boot_admin_server")
+def test_registration_basic_auth_bad_creds(registration_tracker: RegistrationTrackerFixture, caplog: Any) -> None:
+    registration_handler = get_registration_handler(
+        "http://localhost:8001/register-with-basic-auth",
+        BasicAuth("kuki", "puki")
+    )
+
+    try:
+        registration_handler.start()
+        assert registration_tracker.count == 0
+
+        error_message = "Failed registering with boot-admin, got %s - %s"
+        assert error_message in [record.msg for record in caplog.records]
+
+        error_args = (401, b'{"detail":"Moo haha"}')
+        assert error_args in [record.args for record in caplog.records if record.msg == error_message]
+
+    finally:
+        registration_handler.stop()
+
+
+@pytest.mark.usefixtures("boot_admin_server")
+def test_registration_basic_auth(registration_tracker: RegistrationTrackerFixture) -> None:
+    registration_handler = get_registration_handler(
+        "http://localhost:8001/register-with-basic-auth",
+        BasicAuth("moo", "haha")
+    )
+
+    try:
+        registration_handler.start()
+        assert registration_tracker.count == 1
+
+    finally:
+        registration_handler.stop()
+
+
+def get_registration_handler(registration_url: str, registration_auth: Optional[Auth]) -> BootAdminRegistrationHandler:
+    return BootAdminRegistrationHandler(
+        registration_url=registration_url,
+        registration_auth=registration_auth,
+        application_name="noauth",
+        pyctuator_base_url="http://whatever/pyctuator",
+        start_time=datetime.now(),
+        service_url="http://whatever/service",
+        registration_interval_sec=100
+    )