newsreader/src/newsreader/news/collection/tests/feed/stream/tests.py
sonny 177755e302 Squashed commit of the following:
commit 99fd94580f95dcbfb77b73e2de846f76a5709ef9
Author: Sonny <sonnyba871@gmail.com>
Date:   Sat Feb 15 21:45:16 2020 +0100

    Use postgres password

    As of https://gitlab.com/gitlab-com/support-forum/issues/5199
2020-07-12 20:26:14 +02:00

105 lines
3 KiB
Python

from unittest.mock import MagicMock, patch
from django.test import TestCase
from newsreader.news.collection.exceptions import (
StreamDeniedException,
StreamException,
StreamForbiddenException,
StreamNotFoundException,
StreamParseException,
StreamTimeOutException,
)
from newsreader.news.collection.feed import FeedStream
from newsreader.news.collection.tests.factories import FeedFactory
from .mocks import simple_mock, simple_mock_parsed
class FeedStreamTestCase(TestCase):
def setUp(self):
self.maxDiff = None
self.patched_fetch = patch("newsreader.news.collection.feed.fetch")
self.mocked_fetch = self.patched_fetch.start()
def tearDown(self):
patch.stopall()
def test_simple_stream(self):
self.mocked_fetch.return_value = MagicMock(content=simple_mock)
rule = FeedFactory()
stream = FeedStream(rule)
data, stream = stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
self.assertEquals(data, simple_mock_parsed)
self.assertEquals(stream.rule, rule)
def test_stream_raises_exception(self):
self.mocked_fetch.side_effect = StreamException
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamException):
stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
def test_stream_raises_denied_exception(self):
self.mocked_fetch.side_effect = StreamDeniedException
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamDeniedException):
stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
def test_stream_raises_not_found_exception(self):
self.mocked_fetch.side_effect = StreamNotFoundException
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamNotFoundException):
stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
def test_stream_raises_time_out_exception(self):
self.mocked_fetch.side_effect = StreamTimeOutException
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamTimeOutException):
stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
def test_stream_raises_forbidden_exception(self):
self.mocked_fetch.side_effect = StreamForbiddenException
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamForbiddenException):
stream.read()
self.mocked_fetch.assert_called_once_with(rule.url)
@patch("newsreader.news.collection.feed.parse")
def test_stream_raises_parse_exception(self, mocked_parse):
self.mocked_fetch.return_value = MagicMock()
mocked_parse.side_effect = TypeError
rule = FeedFactory()
stream = FeedStream(rule)
with self.assertRaises(StreamParseException):
stream.read()