newsreader/src/newsreader/news/collection/tests/twitter/collector/tests.py

from datetime import datetime
from unittest.mock import patch
from uuid import uuid4

from django.test import TestCase
from django.utils import timezone

import pytz

from freezegun import freeze_time
from ftfy import fix_text

from newsreader.news.collection.choices import RuleTypeChoices
from newsreader.news.collection.exceptions import (
    StreamDeniedException,
    StreamForbiddenException,
    StreamNotFoundException,
    StreamTimeOutException,
)
from newsreader.news.collection.tests.factories import TwitterTimelineFactory
from newsreader.news.collection.tests.twitter.collector.mocks import (
    empty_mock,
    simple_mock,
)
from newsreader.news.collection.twitter import TWITTER_URL, TwitterCollector
from newsreader.news.collection.utils import truncate_text
from newsreader.news.core.models import Post


@freeze_time("2020-09-26 14:40:00")
class TwitterCollectorTestCase(TestCase):
    def setUp(self):
        patched_get = patch("newsreader.news.collection.twitter.fetch")
        self.mocked_fetch = patched_get.start()

        patched_parse = patch("newsreader.news.collection.twitter.TwitterStream.parse")
        self.mocked_parse = patched_parse.start()

    def tearDown(self):
        patch.stopall()

    def test_simple_batch(self):
        self.mocked_parse.return_value = simple_mock

        timeline = TwitterTimelineFactory(
            user__twitter_oauth_token=str(uuid4()),
            user__twitter_oauth_token_secret=str(uuid4()),
            screen_name="RobertsSpaceInd",
            enabled=True,
        )

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertCountEqual(
            Post.objects.values_list("remote_identifier", flat=True),
            ("1307054882210435074", "1307029168941461504"),
        )

        self.assertEquals(timeline.succeeded, True)
        self.assertEquals(timeline.last_run, timezone.now())
        self.assertIsNone(timeline.error)

        post = Post.objects.get(
            remote_identifier="1307054882210435074",
            rule__type=RuleTypeChoices.twitter_timeline,
        )

        self.assertEquals(
            post.publication_date, pytz.utc.localize(datetime(2020, 9, 18, 20, 32, 22))
        )

        title = truncate_text(
            Post,
            "title",
            "It's a close match-up for #SCShipShowdown today! Which Aegis ship"
            " do you think will make it to the Semi-Finals?",
        )

        self.assertEquals(post.author, "RobertsSpaceInd")
        self.assertEquals(post.title, title)
        self.assertEquals(
            post.url, f"{TWITTER_URL}/RobertsSpaceInd/status/1307054882210435074"
        )

        post = Post.objects.get(
            remote_identifier="1307029168941461504",
            rule__type=RuleTypeChoices.twitter_timeline,
        )

        self.assertEquals(
            post.publication_date, pytz.utc.localize(datetime(2020, 9, 18, 18, 50, 11))
        )

        body = fix_text(
            "We\u2019re welcoming members of our Builds, Publishes and Platform"
            " teams on Star Citizen Live to talk about the process involved in"
            " bringing everyone\u2019s work together and getting it out into your"
            " hands. Going live on #Twitch in 10 minutes."
            " \ud83c\udfa5\ud83d\udd34 \n\nTune in:"
            " https://t.co/2AdNovhpFW https://t.co/Cey5JpR1i9"
        )

        title = truncate_text(Post, "title", body)

        self.assertEquals(post.author, "RobertsSpaceInd")
        self.assertEquals(post.title, title)
        self.assertEquals(
            post.url, f"{TWITTER_URL}/RobertsSpaceInd/status/1307029168941461504"
        )

    def test_empty_batch(self):
        self.mocked_parse.return_value = empty_mock

        timeline = TwitterTimelineFactory()

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertEquals(Post.objects.count(), 0)

        self.assertEquals(timeline.succeeded, True)
        self.assertEquals(timeline.last_run, timezone.now())
        self.assertIsNone(timeline.error)

    def test_not_found(self):
        self.mocked_fetch.side_effect = StreamNotFoundException

        timeline = TwitterTimelineFactory()

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertEquals(Post.objects.count(), 0)
        self.assertEquals(timeline.succeeded, False)
        self.assertEquals(timeline.error, "Stream not found")

    def test_denied(self):
        self.mocked_fetch.side_effect = StreamDeniedException

        timeline = TwitterTimelineFactory(
            user__twitter_oauth_token=str(uuid4()),
            user__twitter_oauth_token_secret=str(uuid4()),
        )

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertEquals(Post.objects.count(), 0)
        self.assertEquals(timeline.succeeded, False)
        self.assertEquals(timeline.error, "Stream does not have sufficient permissions")

        user = timeline.user

        self.assertIsNone(user.twitter_oauth_token)
        self.assertIsNone(user.twitter_oauth_token_secret)

    def test_forbidden(self):
        self.mocked_fetch.side_effect = StreamForbiddenException

        timeline = TwitterTimelineFactory()

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertEquals(Post.objects.count(), 0)
        self.assertEquals(timeline.succeeded, False)
        self.assertEquals(timeline.error, "Stream forbidden")

    def test_timed_out(self):
        self.mocked_fetch.side_effect = StreamTimeOutException

        timeline = TwitterTimelineFactory()

        collector = TwitterCollector()
        collector.collect(rules=[timeline])

        self.assertEquals(Post.objects.count(), 0)
        self.assertEquals(timeline.succeeded, False)
        self.assertEquals(timeline.error, "Stream timed out")