From 6c7ddedd342553b53dd98c8de9cbe9e8e2e8cd7c Mon Sep 17 00:00:00 2001 From: Michael Manfre Date: Sun, 27 Nov 2022 13:09:46 -0500 Subject: Post editing --- tests/activities/models/test_post.py | 47 +++++++++++++++++++++++++++++++++++- tests/activities/views/test_posts.py | 44 ++++++++++++++++++++++++++++++++- tests/conftest.py | 26 ++++++++++++++++++++ 3 files changed, 115 insertions(+), 2 deletions(-) (limited to 'tests') diff --git a/tests/activities/models/test_post.py b/tests/activities/models/test_post.py index 9d5207c..baeb55a 100644 --- a/tests/activities/models/test_post.py +++ b/tests/activities/models/test_post.py @@ -1,7 +1,10 @@ +import asyncio + import pytest +from asgiref.sync import async_to_sync from pytest_httpx import HTTPXMock -from activities.models import Post +from activities.models import Post, PostStates @pytest.mark.django_db @@ -112,3 +115,45 @@ def test_linkify_mentions_local(identity, remote_identity): local=True, ) assert post.safe_content_local() == "

@test@example.com, welcome!

" + + +async def stator_process_tasks(stator): + """ + Guarded wrapper to simply async_to_sync and ensure all stator tasks are + run to completion without blocking indefinitely. + """ + await asyncio.wait_for(stator.fetch_and_process_tasks(), timeout=1) + for _ in range(100): + if not stator.tasks: + break + stator.remove_completed_tasks() + await asyncio.sleep(0.01) + + +@pytest.mark.django_db +def test_post_transitions(identity, stator_runner): + + # Create post + post = Post.objects.create( + content="

Hello!

", + author=identity, + local=False, + visibility=Post.Visibilities.mentioned, + ) + # Test: | --> new --> fanned_out + assert post.state == str(PostStates.new) + async_to_sync(stator_process_tasks)(stator_runner) + post = Post.objects.get(id=post.id) + assert post.state == str(PostStates.fanned_out) + + # Test: fanned_out --> (forced) edited --> edited_fanned_out + Post.transition_perform(post, PostStates.edited) + async_to_sync(stator_process_tasks)(stator_runner) + post = Post.objects.get(id=post.id) + assert post.state == str(PostStates.edited_fanned_out) + + # Test: edited_fanned_out --> (forced) deleted --> deleted_fanned_out + Post.transition_perform(post, PostStates.deleted) + async_to_sync(stator_process_tasks)(stator_runner) + post = Post.objects.get(id=post.id) + assert post.state == str(PostStates.deleted_fanned_out) diff --git a/tests/activities/views/test_posts.py b/tests/activities/views/test_posts.py index b04c30f..c73dcd6 100644 --- a/tests/activities/views/test_posts.py +++ b/tests/activities/views/test_posts.py @@ -2,8 +2,10 @@ import re import mock import pytest +from django.core.exceptions import PermissionDenied -from activities.views.posts import Compose +from activities.models import Post +from activities.views.posts import Compose, Delete @pytest.mark.django_db @@ -22,3 +24,43 @@ def test_content_warning_text(identity, user, rf, config_system): assert re.search( r"\s*Content Summary\s*", content, flags=re.MULTILINE ) + + +@pytest.mark.django_db +def test_post_delete_security(identity, user, rf, other_identity): + # Create post + other_post = Post.objects.create( + content="

OTHER POST!

", + author=other_identity, + local=True, + visibility=Post.Visibilities.public, + ) + + request = rf.post(other_post.get_absolute_url() + "delete/") + request.user = user + request.identity = identity + + view = Delete.as_view() + with pytest.raises(PermissionDenied) as ex: + view(request, handle=other_identity.handle.lstrip("@"), post_id=other_post.id) + assert str(ex.value) == "Post author is not requestor" + + +@pytest.mark.django_db +def test_post_edit_security(identity, user, rf, other_identity): + # Create post + other_post = Post.objects.create( + content="

OTHER POST!

", + author=other_identity, + local=True, + visibility=Post.Visibilities.public, + ) + + request = rf.get(other_post.get_absolute_url() + "edit/") + request.user = user + request.identity = identity + + view = Compose.as_view() + with pytest.raises(PermissionDenied) as ex: + view(request, handle=other_identity.handle.lstrip("@"), post_id=other_post.id) + assert str(ex.value) == "Post author is not requestor" diff --git a/tests/conftest.py b/tests/conftest.py index d506c5c..a3feaca 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,9 @@ +import time + import pytest from core.models import Config +from stator.runner import StatorModel, StatorRunner from users.models import Domain, Identity, User @@ -120,3 +123,26 @@ def remote_identity() -> Identity: name="Test Remote User", local=False, ) + + +@pytest.fixture +def stator_runner(config_system) -> StatorRunner: + """ + Return an initialized StatorRunner for tests that need state transitioning + to happen. + + Example: + # Do some tasks with state side effects + async_to_sync(stator_runner.fetch_and_process_tasks)() + """ + runner = StatorRunner( + StatorModel.subclasses, + concurrency=100, + schedule_interval=30, + ) + runner.handled = 0 + runner.started = time.monotonic() + runner.last_clean = time.monotonic() - runner.schedule_interval + runner.tasks = [] + + return runner -- cgit v1.2.3