Add webserver tests (#25)

f2ap/config.py

@@ -50,56 +50,61 @@ def __init__(
         self.following = followings if followings is not None else []
         self.attachments = attachments
 
-        avatar_header_preferred_mimes = [
-            "image/png",
-            "image/jpeg",
-            "image/gif",
-            "image/webp",
-        ]
+        with open(public_key, "r") as file:
+            self.public_key = file.read()
+        with open(private_key, "r") as file:
+            self.private_key = file.read()
 
         for what, url in [("avatar", avatar), ("header", header)]:
             if url is None:
                 continue
 
             try:
-                r = requests.head(url)
-                r.raise_for_status()
-                content_type = r.headers.get("Content-Type")
-
-                if content_type is None:
-                    logging.warning(
-                        f"Could not determine the type of the {what} at {url}:"
-                        f" server does not provide a Content-Type."
-                        f" It may not appear on social applications."
-                    )
-                elif not content_type.startswith("image/"):
-                    logging.warning(
-                        f"The {what} at {url} is reported with MIME type {content_type},"
-                        f" which does not match an image. It may not appear on social applications."
-                    )
-                elif content_type not in avatar_header_preferred_mimes:
-                    logging.warning(
-                        f"The {what} at {url} is reported with MIME type {content_type},"
-                        f" which is unusual image type for the Web"
-                        f" (usual images types are {', '.join(avatar_header_preferred_mimes)})."
-                        f" It may not appear on social applications."
-                    )
-
+                ct = self.get_attachment_type(what, url)
                 if what == "avatar":
-                    self.avatar_type = content_type
+                    self.avatar_type = ct
                 else:
-                    self.header_type = content_type
+                    self.header_type = ct
 
             except requests.HTTPError as e:
                 logging.warning(
                     f"Could not load the {what} metadata at {url}: {e}."
                     f" It may not appear correctly on social applications."
                 )
 
-        with open(public_key, "r") as file:
-            self.public_key = file.read()
-        with open(private_key, "r") as file:
-            self.private_key = file.read()
+    @staticmethod
+    def get_attachment_type(what: str, url: str) -> str:
+        avatar_header_preferred_mimes = [
+            "image/png",
+            "image/jpeg",
+            "image/gif",
+            "image/webp",
+        ]
+
+        r = requests.head(url)
+        r.raise_for_status()
+        content_type = r.headers.get("Content-Type")
+
+        if content_type is None:
+            logging.warning(
+                f"Could not determine the type of the {what} at {url}:"
+                f" server does not provide a Content-Type."
+                f" It may not appear on social applications."
+            )
+        elif not content_type.startswith("image/"):
+            logging.warning(
+                f"The {what} at {url} is reported with MIME type {content_type},"
+                f" which does not match an image. It may not appear on social applications."
+            )
+        elif content_type not in avatar_header_preferred_mimes:
+            logging.warning(
+                f"The {what} at {url} is reported with MIME type {content_type},"
+                f" which is unusual image type for the Web"
+                f" (usual images types are {', '.join(avatar_header_preferred_mimes)})."
+                f" It may not appear on social applications."
+            )
+
+        return content_type
 
     @property
     def id(self) -> str:

f2ap/webserver.py

@@ -2,7 +2,6 @@
 import threading
 
 import uvicorn
-import requests
 import json
 import base64
 import hashlib
@@ -11,10 +10,10 @@
 from typing import Union, Any, Optional
 from fastapi import FastAPI, BackgroundTasks
 from fastapi import Request
-from fastapi.responses import Response, JSONResponse, RedirectResponse
+from fastapi.responses import Response, JSONResponse
 from pydantic import BaseModel
 
-from . import postie, signature, activitypub
+from . import postie, activitypub
 from .config import Configuration
 from .data import Database
 from .exceptions import HttpError
@@ -90,16 +89,14 @@ def f(coroutine):
     return decorator
 
 
-def start_server(
-    config: Configuration, port: int, log_level: str, skip_following: bool = False
-):
+def get_server(config: Configuration, skip_following: bool = False):
     app = FastAPI(docs_url=None)
     app.activitypub = get_activitypub_decorator(app)
     db = Database(config)
-    start_server.following = None
+    get_server.following = None
 
     if skip_following:
-        start_server.following = []
+        get_server.following = []
         logging.debug("Following is disabled.")
 
     @app.middleware("http")
@@ -109,8 +106,8 @@ async def on_request(request: Request, call_next):
         )
 
         # If the server has just started, follow the users specified in the configuration.
-        if not skip_following and start_server.following is None:
-            start_server.following = []
+        if not skip_following and get_server.following is None:
+            get_server.following = []
             follow_task = FollowThread(config, config.actor.following)
             follow_task.start()
 
@@ -129,8 +126,8 @@ async def on_request(request: Request, call_next):
 
     @app.on_event("shutdown")
     async def on_stop():
-        if start_server.following is not None:
-            activitypub.unfollow_users(config, start_server.following)
+        if get_server.following is not None:
+            activitypub.unfollow_users(config, get_server.following)
 
     @app.get("/.well-known/webfinger")
     async def webfinger(resource: Union[str, None]):
@@ -169,7 +166,7 @@ async def get_following(username, page: Optional[int] = 0):
             return Response(status_code=404)
 
         following = []
-        for _, account in start_server.following:
+        for _, account in get_server.following:
             following.append(account)
 
         return respond(
@@ -228,7 +225,7 @@ async def post_inbox(
                 db,
                 dict(request.headers),
                 inbox,
-                lambda i, a: start_server.following.append((i, a)),
+                lambda i, a: get_server.following.append((i, a)),
             )
         except HttpError as e:
             return Response(e.body, status_code=e.status_code)
@@ -246,8 +243,14 @@ async def post_inbox(
     async def get_messages(uuid: UUID):
         return respond(db.get_message(uuid))
 
+    return app
+
+
+def start_server(
+    config: Configuration, port: int, log_level: str, skip_following: bool = False
+):
     uvicorn.run(
-        app,
+        get_server(config, skip_following),
         host="0.0.0.0",
         port=port,
         log_level=log_level.lower(),