import pytest from django.utils import timezone from activities.models import Post, TimelineEvent from activities.services import PostService from core.ld import format_ld_date from users.models import Block, Follow, Identity, InboxMessage from users.services import IdentityService @pytest.mark.django_db @pytest.mark.parametrize("local", [True, False]) @pytest.mark.parametrize("blocked", ["full", "mute", "no"]) def test_mentioned( identity: Identity, other_identity: Identity, remote_identity: Identity, stator, local: bool, blocked: bool, ): """ Ensures that a new or incoming post that mentions a local identity results in a mentioned timeline event, unless the author is blocked. """ if local: Post.create_local(author=other_identity, content=f"Hello @{identity.handle}!") else: # Create an inbound new post message message = { "id": "test", "type": "Create", "actor": remote_identity.actor_uri, "object": { "id": "https://remote.test/test-post", "type": "Note", "published": format_ld_date(timezone.now()), "attributedTo": remote_identity.actor_uri, "content": f"Hello @{identity.handle}!", "tag": { "type": "Mention", "href": identity.actor_uri, "name": f"@{identity.handle}", }, }, } InboxMessage.objects.create(message=message) # Implement any blocks author = other_identity if local else remote_identity if blocked == "full": Block.create_local_block(identity, author) elif blocked == "mute": Block.create_local_mute(identity, author) # Run stator twice - to make fanouts and then process them stator.run_single_cycle_sync() stator.run_single_cycle_sync() if blocked in ["full", "mute"]: # Verify we were not mentioned assert not TimelineEvent.objects.filter( type=TimelineEvent.Types.mentioned, identity=identity ).exists() else: # Verify we got mentioned event = TimelineEvent.objects.filter( type=TimelineEvent.Types.mentioned, identity=identity ).first() assert event assert event.subject_identity == author assert "Hello " in event.subject_post.content @pytest.mark.django_db @pytest.mark.parametrize("local", [True, False]) @pytest.mark.parametrize("type", ["like", "boost"]) @pytest.mark.parametrize("blocked", ["full", "mute", "mute_with_notifications", "no"]) def test_interaction_local_post( identity: Identity, other_identity: Identity, remote_identity: Identity, stator, local: bool, type: str, blocked: bool, ): """ Ensures that a like of a local Post notifies its author """ post = Post.create_local(author=identity, content="I love birds!") if local: if type == "boost": PostService(post).boost_as(other_identity) else: PostService(post).like_as(other_identity) else: if type == "boost": message = { "id": "test", "type": "Announce", "to": "as:Public", "actor": remote_identity.actor_uri, "object": post.object_uri, } else: message = { "id": "test", "type": "Like", "actor": remote_identity.actor_uri, "object": post.object_uri, } InboxMessage.objects.create(message=message) # Implement any blocks interactor = other_identity if local else remote_identity if blocked == "full": Block.create_local_block(identity, interactor) elif blocked == "mute": Block.create_local_mute(identity, interactor) elif blocked == "mute_with_notifications": Block.create_local_mute(identity, interactor, include_notifications=True) # Run stator twice - to make fanouts and then process them stator.run_single_cycle_sync() stator.run_single_cycle_sync() timeline_event_type = ( TimelineEvent.Types.boosted if type == "boost" else TimelineEvent.Types.liked ) if blocked in ["full", "mute_with_notifications"]: # Verify we did not get an event assert not TimelineEvent.objects.filter( type=timeline_event_type, identity=identity ).exists() else: # Verify we got an event event = TimelineEvent.objects.filter( type=timeline_event_type, identity=identity ).first() assert event assert event.subject_identity == interactor @pytest.mark.django_db @pytest.mark.parametrize("old", [True, False]) def test_old_new_post( identity: Identity, remote_identity: Identity, stator, old: bool, ): """ Ensures that old remote posts don't appear on the timeline, but new ones do. """ # Follow the remote user Follow.create_local(identity, remote_identity) # Create an inbound new post message message = { "id": "test", "type": "Create", "actor": remote_identity.actor_uri, "object": { "id": "https://remote.test/test-post", "type": "Note", "published": "2022-01-01T00:00:00Z" if old else format_ld_date(timezone.now()), "attributedTo": remote_identity.actor_uri, "content": f"Hello @{identity.handle}!", "tag": { "type": "Mention", "href": identity.actor_uri, "name": f"@{identity.handle}", }, }, } InboxMessage.objects.create(message=message) # Run stator twice - to make fanouts and then process them stator.run_single_cycle_sync() stator.run_single_cycle_sync() if old: # Verify it did not appear on the timeline assert not TimelineEvent.objects.filter( type=TimelineEvent.Types.post, identity=identity ).exists() else: # Verify it appeared on the timeline event = TimelineEvent.objects.filter( type=TimelineEvent.Types.post, identity=identity ).first() assert event assert "Hello " in event.subject_post.content @pytest.mark.django_db @pytest.mark.parametrize("full", [True, False]) def test_clear_timeline( identity: Identity, remote_identity: Identity, stator, full: bool, ): """ Ensures that timeline clearing works as expected. """ # Follow the remote user service = IdentityService(remote_identity) service.follow_from(identity) # Create an inbound new post message mentioning us message = { "id": "test", "type": "Create", "actor": remote_identity.actor_uri, "object": { "id": "https://remote.test/test-post", "type": "Note", "published": format_ld_date(timezone.now()), "attributedTo": remote_identity.actor_uri, "content": f"Hello @{identity.handle}!", "tag": { "type": "Mention", "href": identity.actor_uri, "name": f"@{identity.handle}", }, }, } InboxMessage.objects.create(message=message) # Run stator twice - to make fanouts and then process them stator.run_single_cycle_sync() stator.run_single_cycle_sync() # Make sure it appeared on our timeline as a post and a mentioned assert TimelineEvent.objects.filter( type=TimelineEvent.Types.post, identity=identity ).exists() assert TimelineEvent.objects.filter( type=TimelineEvent.Types.mentioned, identity=identity ).exists() # Now, submit either a user block (for full clear) or unfollow (for post clear) if full: service.block_from(identity) else: service.unfollow_from(identity) # Run stator once to process the timeline clear message stator.run_single_cycle_sync() # Verify that the right things vanished assert not TimelineEvent.objects.filter( type=TimelineEvent.Types.post, identity=identity ).exists() assert TimelineEvent.objects.filter( type=TimelineEvent.Types.mentioned, identity=identity ).exists() == (not full)