summaryrefslogtreecommitdiffstats
path: root/tests/activities/models
diff options
context:
space:
mode:
Diffstat (limited to 'tests/activities/models')
-rw-r--r--tests/activities/models/test_post.py24
1 files changed, 4 insertions, 20 deletions
diff --git a/tests/activities/models/test_post.py b/tests/activities/models/test_post.py
index 21e7000..00db61b 100644
--- a/tests/activities/models/test_post.py
+++ b/tests/activities/models/test_post.py
@@ -1,7 +1,4 @@
-import asyncio
-
import pytest
-from asgiref.sync import async_to_sync
from pytest_httpx import HTTPXMock
from activities.models import Post, PostStates
@@ -128,21 +125,8 @@ def test_linkify_mentions_local(identity, remote_identity):
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):
+def test_post_transitions(identity, stator):
# Create post
post = Post.objects.create(
@@ -153,18 +137,18 @@ def test_post_transitions(identity, stator_runner):
)
# Test: | --> new --> fanned_out
assert post.state == str(PostStates.new)
- async_to_sync(stator_process_tasks)(stator_runner)
+ stator.run_single_cycle_sync()
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)
+ stator.run_single_cycle_sync()
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)
+ stator.run_single_cycle_sync()
post = Post.objects.get(id=post.id)
assert post.state == str(PostStates.deleted_fanned_out)