From a2cfc36dab8adf440704a17cc066db219143207a Mon Sep 17 00:00:00 2001 From: Thomas Sileo Date: Wed, 22 Jun 2022 22:17:07 +0200 Subject: [PATCH] More tests --- app/activitypub.py | 1 + tests/factories.py | 66 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_public.py | 37 ++++++++++++++++++++----- 3 files changed, 97 insertions(+), 7 deletions(-) diff --git a/app/activitypub.py b/app/activitypub.py index 4226056..6869a0c 100644 --- a/app/activitypub.py +++ b/app/activitypub.py @@ -6,6 +6,7 @@ from typing import Any import httpx from app import config +from app.config import AP_CONTENT_TYPE # noqa: F401 from app.httpsig import auth from app.key import get_pubkey_as_pem diff --git a/tests/factories.py b/tests/factories.py index f0aff49..2742b20 100644 --- a/tests/factories.py +++ b/tests/factories.py @@ -1,7 +1,9 @@ +from urllib.parse import urlparse from uuid import uuid4 import factory # type: ignore from Crypto.PublicKey import RSA +from dateutil.parser import isoparse from sqlalchemy import orm from app import activitypub as ap @@ -10,6 +12,7 @@ from app import models from app.actor import RemoteActor from app.ap_object import RemoteObject from app.database import engine +from app.database import now _Session = orm.scoped_session(orm.sessionmaker(bind=engine)) @@ -47,6 +50,36 @@ def build_accept_activity( } +def build_note_object( + from_remote_actor: actor.RemoteActor, + outbox_public_id: str | None = None, + content: str = "Hello", + to: list[str] = None, + cc: list[str] = None, + tags: list[ap.RawObject] = None, +) -> ap.RawObject: + published = now().replace(microsecond=0).isoformat().replace("+00:00", "Z") + context = from_remote_actor.ap_id + "/ctx/" + uuid4().hex + note_id = outbox_public_id or uuid4().hex + return { + "@context": ap.AS_CTX, + "type": "Note", + "id": from_remote_actor.ap_id + "/note/" + note_id, + "attributedTo": from_remote_actor.ap_id, + "content": content, + "to": to or [ap.AS_PUBLIC], + "cc": cc or [], + "published": published, + "context": context, + "conversation": context, + "url": from_remote_actor.ap_id + "/note/" + note_id, + "tag": tags or [], + "summary": None, + "inReplyTo": None, + "sensitive": False, + } + + class BaseModelMeta: sqlalchemy_session = _Session sqlalchemy_session_persistence = "commit" @@ -138,3 +171,36 @@ class OutgoingActivityFactory(factory.alchemy.SQLAlchemyModelFactory): # recipient # outbox_object_id + + +class InboxObjectFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseModelMeta): + model = models.InboxObject + + @classmethod + def from_remote_object( + cls, + ro: RemoteObject, + actor: models.Actor, + relates_to_inbox_object_id: int | None = None, + relates_to_outbox_object_id: int | None = None, + ): + ap_published_at = now() + if "published" in ro.ap_object: + ap_published_at = isoparse(ro.ap_object["published"]) + return cls( + server=urlparse(ro.ap_id).netloc, + actor_id=actor.id, + ap_actor_id=actor.ap_id, + ap_type=ro.ap_type, + ap_id=ro.ap_id, + ap_context=ro.context, + ap_published_at=ap_published_at, + ap_object=ro.ap_object, + visibility=ro.visibility, + relates_to_inbox_object_id=relates_to_inbox_object_id, + relates_to_outbox_object_id=relates_to_outbox_object_id, + activity_object_ap_id=ro.activity_object_ap_id, + # Hide replies from the stream + is_hidden_from_stream=True if ro.in_reply_to else False, + ) diff --git a/tests/test_public.py b/tests/test_public.py index 4f6a150..0ea60c8 100644 --- a/tests/test_public.py +++ b/tests/test_public.py @@ -1,6 +1,8 @@ import pytest from fastapi.testclient import TestClient +from app import activitypub as ap +from app.actor import LOCAL_ACTOR from app.database import Session _ACCEPTED_AP_HEADERS = [ @@ -11,20 +13,41 @@ _ACCEPTED_AP_HEADERS = [ ] -@pytest.mark.anyio -def test_index(db: Session, client: TestClient): +def test_index__html(db: Session, client: TestClient): response = client.get("/") assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") @pytest.mark.parametrize("accept", _ACCEPTED_AP_HEADERS) -def test__ap_version(client, db, accept: str) -> None: - response = client.get("/followers", headers={"Accept": accept}) +def test_index__ap(db: Session, client: TestClient, accept: str): + response = client.get("/", headers={"Accept": accept}) assert response.status_code == 200 - assert response.headers["content-type"] == "application/activity+json" + assert response.headers["content-type"] == ap.AP_CONTENT_TYPE + assert response.json() == LOCAL_ACTOR.ap_actor + + +def test_followers__ap(client, db) -> None: + response = client.get("/followers", headers={"Accept": ap.AP_CONTENT_TYPE}) + assert response.status_code == 200 + assert response.headers["content-type"] == ap.AP_CONTENT_TYPE assert response.json()["id"].endswith("/followers") -def test__html(client, db) -> None: - response = client.get("/followers", headers={"Accept": "application/activity+json"}) +def test_followers__html(client, db) -> None: + response = client.get("/followers") assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html") + + +def test_following__ap(client, db) -> None: + response = client.get("/following", headers={"Accept": ap.AP_CONTENT_TYPE}) + assert response.status_code == 200 + assert response.headers["content-type"] == ap.AP_CONTENT_TYPE + assert response.json()["id"].endswith("/following") + + +def test_following__html(client, db) -> None: + response = client.get("/following") + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/html")