moviewyrm/bookwyrm/tests/views/inbox/test_inbox.py

""" tests incoming activities"""
import json
import pathlib
from unittest.mock import patch

from django.core.exceptions import PermissionDenied
from django.http import HttpResponseNotAllowed, HttpResponseNotFound
from django.test import TestCase, Client
from django.test.client import RequestFactory

from bookwyrm import models, views


# pylint: disable=too-many-public-methods
class Inbox(TestCase):
    """readthrough tests"""

    def setUp(self):
        """basic user and book data"""
        self.client = Client()
        self.factory = RequestFactory()

        with patch("bookwyrm.suggested_users.rerank_suggestions_task.delay"), patch(
            "bookwyrm.activitystreams.populate_stream_task.delay"
        ), patch("bookwyrm.lists_stream.populate_lists_task.delay"):
            local_user = models.User.objects.create_user(
                "mouse@example.com",
                "mouse@mouse.com",
                "mouseword",
                local=True,
                localname="mouse",
            )
        local_user.remote_id = "https://example.com/user/mouse"
        local_user.save(broadcast=False, update_fields=["remote_id"])
        with patch("bookwyrm.models.user.set_remote_server.delay"):
            self.remote_user = models.User.objects.create_user(
                "rat",
                "rat@rat.com",
                "ratword",
                local=False,
                remote_id="https://example.com/users/rat",
                inbox="https://example.com/users/rat/inbox",
                outbox="https://example.com/users/rat/outbox",
            )
        self.create_json = {
            "id": "hi",
            "type": "Create",
            "actor": "hi",
            "to": ["https://www.w3.org/ns/activitystreams#public"],
            "cc": ["https://example.com/user/mouse/followers"],
            "object": {},
        }
        models.SiteSettings.objects.create()

    def test_inbox_invalid_get(self):
        """shouldn't try to handle if the user is not found"""
        result = self.client.get("/inbox", content_type="application/json")
        self.assertIsInstance(result, HttpResponseNotAllowed)

    def test_inbox_invalid_user(self):
        """shouldn't try to handle if the user is not found"""
        result = self.client.post(
            "/user/bleh/inbox",
            '{"type": "Test", "object": "exists"}',
            content_type="application/json",
        )
        self.assertIsInstance(result, HttpResponseNotFound)

    def test_inbox_invalid_bad_signature(self):
        """bad request for invalid signature"""
        with patch("bookwyrm.views.inbox.has_valid_signature") as mock_valid:
            mock_valid.return_value = False
            result = self.client.post(
                "/user/mouse/inbox",
                '{"type": "Announce", "object": "exists"}',
                content_type="application/json",
            )
            self.assertEqual(result.status_code, 401)

    def test_inbox_invalid_bad_signature_delete(self):
        """invalid signature for Delete is okay though"""
        with patch("bookwyrm.views.inbox.has_valid_signature") as mock_valid:
            mock_valid.return_value = False
            result = self.client.post(
                "/user/mouse/inbox",
                '{"type": "Delete", "object": "exists"}',
                content_type="application/json",
            )
            self.assertEqual(result.status_code, 200)

    def test_inbox_unknown_type(self):
        """never heard of that activity type, don't have a handler for it"""
        with patch("bookwyrm.views.inbox.has_valid_signature") as mock_valid:
            result = self.client.post(
                "/inbox",
                '{"type": "Fish", "object": "exists"}',
                content_type="application/json",
            )
            mock_valid.return_value = True
            self.assertIsInstance(result, HttpResponseNotFound)

    def test_inbox_success(self):
        """a known type, for which we start a task"""
        activity = self.create_json
        activity["object"] = {
            "id": "https://example.com/list/22",
            "type": "BookList",
            "totalItems": 1,
            "first": "https://example.com/list/22?page=1",
            "last": "https://example.com/list/22?page=1",
            "name": "Test List",
            "owner": "https://example.com/user/mouse",
            "to": ["https://www.w3.org/ns/activitystreams#Public"],
            "cc": ["https://example.com/user/mouse/followers"],
            "summary": "summary text",
            "curation": "curated",
            "@context": "https://www.w3.org/ns/activitystreams",
        }
        with patch("bookwyrm.views.inbox.has_valid_signature") as mock_valid:
            mock_valid.return_value = True

            with patch("bookwyrm.views.inbox.activity_task.delay"):
                result = self.client.post(
                    "/inbox", json.dumps(activity), content_type="application/json"
                )
        self.assertEqual(result.status_code, 200)

    def test_is_blocked_user_agent(self):
        """check for blocked servers"""
        request = self.factory.post(
            "",
            HTTP_USER_AGENT="http.rb/4.4.1 (Mastodon/3.3.0; +https://mastodon.social/)",
        )
        self.assertIsNone(views.inbox.raise_is_blocked_user_agent(request))

        models.FederatedServer.objects.create(
            server_name="mastodon.social", status="blocked"
        )
        with self.assertRaises(PermissionDenied):
            views.inbox.raise_is_blocked_user_agent(request)

    def test_is_blocked_activity(self):
        """check for blocked servers"""
        activity = {"actor": "https://mastodon.social/user/whaatever/else"}
        self.assertIsNone(views.inbox.raise_is_blocked_activity(activity))

        models.FederatedServer.objects.create(
            server_name="mastodon.social", status="blocked"
        )
        with self.assertRaises(PermissionDenied):
            views.inbox.raise_is_blocked_activity(activity)

    @patch("bookwyrm.suggested_users.remove_user_task.delay")
    def test_create_by_deactivated_user(self, _):
        """don't let deactivated users post"""
        self.remote_user.delete(broadcast=False)
        self.assertTrue(self.remote_user.deleted)
        datafile = pathlib.Path(__file__).parent.joinpath("../../data/ap_note.json")
        status_data = json.loads(datafile.read_bytes())
        activity = self.create_json
        activity["actor"] = self.remote_user.remote_id
        activity["object"] = status_data
        activity["type"] = "Create"

        response = self.client.post(
            "/inbox",
            json.dumps(activity),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 403)