[tests] add inbox test
This commit is contained in:
parent
27c2f39f12
commit
25c9e1ef11
3 changed files with 108 additions and 10 deletions
|
@ -87,6 +87,10 @@ class BaseActor:
|
||||||
def ap_actor(self) -> RawObject:
|
def ap_actor(self) -> RawObject:
|
||||||
return self._ap_actor
|
return self._ap_actor
|
||||||
|
|
||||||
|
@property
|
||||||
|
def ap_id(self) -> RawObject:
|
||||||
|
return self._ap_actor["id"]
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def inbox_url(self) -> str:
|
def inbox_url(self) -> str:
|
||||||
return self.ap_actor["inbox"]
|
return self.ap_actor["inbox"]
|
||||||
|
|
63
tests/factories.py
Normal file
63
tests/factories.py
Normal file
|
@ -0,0 +1,63 @@
|
||||||
|
#!/usr/bin/env python3
|
||||||
|
import factory
|
||||||
|
import app.activitypub as ap
|
||||||
|
from app import models
|
||||||
|
|
||||||
|
from sqlalchemy import orm
|
||||||
|
|
||||||
|
from app import activitypub as ap
|
||||||
|
from app import actor
|
||||||
|
from app import models
|
||||||
|
from app.database import SessionLocal
|
||||||
|
|
||||||
|
_Session = orm.scoped_session(SessionLocal)
|
||||||
|
|
||||||
|
|
||||||
|
class BaseModelMeta:
|
||||||
|
sqlalchemy_session = _Session
|
||||||
|
sqlalchemy_session_persistence = "commit"
|
||||||
|
|
||||||
|
|
||||||
|
class RemoteActorFactory(factory.Factory): # pyright: disable
|
||||||
|
class Meta:
|
||||||
|
model = ap.BaseActor
|
||||||
|
exclude = (
|
||||||
|
"base_url",
|
||||||
|
"username",
|
||||||
|
"public_key",
|
||||||
|
"also_known_as",
|
||||||
|
)
|
||||||
|
|
||||||
|
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",
|
||||||
|
"inbox": o.base_url + "/inbox",
|
||||||
|
"outbox": o.base_url + "/outbox",
|
||||||
|
"preferredUsername": o.username,
|
||||||
|
"name": o.username,
|
||||||
|
"summary": "test user",
|
||||||
|
"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": [],
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class ActorFactory(factory.alchemy.SQLAlchemyModelFactory):
|
||||||
|
class Meta(BaseModelMeta):
|
||||||
|
model = models.Actor
|
||||||
|
|
||||||
|
ap_type = "Person"
|
||||||
|
ap_id = "stub"
|
|
@ -2,16 +2,23 @@
|
||||||
import pytest
|
import pytest
|
||||||
import fastapi
|
import fastapi
|
||||||
import respx
|
import respx
|
||||||
|
import httpx
|
||||||
|
from uuid import uuid4
|
||||||
|
from tests import factories # type: ignore
|
||||||
|
|
||||||
|
from unittest import mock
|
||||||
from app.main import app
|
from app.main import app
|
||||||
from app.utils import precheck
|
from app.utils import precheck
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
from app import activitypub as ap, httpsig
|
from app import activitypub as ap, httpsig
|
||||||
from app.config import AP_CONTENT_TYPE
|
from app.config import AP_CONTENT_TYPE
|
||||||
|
from app import models
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def test_inbox_announce_request(
|
def test_inbox_follow_request(
|
||||||
db: Session,
|
db: Session,
|
||||||
client: TestClient,
|
client: TestClient,
|
||||||
respx_mock: respx.MockRouter,
|
respx_mock: respx.MockRouter,
|
||||||
|
@ -22,14 +29,38 @@ def test_inbox_announce_request(
|
||||||
) -> bool:
|
) -> bool:
|
||||||
return True
|
return True
|
||||||
|
|
||||||
app.dependency_overrides[precheck.inbox_prechecker] = inbox_prechecker
|
ra = factories.RemoteActorFactory(
|
||||||
|
base_url="https://example.com",
|
||||||
response = client.post(
|
username="test",
|
||||||
"/inbox",
|
public_key="pk",
|
||||||
headers={"Content-Type": AP_CONTENT_TYPE},
|
|
||||||
json={"stub": 1},
|
|
||||||
)
|
)
|
||||||
|
|
||||||
print(response.headers)
|
ap_id = ra.ap_id # type: ignore
|
||||||
print(response.content)
|
respx_mock.get(ap_id).mock(return_value=httpx.Response(200,json=ra.ap_actor))
|
||||||
assert response.status_code == 200
|
respx_mock.post(ap_id + "/inbox").mock(return_value=httpx.Response(202))
|
||||||
|
|
||||||
|
app.dependency_overrides[precheck.inbox_prechecker] = inbox_prechecker
|
||||||
|
|
||||||
|
with mock.patch("app.boxes.MANUALLY_APPROVES_FOLLOWERS", False):
|
||||||
|
response = client.post(
|
||||||
|
"/inbox",
|
||||||
|
headers={"Content-Type": AP_CONTENT_TYPE},
|
||||||
|
json={
|
||||||
|
"@context": ap.AS_CTX,
|
||||||
|
"type": "Follow",
|
||||||
|
"id": ap_id + "/follow/" + (uuid4().hex),
|
||||||
|
"actor": ap_id,
|
||||||
|
"object": ap.ME["id"],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 202
|
||||||
|
|
||||||
|
saved_actor = db.execute(select(models.Actor)).scalar_one()
|
||||||
|
assert saved_actor.ap_id == ap_id
|
||||||
|
|
||||||
|
follower_actor = db.execute(select(models.Follower)).scalar_one()
|
||||||
|
assert follower_actor.ap_actor_id == ap_id
|
||||||
|
|
||||||
|
outbox_object = db.execute(select(models.OutboxObject)).scalar_one()
|
||||||
|
assert outbox_object.ap_type == "Accept"
|
||||||
|
|
Loading…
Reference in a new issue