Add tests for poll creation

main
Thomas Sileo 2022-07-25 08:14:34 +02:00
parent 4d0f439c8c
commit c8bd23b696
2 changed files with 140 additions and 4 deletions

View File

@ -24,10 +24,12 @@ async def async_db_session():
@pytest.fixture @pytest.fixture
def db() -> Generator: def db() -> Generator:
Base.metadata.create_all(bind=engine) Base.metadata.create_all(bind=engine)
try: with _Session() as db_session:
yield _Session try:
finally: yield db_session
Base.metadata.drop_all(bind=engine) finally:
db_session.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture @pytest.fixture

View File

@ -161,3 +161,137 @@ def test_send_create_activity__with_followers(
outgoing_activity = db.query(models.OutgoingActivity).one() outgoing_activity = db.query(models.OutgoingActivity).one()
assert outgoing_activity.outbox_object_id == outbox_object.id assert outgoing_activity.outbox_object_id == outbox_object.id
assert outgoing_activity.recipient == follower.actor.inbox_url assert outgoing_activity.recipient == follower.actor.inbox_url
def test_send_create_activity__question__one_of(
db: Session,
client: TestClient,
respx_mock: respx.MockRouter,
) -> None:
# given a remote actor
ra = factories.RemoteActorFactory(
base_url="https://example.com",
username="toto",
public_key="pk",
)
respx_mock.get(ra.ap_id).mock(return_value=httpx.Response(200, json=ra.ap_actor))
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,
)
with mock.patch.object(webfinger, "get_actor_url", return_value=ra.ap_id):
response = client.post(
"/admin/actions/new",
data={
"redirect_url": "http://testserver/",
"content": "hi followers",
"visibility": ap.VisibilityEnum.PUBLIC.name,
"csrf_token": generate_csrf_token(),
"poll_type": "oneOf",
"poll_duration": 5,
"poll_answer_1": "A",
"poll_answer_2": "B",
},
cookies=generate_admin_session_cookies(),
)
# Then the server returns a 302
assert response.status_code == 302
# And the Follow activity was created in the outbox
outbox_object = db.query(models.OutboxObject).one()
assert outbox_object.ap_type == "Question"
assert outbox_object.is_one_of_poll is True
assert len(outbox_object.poll_items) == 2
assert {pi["name"] for pi in outbox_object.poll_items} == {"A", "B"}
assert outbox_object.is_poll_ended is False
# And an outgoing activity was queued
outgoing_activity = db.query(models.OutgoingActivity).one()
assert outgoing_activity.outbox_object_id == outbox_object.id
assert outgoing_activity.recipient == follower.actor.inbox_url
def test_send_create_activity__question__any_of(
db: Session,
client: TestClient,
respx_mock: respx.MockRouter,
) -> None:
# given a remote actor
ra = factories.RemoteActorFactory(
base_url="https://example.com",
username="toto",
public_key="pk",
)
respx_mock.get(ra.ap_id).mock(return_value=httpx.Response(200, json=ra.ap_actor))
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,
)
with mock.patch.object(webfinger, "get_actor_url", return_value=ra.ap_id):
response = client.post(
"/admin/actions/new",
data={
"redirect_url": "http://testserver/",
"content": "hi followers",
"visibility": ap.VisibilityEnum.PUBLIC.name,
"csrf_token": generate_csrf_token(),
"poll_type": "anyOf",
"poll_duration": 10,
"poll_answer_1": "A",
"poll_answer_2": "B",
"poll_answer_3": "C",
"poll_answer_4": "D",
},
cookies=generate_admin_session_cookies(),
)
# Then the server returns a 302
assert response.status_code == 302
# And the Follow activity was created in the outbox
outbox_object = db.query(models.OutboxObject).one()
assert outbox_object.ap_type == "Question"
assert outbox_object.is_one_of_poll is False
assert len(outbox_object.poll_items) == 4
assert {pi["name"] for pi in outbox_object.poll_items} == {"A", "B", "C", "D"}
assert outbox_object.is_poll_ended is False
# And an outgoing activity was queued
outgoing_activity = db.query(models.OutgoingActivity).one()
assert outgoing_activity.outbox_object_id == outbox_object.id
assert outgoing_activity.recipient == follower.actor.inbox_url