microblog.pub/tests/factories.py

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
from app import actor
from app import models
from app.actor import RemoteActor
from app.ap_object import RemoteObject
from app.database import SessionLocal
from app.utils.datetime import now

_Session = orm.scoped_session(SessionLocal)


def generate_key() -> tuple[str, str]:
    k = RSA.generate(1024)
    return k.exportKey("PEM").decode(), k.publickey().exportKey("PEM").decode()


def build_follow_activity(
    from_remote_actor: actor.RemoteActor,
    for_remote_actor: actor.RemoteActor,
    outbox_public_id: str | None = None,
) -> ap.RawObject:
    return {
        "@context": ap.AS_CTX,
        "type": "Follow",
        "id": from_remote_actor.ap_id + "/follow/" + (outbox_public_id or uuid4().hex),
        "actor": from_remote_actor.ap_id,
        "object": for_remote_actor.ap_id,
    }


def build_delete_activity(
    from_remote_actor: actor.RemoteActor | models.Actor,
    deleted_object_ap_id: str,
    outbox_public_id: str | None = None,
) -> ap.RawObject:
    return {
        "@context": ap.AS_CTX,
        "type": "Delete",
        "id": (
            from_remote_actor.ap_id  # type: ignore
            + "/follow/"
            + (outbox_public_id or uuid4().hex)
        ),
        "actor": from_remote_actor.ap_id,
        "object": deleted_object_ap_id,
    }


def build_accept_activity(
    from_remote_actor: actor.RemoteActor,
    for_remote_object: RemoteObject,
    outbox_public_id: str | None = None,
) -> ap.RawObject:
    return {
        "@context": ap.AS_CTX,
        "type": "Accept",
        "id": from_remote_actor.ap_id + "/accept/" + (outbox_public_id or uuid4().hex),
        "actor": from_remote_actor.ap_id,
        "object": for_remote_object.ap_id,
    }


def build_block_activity(
    from_remote_actor: actor.RemoteActor,
    for_remote_actor: actor.RemoteActor,
    outbox_public_id: str | None = None,
) -> ap.RawObject:
    return {
        "@context": ap.AS_CTX,
        "type": "Block",
        "id": from_remote_actor.ap_id + "/block/" + (outbox_public_id or uuid4().hex),
        "actor": from_remote_actor.ap_id,
        "object": for_remote_actor.ap_id,
    }


def build_move_activity(
    from_remote_actor: actor.RemoteActor,
    for_remote_object: actor.RemoteActor,
    outbox_public_id: str | None = None,
) -> ap.RawObject:
    return {
        "@context": ap.AS_CTX,
        "type": "Move",
        "id": from_remote_actor.ap_id + "/move/" + (outbox_public_id or uuid4().hex),
        "actor": from_remote_actor.ap_id,
        "object": from_remote_actor.ap_id,
        "target": for_remote_object.ap_id,
    }


def build_note_object(
    from_remote_actor: actor.RemoteActor | models.Actor,
    outbox_public_id: str | None = None,
    content: str = "Hello",
    to: list[str] = None,
    cc: list[str] = None,
    tags: list[ap.RawObject] = None,
    in_reply_to: str | None = 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,
        "sensitive": False,
        "inReplyTo": in_reply_to,
    }


def build_create_activity(obj: ap.RawObject) -> ap.RawObject:
    return {
        "@context": ap.AS_EXTENDED_CTX,
        "actor": obj["attributedTo"],
        "to": obj.get("to", []),
        "cc": obj.get("cc", []),
        "id": obj["id"] + "/activity",
        "object": ap.remove_context(obj),
        "published": obj["published"],
        "type": "Create",
    }


class BaseModelMeta:
    sqlalchemy_session = _Session
    sqlalchemy_session_persistence = "commit"


class RemoteActorFactory(factory.Factory):
    class Meta:
        model = RemoteActor
        exclude = (
            "base_url",
            "username",
            "public_key",
            "also_known_as",
        )

    class Params:
        icon_url = None
        summary = "I like unit tests"
        also_known_as: list[str] = []

    ap_actor = factory.LazyAttribute(
        lambda o: {
            "@context": ap.AS_CTX,
            "type": "Person",
            "id": o.base_url,
            "following": o.base_url + "/following",
            "followers": o.base_url + "/followers",
            # "featured": ID + "/featured",
            "inbox": o.base_url + "/inbox",
            "outbox": o.base_url + "/outbox",
            "preferredUsername": o.username,
            "name": o.username,
            "summary": o.summary,
            "endpoints": {},
            "url": o.base_url,
            "manuallyApprovesFollowers": False,
            "attachment": [],
            "icon": {},
            "publicKey": {
                "id": f"{o.base_url}#main-key",
                "owner": o.base_url,
                "publicKeyPem": o.public_key,
            },
            "alsoKnownAs": o.also_known_as,
        }
    )


class ActorFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta(BaseModelMeta):
        model = models.Actor

    # ap_actor
    # ap_id
    ap_type = "Person"

    @classmethod
    def from_remote_actor(cls, ra):
        return cls(
            ap_type=ra.ap_type,
            ap_actor=ra.ap_actor,
            ap_id=ra.ap_id,
        )


class OutboxObjectFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta(BaseModelMeta):
        model = models.OutboxObject

    # public_id
    # relates_to_inbox_object_id
    # relates_to_outbox_object_id

    @classmethod
    def from_remote_object(cls, public_id, ro):
        return cls(
            public_id=public_id,
            ap_type=ro.ap_type,
            ap_id=ro.ap_id,
            ap_context=ro.ap_context,
            ap_object=ro.ap_object,
            visibility=ro.visibility,
            og_meta=ro.og_meta,
            activity_object_ap_id=ro.activity_object_ap_id,
            is_hidden_from_homepage=True if ro.in_reply_to else False,
        )


class OutgoingActivityFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta(BaseModelMeta):
        model = models.OutgoingActivity

    # 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).hostname,
            actor_id=actor.id,
            ap_actor_id=actor.ap_id,
            ap_type=ro.ap_type,
            ap_id=ro.ap_id,
            ap_context=ro.ap_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,
        )


class FollowerFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta(BaseModelMeta):
        model = models.Follower


class FollowingFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta(BaseModelMeta):
        model = models.Following