microblog.pub/tests/factories.py

267 lines
7.6 KiB
Python

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_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