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() == "<p>@test@example.com, welcome!</p>"
+
+
+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="<p>Hello!</p>",
+        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"<label.*>\s*Content Summary\s*</label>", 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="<p>OTHER POST!</p>",
+        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="<p>OTHER POST!</p>",
+        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