TestAtomFeedView

Complexity

Total Complexity

2

Size/Duplication

Total Lines	14
Duplicated Lines	0 %

Importance

Changes	2
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	generate_atom_feed()	0	4	1
A	__init__()	0	6	1

# -*- coding: utf-8 -*-
import os
import unittest

from sqlalchemy.orm import sessionmaker
from pyramid.compat import text_

from oe_utils.views.atom import AtomFeedView
from oe_utils.data.data_managers import AtomFeedManager, FeedArchiveNotFound
from mock import mock_open

try:
    from unittest.mock import Mock, patch, MagicMock, call
except ImportError:
    from mock import Mock, patch, MagicMock, call  # pragma: no cover
from sys import version_info
if version_info.major == 2:
    import __builtin__ as builtins  # pragma: no cover
else:
    import builtins
import tempfile
try:
    from initfeedgen import fill_db_feeds, Feed, FeedEntry
except:
    from tests.initfeedgen import fill_db_feeds, Feed, FeedEntry
from sqlalchemy import func, engine_from_config
try:
    import configparser
except ImportError:
    import ConfigParser as configparser


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()


class TestAtomFeedView(AtomFeedView):

    def __init__(self, request):
        super(TestAtomFeedView, self).__init__(
            request=request,
            atom_feed_manager=request.data_managers['atom_feed_manager'],
            get_atom_feed_url='test',
            generate_atom_feed=self.generate_atom_feed
        )

    def generate_atom_feed(self, feed):
        atom_feed = self.init_atom_feed(feed)
        atom_feed.title("Feed test")
        return atom_feed


def setUpModule():
    fill_db_feeds(engine)


class TestAtomFeedViews(unittest.TestCase):

    def setUp(self):
        session_maker = sessionmaker(
            bind=engine
        )
        self.session = session_maker()
        self.atom_feed_manager = AtomFeedManager(self.session, fixture_directory, Feed, FeedEntry)
        self.files_to_clean = []

        self.request = MagicMock()
        self.request.data_managers = {'atom_feed_manager': self.atom_feed_manager}
        self.request.route_url = MagicMock(side_effect=self.route_url_side_effect)
        self.request.matchdict = {'id': '1'}
        self.request.registry.settings = {'feed.archive.split': '4'}
        self.rest_views = TestAtomFeedView(self.request)
        self.min_max = self.session.query(func.min(Feed.id), func.max(Feed.id)).one()

    def route_url_side_effect(self, route_name, **kwargs):
        response = route_name
        for arg in kwargs:
            response += '__' + arg + '_' + str(kwargs[arg])
        return response

    def tearDown(self):
        for file_to_clean in self.files_to_clean:
            if os.path.isfile(file_to_clean):
                os.remove(file_to_clean)
        self.session.rollback()
        self.session.close()

    def test_get_current_atom_feed(self):
        response = self.rest_views.get_current_atom_feed()
        self.assertEqual(response, text_('test__id_2'))

    def test_can_get_current_atom_feed_by_id(self):
        feed_id = self.atom_feed_manager.current_feed.id
        self.request.matchdict = {'id': str(feed_id)}
        response = self.rest_views.get_atom_feed()
        self.assertIn(text_('<id>test__id_2</id>'), text_(response))

    @patch('oe_utils.data.data_managers.AtomFeedManager.current_feed')
    @patch('oe_utils.data.data_managers.AtomFeedManager.get_from_archive')
    def test_get_archived_atom_feed(self, get_from_archive_mock, current_feed_mock):
        self.request.matchdict['id'] = '1'
        current_feed_mock.id = 15
        get_from_archive_mock.return_value = 'whatever'
        response = self.rest_views.get_atom_feed()
        get_from_archive_mock.assert_called_once_with(1)
        self.assertIsNotNone(response)
        self.assertEqual(response, text_('whatever'))

    @patch('oe_utils.data.data_managers.AtomFeedManager.current_feed')
    @patch('oe_utils.data.data_managers.AtomFeedManager.get_from_archive')
    def test_get_archived_atom_feed_not_existing(self, get_from_archive_mock, current_feed_mock):
        self.request.matchdict['id'] = '1'
        current_feed_mock.id = 15

        def side_effect(*args, **kwargs):
            raise FeedArchiveNotFound()

        get_from_archive_mock.side_effect = side_effect
        with self.assertRaises(FeedArchiveNotFound):
            self.rest_views.get_atom_feed()
        get_from_archive_mock.assert_called_once_with(1)

    def test_archive_split_current_feed(self):

This code seems to be duplicated in your project.

        old_current_id = self.atom_feed_manager.current_feed.id
        feed_archive_path = fixture_directory + '/' + str(old_current_id) + '.xml'
        m = mock_open()
        with patch.object(builtins, 'open', m):
            self.rest_views.archive_current_feed()
        m.assert_has_calls(call(feed_archive_path, 'w'))
        handle = m()
        self.assertTrue(handle.write.called)
        new_current_id = self.atom_feed_manager.current_feed.id
        self.assertNotEqual(new_current_id, old_current_id)
        self.request.route_url.assert_has_calls([call('test', id=new_current_id)])
        self.assertEqual(2, len(self.atom_feed_manager.current_feed.entries))

    def test_archive_current_feed(self):

This code seems to be duplicated in your project.

        self.request.registry.settings['feed.archive.split'] = None
        old_current_id = self.atom_feed_manager.current_feed.id
        feed_archive_path = fixture_directory + '/' + str(old_current_id) + '.xml'
        m = mock_open()
        with patch.object(builtins, 'open', m):
            self.rest_views.archive_current_feed()
        m.assert_called_once_with(feed_archive_path, 'w')
        handle = m()
        self.assertTrue(handle.write.called)
        new_current_id = self.atom_feed_manager.current_feed.id
        self.assertNotEqual(new_current_id, old_current_id)
        self.request.route_url.assert_has_calls([call('test', id=new_current_id)])

    def validate_atom_feed(self, feed, atom_feed, siblings={}):
        self.assertEqual(atom_feed.id(), 'test__id_' + str(feed.id))
        atom_feed_url = atom_feed.link()[0]
        self.assertEqual(atom_feed_url['href'], 'test__id_' + str(feed.id))
        self.assertEqual(atom_feed_url['rel'], 'self')
        self.assertEqual(atom_feed.title(), 'Feed test')
        route_calls = [call('test', id=feed.id)]
        if 'previous' in siblings:
            route_calls.append(call('test', id=siblings['previous']))
        if 'next' in siblings:
            route_calls.append(call('test', id=siblings['next']))

    def test_get_first_atom_feed(self):

This code seems to be duplicated in your project.

        feed = self.session.query(Feed).filter(Feed.id == self.min_max[0]).one()
        self.request.matchdict = {'id': feed.id}
        atom_feed = self.rest_views.generate_atom_feed(feed)
        next_id = feed.id + 1
        self.validate_atom_feed(feed, atom_feed, {'next': next_id})
        self.assertEqual(len(atom_feed.link()), 2)
        link = atom_feed.link()[1]
        self.assertIsInstance(link, dict)
        self.assertEqual(link['rel'], 'next-archive')
        self.assertEqual(link['href'], 'test__id_' + str(next_id))

    def test_get_last_atom_feed(self):

This code seems to be duplicated in your project.

        feed = self.session.query(Feed).filter(Feed.id == self.min_max[1]).one()
        atom_feed = self.rest_views.generate_atom_feed(feed)
        previous_id = feed.id - 1
        self.validate_atom_feed(feed, atom_feed, {'previous': previous_id})
        self.assertEqual(len(atom_feed.link()), 2)
        link = atom_feed.link()[1]
        self.assertIsInstance(link, dict)
        self.assertEqual(link['rel'], 'prev-archive')
        self.assertEqual(link['href'], 'test__id_' + str(previous_id))

    def test_get_atom_feed_entry(self):
        entry = self.session.query(FeedEntry).first()
        self.request.matchdict['id'] = entry.id
        entry2 = self.rest_views.get_atom_feed_entry()
        self.assertEqual(entry2, entry)

    def test_default_generate_atomfeed(self):
        atom_feed_view = AtomFeedView(self.request, self.atom_feed_manager, 'test')
        feed = self.session.query(Feed).filter(Feed.id == self.min_max[1]).one()
        atom_feed = atom_feed_view.generate_atom_feed(feed)
        self.assertEqual(atom_feed.id(), 'test__id_' + str(feed.id))
        atom_feed_url = atom_feed.link()[0]
        self.assertEqual(atom_feed_url['href'], 'test__id_' + str(feed.id))
        self.assertEqual(atom_feed_url['rel'], 'self')
        self.assertEqual(atom_feed.title(), 'Feed')