microblog.pub/tests/utils.py

251 lines
6.8 KiB
Python
Raw Normal View History

2022-07-26 13:26:34 -05:00
import asyncio
2022-06-22 13:11:22 -05:00
from contextlib import contextmanager
2022-06-22 14:15:07 -05:00
from typing import Any
2022-07-26 13:26:34 -05:00
from uuid import uuid4
2022-06-22 13:11:22 -05:00
import fastapi
2022-07-26 13:26:34 -05:00
import httpx
import respx
2022-06-22 13:11:22 -05:00
from app import activitypub as ap
2022-06-22 13:11:22 -05:00
from app import actor
from app import httpsig
2022-07-26 13:26:34 -05:00
from app import models
from app.actor import LOCAL_ACTOR
from app.ap_object import RemoteObject
2022-06-22 13:11:22 -05:00
from app.config import session_serializer
from app.database import AsyncSession
2022-07-26 13:26:34 -05:00
from app.database import async_session
from app.incoming_activities import fetch_next_incoming_activity
from app.incoming_activities import process_next_incoming_activity
2022-06-22 13:11:22 -05:00
from app.main import app
2022-07-26 13:26:34 -05:00
from tests import factories
2022-06-22 13:11:22 -05:00
@contextmanager
def mock_httpsig_checker(
ra: actor.RemoteActor,
has_valid_signature: bool = True,
is_ap_actor_gone: bool = False,
):
2022-06-22 13:11:22 -05:00
async def httpsig_checker(
request: fastapi.Request,
) -> httpsig.HTTPSigInfo:
return httpsig.HTTPSigInfo(
has_valid_signature=has_valid_signature,
2022-06-22 13:11:22 -05:00
signed_by_ap_actor_id=ra.ap_id,
is_ap_actor_gone=is_ap_actor_gone,
2022-06-22 13:11:22 -05:00
)
app.dependency_overrides[httpsig.httpsig_checker] = httpsig_checker
try:
yield
finally:
del app.dependency_overrides[httpsig.httpsig_checker]
2022-06-22 14:15:07 -05:00
def generate_admin_session_cookies() -> dict[str, Any]:
2022-06-22 13:11:22 -05:00
return {"session": session_serializer.dumps({"is_logged_in": True})}
2022-07-26 13:26:34 -05:00
2022-08-16 15:15:05 -05:00
def setup_remote_actor(
respx_mock: respx.MockRouter,
base_url="https://example.com",
also_known_as=None,
) -> actor.RemoteActor:
2022-07-26 13:26:34 -05:00
ra = factories.RemoteActorFactory(
2022-08-16 15:15:05 -05:00
base_url=base_url,
2022-07-26 13:26:34 -05:00
username="toto",
public_key="pk",
2022-08-16 15:15:05 -05:00
also_known_as=also_known_as if also_known_as else [],
2022-07-26 13:26:34 -05:00
)
respx_mock.get(ra.ap_id + "/outbox").mock(
return_value=httpx.Response(
200,
json={
"@context": ap.AS_EXTENDED_CTX,
"id": f"{ra.ap_id}/outbox",
"type": "OrderedCollection",
"totalItems": 0,
"orderedItems": [],
},
)
)
2022-07-26 13:26:34 -05:00
respx_mock.get(ra.ap_id).mock(return_value=httpx.Response(200, json=ra.ap_actor))
return ra
def setup_remote_actor_as_follower(ra: actor.RemoteActor) -> models.Follower:
actor = factories.ActorFactory.from_remote_actor(ra)
follow_id = uuid4().hex
follow_from_inbox = RemoteObject(
factories.build_follow_activity(
from_remote_actor=ra,
for_remote_actor=LOCAL_ACTOR,
outbox_public_id=follow_id,
),
ra,
)
inbox_object = factories.InboxObjectFactory.from_remote_object(
follow_from_inbox, actor
)
follower = factories.FollowerFactory(
inbox_object_id=inbox_object.id,
actor_id=actor.id,
ap_actor_id=actor.ap_id,
)
return follower
2022-08-16 15:15:05 -05:00
def setup_remote_actor_as_following(ra: actor.RemoteActor) -> models.Following:
actor = factories.ActorFactory.from_remote_actor(ra)
follow_id = uuid4().hex
follow_from_outbox = RemoteObject(
factories.build_follow_activity(
from_remote_actor=LOCAL_ACTOR,
for_remote_actor=ra,
outbox_public_id=follow_id,
),
LOCAL_ACTOR,
)
outbox_object = factories.OutboxObjectFactory.from_remote_object(
follow_id, follow_from_outbox
)
following = factories.FollowingFactory(
outbox_object_id=outbox_object.id,
actor_id=actor.id,
ap_actor_id=actor.ap_id,
)
return following
def setup_remote_actor_as_following_and_follower(
ra: actor.RemoteActor,
) -> tuple[models.Following, models.Follower]:
actor = factories.ActorFactory.from_remote_actor(ra)
follow_id = uuid4().hex
follow_from_outbox = RemoteObject(
factories.build_follow_activity(
from_remote_actor=LOCAL_ACTOR,
for_remote_actor=ra,
outbox_public_id=follow_id,
),
LOCAL_ACTOR,
)
outbox_object = factories.OutboxObjectFactory.from_remote_object(
follow_id, follow_from_outbox
)
following = factories.FollowingFactory(
outbox_object_id=outbox_object.id,
actor_id=actor.id,
ap_actor_id=actor.ap_id,
)
follow_id = uuid4().hex
follow_from_inbox = RemoteObject(
factories.build_follow_activity(
from_remote_actor=ra,
for_remote_actor=LOCAL_ACTOR,
outbox_public_id=follow_id,
),
ra,
)
inbox_object = factories.InboxObjectFactory.from_remote_object(
follow_from_inbox, actor
)
follower = factories.FollowerFactory(
inbox_object_id=inbox_object.id,
actor_id=actor.id,
ap_actor_id=actor.ap_id,
)
return following, follower
2022-08-30 13:05:10 -05:00
def setup_outbox_note(
content: str = "Hello",
to: list[str] = None,
cc: list[str] = None,
tags: list[ap.RawObject] = None,
in_reply_to: str | None = None,
) -> models.OutboxObject:
note_id = uuid4().hex
note_from_outbox = RemoteObject(
factories.build_note_object(
from_remote_actor=LOCAL_ACTOR,
outbox_public_id=note_id,
content=content,
to=to,
cc=cc,
tags=tags,
in_reply_to=in_reply_to,
),
LOCAL_ACTOR,
)
return factories.OutboxObjectFactory.from_remote_object(note_id, note_from_outbox)
def setup_inbox_note(
actor: models.Actor,
content: str = "Hello",
to: list[str] = None,
cc: list[str] = None,
tags: list[ap.RawObject] = None,
in_reply_to: str | None = None,
) -> models.OutboxObject:
note_id = uuid4().hex
note_from_outbox = RemoteObject(
factories.build_note_object(
from_remote_actor=actor,
outbox_public_id=note_id,
content=content,
to=to,
cc=cc,
tags=tags,
in_reply_to=in_reply_to,
),
actor,
)
return factories.InboxObjectFactory.from_remote_object(note_from_outbox, actor)
2022-07-26 14:10:59 -05:00
def setup_inbox_delete(
actor: models.Actor, deleted_object_ap_id: str
) -> models.InboxObject:
follow_from_inbox = RemoteObject(
factories.build_delete_activity(
from_remote_actor=actor,
deleted_object_ap_id=deleted_object_ap_id,
),
actor,
)
inbox_object = factories.InboxObjectFactory.from_remote_object(
follow_from_inbox, actor
)
return inbox_object
2022-07-26 13:26:34 -05:00
def run_async(func, *args, **kwargs):
async def _func():
async with async_session() as db:
return await func(db, *args, **kwargs)
asyncio.run(_func())
async def _process_next_incoming_activity(db_session: AsyncSession) -> None:
next_activity = await fetch_next_incoming_activity(db_session)
assert next_activity
await process_next_incoming_activity(db_session, next_activity)
def run_process_next_incoming_activity() -> None:
run_async(_process_next_incoming_activity)