TestAtomFeedManager

Complexity

Total Complexity

9

Size/Duplication

Total Lines	31
Duplicated Lines	0 %

Importance

Changes	2
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	test_current_feed()	0	3	1
A	test_fail_to_not_existing_from_archive()	0	4	2
A	setUp()	0	4	1
A	tearDown()	0	6	3
A	test_get_from_archive()	0	9	2

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

from sqlalchemy import func, engine_from_config
from sqlalchemy.orm import sessionmaker

from oe_utils.data.data_managers import AtomFeedManager, FeedArchiveNotFound
try:
    from initfeedgen import fill_db_feeds, Feed, FeedEntry
except:
    from tests.initfeedgen import fill_db_feeds, Feed, FeedEntry
try:
    from unittest.mock import Mock, patch, MagicMock, call, PropertyMock
except ImportError:
    from mock import Mock, patch, MagicMock, call, PropertyMock  # pragma: no cover
import tempfile
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()


def setUpModule():
    fill_db_feeds(engine)


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

    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_current_feed(self):
        current_feed_id = self.session.query(func.max(Feed.id)).scalar()
        self.assertEqual(current_feed_id, self.atom_feed_manager.current_feed.id)

    def test_get_from_archive(self):
        feed_id = self.atom_feed_manager.current_feed.id
        archive_content = 'some xml string'
        feed_path = fixture_directory + '/' + str(feed_id) + '.xml'
        with open(feed_path, 'w') as rec_file:
            rec_file.write(archive_content)
        self.files_to_clean.append(feed_path)
        response = self.atom_feed_manager.get_from_archive(feed_id)
        self.assertEqual(archive_content, response)

    def test_fail_to_not_existing_from_archive(self):
        feed_id = '999'
        with self.assertRaises(FeedArchiveNotFound):
            self.atom_feed_manager.get_from_archive(feed_id)