ArchiveFeedTests

Complexity

Total Complexity

9

Size/Duplication

Total Lines	71
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	tearDown()	0	2	1
A	setUp()	0	2	1
A	test_archive_feed_fail()	0	24	3
A	test_archive_feed_not_enough_entries()	0	18	2
A	test_archive_feed()	0	19	2

import os
from sqlalchemy import engine_from_config
import unittest
from oe_utils.scripts import archive_feed
try:
    from unittest.mock import Mock, patch, MagicMock, call
except:
    from mock import Mock, patch, MagicMock, call
from oe_utils.data.data_managers import AtomFeedManager
from sqlalchemy.orm import sessionmaker
import responses
from requests.exceptions import HTTPError
try:
    from initfeedgen import fill_db_feeds, Feed, FeedEntry
except:
    from tests.initfeedgen import fill_db_feeds, Feed, FeedEntry
import tempfile

here = os.path.dirname(os.path.abspath(__file__))
settings = {"sqlalchemy.url": "sqlite:///{}/test.sqlite".format(here)}
engine = engine_from_config(settings, prefix='sqlalchemy.')
here = os.path.dirname(os.path.abspath(__file__))
fixture_directory = tempfile.gettempdir()


def setUpModule():
    fill_db_feeds(engine)


class TestAtomFeedManager(AtomFeedManager):
    '''
    Een datamanager voor de feeds.
    '''

    def __init__(self, session, feed_repository):
        super(TestAtomFeedManager, self).__init__(session, feed_repository, Feed, FeedEntry)


class ArchiveFeedTests(unittest.TestCase):

    def setUp(self):
        self.db = sessionmaker(bind=engine)()

    def tearDown(self):
        self.db.close()

    @responses.activate
    def test_archive_feed_not_enough_entries(self):
        with patch('oe_utils.scripts.archive_feed.log') as log_mock:
            responses.add(
                responses.POST,
                url="https://host.be/rapporten/atomfeed/archive",
                status=200)
            feed_list = [{
                'atom_feed_folder': fixture_directory,
                'atom_feed_manager': TestAtomFeedManager,
                'max_entries': 100,
                'archive_feed_url': 'https://host.be/bron/atomfeed/archive'
            }]
            archive_feed.archive_entries(feed_list=feed_list, session=self.db, system_token='test_token')
            debug_calls = [
                call.debug('Number of entries: 10'),
            ]
            self.assertEqual(log_mock.debug.mock_calls, debug_calls)

    @responses.activate
    def test_archive_feed(self):
        with patch('oe_utils.scripts.archive_feed.log') as log_mock:
            responses.add(
                responses.POST,
                url="https://host.be/bron/atomfeed/archive",
                status=200)
            feed_list = [{
                'atom_feed_folder': fixture_directory,
                'atom_feed_manager': TestAtomFeedManager,
                'max_entries': 2,
                'archive_feed_url': 'https://host.be/bron/atomfeed/archive'
            }]
            archive_feed.archive_entries(feed_list=feed_list, session=self.db, system_token='test_token')
            debug_calls = [
                call.debug('Number of entries: 10'),
                call.debug('Archive current feed')
            ]
            self.assertEqual(log_mock.debug.mock_calls, debug_calls)

    @responses.activate
    def test_archive_feed_fail(self):
        with patch('oe_utils.scripts.archive_feed.log') as log_mock:
            responses.add(
                responses.POST,
                url="https://host.be/bron/atomfeed/archive",
                status=500)
            feed_list = [{
                'atom_feed_folder': fixture_directory,
                'atom_feed_manager': TestAtomFeedManager,
                'max_entries': 2,
                'archive_feed_url': 'https://host.be/bron/atomfeed/archive'
            }]
            with self.assertRaises(HTTPError):
                archive_feed.archive_entries(feed_list=feed_list, session=self.db, system_token='test_token')
            debug_calls = [
                call.debug('Number of entries: 10'),
                call.debug('Archive current feed')
            ]
            self.assertEqual(log_mock.debug.mock_calls, debug_calls)
            error_calls = [
                call.error('Failed to archive on url https://host.be/bron/atomfeed/archive'),
            ]
            self.assertEqual(log_mock.error.mock_calls, error_calls)