Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add webserver tests #25

Merged
merged 1 commit into from
Jan 25, 2023
Merged

Add webserver tests #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 40 additions & 35 deletions f2ap/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
33 changes: 18 additions & 15 deletions f2ap/webserver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
import threading

import uvicorn
import requests
import json
import base64
import hashlib
Expand All @@ -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
Expand Down Expand Up @@ -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")
Expand All @@ -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()

Expand All @@ -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]):
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
Expand All @@ -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(),
Expand Down
Loading