import uuid from unittest import TestCase from mock import patch from werkzeug.exceptions import NotFound from distribution import app from distribution.sequencer import FeedSequencer from test.utils import get_stored_item class FeedSequencerTestCase(TestCase): def test_invalid_sequence(self): with self.assertRaises(ValueError): FeedSequencer({'items': [], 'domain': 'news', 'sequence': 'this_is_an_invalid_sequence'}) def get_items(self): return [get_stored_item(with_meta=False, publication_datetime='Sun Apr 02 2015 12:55:00 GMT-0500'), get_stored_item(with_meta=False, publication_datetime='Wed Mar 04 2015 14:55:00 GMT-0500'), get_stored_item(with_meta=False, publication_datetime='Sun Mar 01 2015 15:55:00 GMT-0500'), get_stored_item(with_meta=False, publication_datetime='Mon Mar 02 2015 16:55:00 GMT-0500')] @patch('distribution.models.feed.FeedItem.manager.purge') @patch('distribution.models.feed.FeedItem.manager.bulk_save') @patch('distribution.models.taxonomy.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_current(self, search_by_field_mock, get_mock, bulk_save_mock, purge_mock): items = self.get_items() sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'current'}) ritems, rerrors = sequencer.execute() purge_mock.assert_called_once_with('news', '2015-03-01T20:55:00.000000Z', 'NOW') self.assertEqual(bulk_save_mock.call_count, 1) self.assertEqual(len(bulk_save_mock.call_args[0][0]), 4) self.assertEqual(rerrors, []) self.assertEqual(ritems, [x.get('id') for x in items]) @patch('distribution.models.feed.FeedItem.manager.purge') @patch('distribution.models.feed.FeedItem.manager.bulk_save') @patch('distribution.models.taxonomy.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_total(self, search_by_field_mock, get_mock, bulk_save_mock, purge_mock): items = self.get_items() sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'total'}) ritems, rerrors = sequencer.execute() purge_mock.assert_called_once_with('news', '*', '*') self.assertEqual(bulk_save_mock.call_count, 1) self.assertEqual(len(bulk_save_mock.call_args[0][0]), 4) self.assertEqual(rerrors, []) self.assertEqual(ritems, [x.get('id') for x in items]) @patch('distribution.models.feed.FeedItem.manager.purge') @patch('distribution.models.feed.FeedItem.manager.bulk_save') @patch('distribution.models.taxonomy.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_chunk(self, search_by_field_mock, get_mock, bulk_save_mock, purge_mock): items = self.get_items() sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'chunk'}) ritems, rerrors = sequencer.execute() purge_mock.assert_called_once_with('news', '2015-03-01T20:55:00.000000Z', '2015-04-02T17:55:00.000000Z') self.assertEqual(bulk_save_mock.call_count, 1) self.assertEqual(len(bulk_save_mock.call_args[0][0]), 4) self.assertEqual(rerrors, []) self.assertEqual(ritems, [x.get('id') for x in items]) @patch('distribution.models.feed.FeedItem.manager.disable_many') @patch('distribution.adapters.models.AMPAdapter.get_amp_version_from_npm', return_value='1.3.0') def test_delete_non_existent_items(self, amp_version_mock, disable_many_mock): app.config['AMP_CACHE_ENABLED'] = False wrong_items = [ str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), ] correct_items = [ str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), ] disable_many_mock.return_value = (wrong_items, ['missing_item'] * len(wrong_items)) items = correct_items + wrong_items sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'delete'}) ritems, rerrors = sequencer.execute() self.assertEqual(list(disable_many_mock.call_args[0][0]), list(items)) for wrong_item_idx in xrange(len(wrong_items)): self.assertEqual(rerrors[wrong_item_idx], { 'id': wrong_items[wrong_item_idx], 'domain': 'news', 'error': 'missing_item'} ) self.assertEqual(set(ritems), set(correct_items)) @patch('distribution.models.feed.FeedItem.manager.disable_many', return_value=([], [])) @patch('distribution.adapters.models.AMPAdapter.get_amp_version_from_npm', return_value='1.3.0') def test_delete(self, amp_version_mock, disable_many_mock): app.config['AMP_CACHE_ENABLED'] = False items = [ str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), str(uuid.uuid1()), ] sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'delete'}) ritems, rerrors = sequencer.execute() self.assertEqual(list(disable_many_mock.call_args[0][0]), list(items)) self.assertEqual(rerrors, []) self.assertEqual(set(items), set(ritems)) @patch('distribution.models.feed.FeedItem.manager.bulk_save') @patch('distribution.models.taxonomy.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_upsert(self, search_by_field_mock, get_mock, bulk_save_mock): items = self.get_items() sequencer = FeedSequencer({'items': items, 'domain': 'news', 'sequence': 'upsert'}) ritems, rerrors = sequencer.execute() self.assertEqual(bulk_save_mock.call_count, 1) self.assertEqual(len(bulk_save_mock.call_args[0][0]), 4) self.assertEqual(rerrors, []) self.assertEqual(ritems, [x.get('id') for x in items])