# -*- coding: utf-8 -*- import json from mock import patch, Mock from werkzeug.exceptions import NotFound from test import utils from distribution import app from . import AuthResourceTestCaseMixin, ResourceTestCase, get_user_mock, get_role_mock MOCK_ID = u'xxxxx' class FeedDetailTestCase(AuthResourceTestCaseMixin, ResourceTestCase): mock_item_id = 'news:ab121241-de46-4776-a2b5-ece310a40aad' mock_item_url = '/api/v1/item/{}/'.format(mock_item_id) mock_item_put_data = json.dumps({'title': 'The new title', 'abstract': 'The new abstract'}) def _get(self): """ Makes an anonymous GET request to the mocked item url. """ return self.app.get(self.mock_item_url) def _auth_get(self, allowed=True): """ Makes GET request with credentials to the mocked item url. If allowed is False, the password check validation will fail. """ patches = self.fake_authentication(allowed) with patches[0], patches[1], patches[2]: return self.get_with_auth(self.mock_item_url) def _put(self, data=None): """ Makes an anonymous PUT request to the mocked item url. """ if data is None: data = self.mock_item_put_data return self.app.put(self.mock_item_url, data=data, content_type='application/json') def _auth_put(self, data=None, allowed=True): """ Makes PUT request with credentials to the mocked item url. If allowed is False, the password check validation will fail. """ if data is None: data = self.mock_item_put_data patches = self.fake_authentication(allowed) with patches[0], patches[1], patches[2]: return self.put_with_auth(self.mock_item_url, data=data, content_type='application/json') def _delete(self): """ Makes an anonymous DELETE request to the mocked item url. """ return self.app.delete(self.mock_item_url) def _auth_delete(self, allowed=True): """ Makes DELETE request with credentials to the mocked item url. If allowed is False, the password check validation will fail. """ patches = self.fake_authentication(allowed) with patches[0], patches[1], patches[2]: return self.delete_with_auth(self.mock_item_url) @patch('distribution.models.feed.FeedItem.localizations', []) @patch('distribution.models.feed.FeedItem.manager.get', autospec=True, return_value=utils.create_feed_item(utils.get_stored_item())) def test_get_existent(self, get_mock): # Unauthorized: res = self._get() self.assertEquals((res.status_code, res.mimetype), (401, 'application/json')) res = self._auth_get(allowed=False) self.assertEquals((res.status_code, res.mimetype), (401, 'application/json')) self.assertFalse(get_mock.called) # Authorized: res = self._auth_get() self.assertEquals((res.status_code, res.mimetype), (200, 'application/json')) item = json.loads(res.data) stored_item = utils.get_stored_item() self.assertEquals(item['title'], stored_item['title']) self.assertEquals(item['publication_datetime'], stored_item['publication_datetime']) self.assertEquals(item['id'], stored_item['id']) feed_id = get_mock.call_args[0][0] self.assertEqual(feed_id, self.mock_item_id) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item(deleted=True)), autospec=True) def test_get_disabled(self, get_mock): res = self._auth_get() self.assertEquals(res.status_code, 410) @patch('distribution.models.feed.FeedItem.manager.get', side_effect=NotFound) def test_get_missing(self, get_mock): res = self._auth_get() self.assertEquals(res.status_code, 404) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item()), autospec=True) @patch('distribution.models.feed.FeedItem.manager.save') @patch('distribution.models.feed.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_put_existent(self, search_by_field_mock, taxonomy_topic_get_mock, save_mock, get_mock): # Unauthorized self.assertEquals(self._put().status_code, 401) self.assertEquals(self._auth_put(allowed=False).status_code, 401) self.assertFalse(save_mock.called) self.assertFalse(get_mock.called) # Authorized res = self._auth_put() self.assertEquals(res.status_code, 200) item = json.loads(res.data) self.assertEquals(item['title'], 'The new title') self.assertEquals(item['abstract'], 'The new abstract') self.assertEquals(item['publication_datetime'], '2015-08-14T14:45:00.000000Z') self.assertEqual(save_mock.call_count, 1) @patch('distribution.models.feed.FeedItem.manager.get', side_effect=NotFound) def test_put_missing(self, get_mock): res = self._auth_put() self.assertEquals(res.status_code, 404) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item(deleted=True)), autospec=True) def test_put_disabled(self, get_mock): res = self._auth_put() self.assertEquals(res.status_code, 410) def test_put_array_input(self): res = self._auth_put(data=json.dumps(['DIS IS INVALID!'])) self.assertEqual(res.status_code, 422) def test_put_invalid_input(self): res = self._auth_put(data='DIS IS INVALID!') self.assertEqual(res.status_code, 422) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item(id=mock_item_id)), autospec=True) @patch('distribution.models.feed.FeedItem.manager.disable_many') @patch('distribution.adapters.models.AMPAdapter.delete') def test_delete_existent(self, amp_delete_mock, disable_many_mock, get_mock): # Unauthorized self.assertEquals(self._delete().status_code, 401) self.assertEquals(self._auth_delete(allowed=False).status_code, 401) self.assertFalse(disable_many_mock.called) self.assertFalse(get_mock.called) res = self._auth_delete() self.assertEquals(res.status_code, 204) disable_many_mock.assert_called_once_with([self.mock_item_id]) amp_delete_mock.assert_called_once_with([self.mock_item_id]) @patch('distribution.models.feed.FeedItem.manager.get', side_effect=NotFound) def test_delete_missing(self, get_mock): res = self._auth_delete() self.assertEquals(res.status_code, 404) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item(deleted=True)), autospec=True) def test_delete_disabled(self, get_mock): res = self._auth_delete() self.assertEquals(res.status_code, 410) class FeedDetailPostTestCase(ResourceTestCase): def _post(self): item_data = utils.get_stored_item() item_data['meta'].pop('domain', None) return self.app.post('/api/v1/item/', data=json.dumps({'item': item_data, 'domain': 'news'}), content_type='application/json') @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item()), autospec=True) @patch('distribution.models.feed.FeedItem.manager.save') @patch('distribution.models.feed.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_post_existent(self, search_by_field_mock, taxonomy_topic_get, save_mock, get_mock): res = self._post() self.assertEquals(res.status_code, 409) self.assertEqual(get_mock.call_count, 1) self.assertEqual(save_mock.call_count, 0) @patch('distribution.models.feed.FeedItem.manager.get', side_effect=NotFound) @patch('distribution.models.feed.FeedItem.manager.save') @patch('distribution.models.feed.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_post_missing(self, search_by_field_mock, taxonomy_topic_get, save_mock, get_mock): res = self._post() self.assertEquals(res.status_code, 201) self.assertEqual(save_mock.call_count, 1) @patch('distribution.models.feed.FeedItem.manager.get', return_value=utils.create_feed_item(utils.get_stored_item(deleted=True)), autospec=True) @patch('distribution.models.feed.FeedItem.manager.save') @patch('distribution.models.feed.FeedItem.manager.delete', autospec=True) @patch('distribution.models.feed.TaxonomyTopic.manager.get', return_value=None) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_post_disabled(self, search_by_field_mock, taxonomy_topic_get, delete_mock, save_mock, get_mock): res = self._post() self.assertEquals(res.status_code, 201) self.assertEqual(save_mock.call_count, 1) def test_put_invalid_schema(self): res = self.app.post('/api/v1/item/', data=json.dumps({'nothing': 'foo', 'invalid': 'news'}), content_type='application/json') self.assertEquals(res.status_code, 422) def test_put_invalid_input(self): res = self.app.post('/api/v1/item/', data='CRASH!', content_type='application/json') self.assertEquals(res.status_code, 422) @patch('distribution.managers.backends.elastic.ElasticBackend._init_indices', Mock()) @patch('distribution.managers.BaseManager.bulk_save') @patch('distribution.managers.sitemap.SitemapURLManager.get_by_field') @patch('distribution.managers.sitemap.SitemapURLManager.search_by_field') class FeedSitemapPostTestCase(ResourceTestCase): def set_mocks_side_effects(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): search_by_field_mock.side_effect = self.search_by_field_mock_side_effect get_by_field_mock.side_effect = self.get_by_field_mock_side_effect bulk_save_mock.side_effect = self.bulk_save_mock_side_effect def validate_mock_calls(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): self.assertEqual(search_by_field_mock.called, True) self.assertEqual(get_by_field_mock.called, True) self.assertEqual(bulk_save_mock.called, True) def search_by_field_mock_side_effect(self, *args): return utils.get_stored_element_children(self.data, args[1]) def get_by_field_mock_side_effect(self, *args): return utils.get_stored_item_by_id(self.data, args[0], args[1]) def bulk_save_mock_side_effect(self, *args): utils.bulk_save_items(self.data, args[0]) def setUp(self): self.data = [] super(FeedSitemapPostTestCase, self).setUp() def _post(self, fixture): request_data = utils.get_json(fixture) return self.app.post('/feed/v1', data=json.dumps(request_data), content_type='application/json') def test_base_case(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): """ Test that the base case with correct data functions correctly """ self.set_mocks_side_effects(search_by_field_mock, get_by_field_mock, bulk_save_mock) request_data = utils.get_json('sitemap/sitemap_localizations_valid_post.json') # Need to run the post two times to update correctly the localization_root self._post('sitemap/sitemap_localizations_valid_post.json') self._post('sitemap/sitemap_localizations_valid_post.json') self.validate_mock_calls(search_by_field_mock, get_by_field_mock, bulk_save_mock) self.assertEqual(len(self.data), 6) self.assertEqual(self.data[-1]['url'], request_data['items'][3]['localizations'][0]['url']) self.assertEqual(self.data[-1]['language'], request_data['items'][3]['localizations'][0]['language']) for element in self.data: self.assertNotIn('localization', element) def test_validate_different_language(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): """ Test that a element, that is going to be stored in distribution, can't have a localization with the same url that an already existent element is DS but have a different language """ self.set_mocks_side_effects(search_by_field_mock, get_by_field_mock, bulk_save_mock) request_data = utils.get_json('sitemap/sitemap_localizations_invalid_different_url.json') self._post('sitemap/sitemap_localizations_valid_post.json') response = json.loads(self._post( 'sitemap/sitemap_localizations_invalid_different_url.json').data) self.validate_mock_calls(search_by_field_mock, get_by_field_mock, bulk_save_mock) self.assertEqual(response['failed'], 1) self.assertEqual(response['succeeded'], 0) msg = 'Could not process item with id {}, an existing element with same URL has a ' \ 'different language'.format(request_data['items'][0]['id']) self.assertEqual(response['errors'][0]['error'], msg) def test_validate_element_already_have_localization(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): """ Test that an new element, that have a parent in DS, can't be stored if the parent already have a localization for the same language """ self.set_mocks_side_effects(search_by_field_mock, get_by_field_mock, bulk_save_mock) request_data = utils.get_json('sitemap/sitemap_localizations_invalid_already_exist.json') self._post('sitemap/sitemap_localizations_valid_post.json') self._post('sitemap/sitemap_localizations_valid_post.json') response = json.loads(self._post( 'sitemap/sitemap_localizations_invalid_already_exist.json').data) self.validate_mock_calls(search_by_field_mock, get_by_field_mock, bulk_save_mock) self.assertEqual(response['failed'], 1) self.assertEqual(response['succeeded'], 0) msg = 'SitemapURL id ({}), Parent id ({}) already have a localization for that language ' \ 'and article'\ .format(request_data['items'][0]['id'], request_data['items'][0]['parent']) self.assertEqual(response['errors'][0]['error'], msg) def test_validate_localization_already_exist(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): """ Test if an element have a localization, and the url of the localization already exist in DS, but have a different language """ self.set_mocks_side_effects(search_by_field_mock, get_by_field_mock, bulk_save_mock) request_data = utils.get_json( 'sitemap/sitemap_localizations_invalid_loc_already_exist.json') self._post('sitemap/sitemap_localizations_valid_post.json') self._post('sitemap/sitemap_localizations_valid_post.json') response = json.loads(self._post( 'sitemap/sitemap_localizations_invalid_loc_already_exist.json').data) self.validate_mock_calls(search_by_field_mock, get_by_field_mock, bulk_save_mock) self.assertEqual(response['failed'], 1) self.assertEqual(response['succeeded'], 0) msg = 'Could not process item with id {}, an existing element with same URL has a ' \ 'different language'.format(request_data['items'][0]['id']) self.assertEqual(response['errors'][0]['error'], msg) def test_cant_update_element(self, search_by_field_mock, get_by_field_mock, bulk_save_mock): """ Test if an element is updated, but the localization does not match, it should generate an error """ self.set_mocks_side_effects(search_by_field_mock, get_by_field_mock, bulk_save_mock) request_data = utils.get_json( 'sitemap/sitemap_localization_invalid_loc.json') self._post('sitemap/sitemap_localizations_valid_post.json') self._post('sitemap/sitemap_localizations_valid_post.json') response = json.loads(self._post( 'sitemap/sitemap_localization_invalid_loc.json').data) self.validate_mock_calls(search_by_field_mock, get_by_field_mock, bulk_save_mock) self.assertEqual(response['failed'], 1) self.assertEqual(response['succeeded'], 0) msg = 'Could not process item with id {}, there is already another element with the same ' \ 'language'.format(request_data['items'][0]['id']) self.assertEqual(response['errors'][0]['error'], msg) @patch('distribution.managers.auth.APIUserManager.get', return_value=get_user_mock()) @patch('distribution.managers.auth.RoleManager.get', return_value=get_role_mock()) @patch('flask_api_auth.models.APIUser.verify_password', return_value=True) class ITFeedTestCase(AuthResourceTestCaseMixin, ResourceTestCase): ext_search_result_mock = dict(items=[utils.create_feed_item(item) for item in utils.get_stored_data()]) @patch('distribution.adapters.utils.get_adapter') @patch('distribution.models.feed.FeedItem.manager.extended_search', autospec=True) @patch('distribution.models.feed.TaxonomyTopic.manager.mget', return_value=[]) def test_format_resolution(self, taxonomy_mock, search_mock, get_adapter_mock, verify_password_mock, role_mock, user_mock): """ Test: Should resolve to the appropriate adapter """ get_adapter_mock.return_value.render.return_value = 'mocked content' self.get_with_auth('/feed/v1/from_path/') get_adapter_mock.assert_called_once_with('from_path') get_adapter_mock.reset_mock() self.get_with_auth('/feed/v1/?format=from_qs') get_adapter_mock.assert_called_once_with('from_qs') get_adapter_mock.reset_mock() self.get_with_auth('/feed/v1/', headers={ 'accept': 'application/vnd.natgeo+from_accept' }) get_adapter_mock.assert_called_once_with('from_accept') get_adapter_mock.reset_mock() # Test: A format specified through the format query parameter # precedes anything set in the Accept header self.get_with_auth('/feed/v1/?format=from_qs', headers={ 'accept': 'application/vnd.natgeo+from_accept' }) get_adapter_mock.assert_called_once_with('from_qs') get_adapter_mock.reset_mock() # Test: A format specified through the path segment precedes # anything set in the query parameter or Accept header self.get_with_auth('/feed/v1/from_path/?format=from_qs', headers={ 'accept': 'application/vnd.natgeo+from_accept' }) get_adapter_mock.assert_called_once_with('from_path') def test_get_unsupported_type_format(self, verify_password_mock, role_mock, user_mock): """ Test: Should respond with code 404 if format is not rss or json """ self.assertEqual(self.get_with_auth('/feed/v1/bad/').status_code, 404) self.assertEqual( self.get_with_auth('/feed/v1/?format=bad').status_code, 404) self.assertEqual(self.get_with_auth('/feed/v1/', headers={ 'accept': 'application/vnd.natgeo+bad' }).status_code, 404) @patch('distribution.models.feed.FeedItem.manager.extended_search') @patch('distribution.models.feed.TaxonomyTopic.manager.mget', return_value=[]) def test_get_default_type_format(self, taxonomy_topic_mget_mock, ext_search_mock, verify_password_mock, role_mock, user_mock): """ Test: Should return the rss feed if format is not supplied """ ext_search_mock.return_value = self.ext_search_result_mock self.assertEqual(self.get_with_auth('/feed/v1/').mimetype, 'text/xml') self.assertEqual(ext_search_mock.call_count, 1) @patch('distribution.models.feed.FeedItem.manager.extended_search') def test_get_default_type_format_forbidden(self, ext_search_mock, verify_password_mock, role_mock, user_mock): """ Test: Should return 401 if apiuser/apiuser are not supplied """ ext_search_mock.return_value = self.ext_search_result_mock response = self.app.get('/feed/v1/') self.assertEqual(ext_search_mock.call_count, 0) self.assertEqual(response.status_code, 401) @patch('distribution.models.feed.FeedItem.manager.extended_search') @patch('distribution.models.feed.TaxonomyTopic.manager.mget', return_value=[]) def test_get_rss_type_format(self, taxonomy_topic_mget_mock, ext_search_mock, verify_password_mock, role_mock, user_mock): """ Test: Should return the rss if format is rss """ ext_search_mock.return_value = self.ext_search_result_mock res = self.get_with_auth('/feed/v1/?format=rss') self.assertEquals(res.status_code, 200) self.assertEqual(res.mimetype, 'text/xml') self.assertEqual(ext_search_mock.call_count, 1) @patch('distribution.models.feed.FeedItem.manager.extended_search') @patch('distribution.models.feed.TaxonomyTopic.manager.mget', return_value=[]) def test_get_rss_embedded_type_format(self, taxonomy_topic_mget_mock, ext_search_mock, verify_password_mock, role_mock, user_mock): """ Test: Should return rss_embedded format """ ext_search_mock.return_value = self.ext_search_result_mock res = self.get_with_auth('/feed/v1/rss_embedded/') self.assertEquals(res.status_code, 200) self.assertEqual(res.mimetype, 'text/xml') self.assertEqual(ext_search_mock.call_count, 1) @patch('distribution.models.feed.FeedItem.localizations', []) @patch('distribution.models.feed.FeedItem.manager.extended_search') @patch('distribution.models.feed.TaxonomyTopic.manager.mget', return_value=[]) def test_get_json_type_format(self, taxonomy_topic_mget_mock, ext_search_mock, verify_password_mock, role_mock, user_mock): """ Test: Should return the json if format is json """ ext_search_mock.return_value = self.ext_search_result_mock res = self.get_with_auth('/feed/v1/?format=json') self.assertEquals(res.status_code, 200) self.assertEqual(res.mimetype, 'application/json') items = json.loads(res.data) self.assertEquals(len(items), 1) self.assertEqual(ext_search_mock.call_count, 1) @patch('distribution.models.feed.TaxonomyTopic.manager.get', autospec=True) @patch('distribution.models.feed.FeedItem.manager.purge', autospec=True) @patch('distribution.models.feed.FeedItem.manager.bulk_save', autospec=True) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_success_json_post(self, search_by_field_mock, bulk_save_mock, purge_mock, taxonomy_mock, verify_password_mock, role_mock, user_mock): response = self.post_with_auth('/feed/v1/', **{ 'data': json.dumps(utils.get_consumer_post_data()), 'content_type': 'application/json' }) data = json.loads(response.data) self.assertEquals(response.status_code, 200) self.assertEqual(data['succeeded'], 1) self.assertEqual(data['failed'], 0) self.assertEqual(data['errors'], []) self.assertEqual(data['items'], ['news:ab121241-de46-4776-a2b5-ece310a40aad']) @patch('distribution.models.feed.FeedItem.manager.save', side_effect=ValueError) def test_unsuccessful_json_post(self, feed_mock, verify_password_mock, role_mock, user_mock): response = self.post_with_auth('/feed/v1/', data={}, content_type='application/json') self.assertEqual(response.status_code, 422) def test_invalid_json_post(self, verify_password_mock, role_mock, user_mock): """ Test: Should respond with code 422 if the payload has an invalid format """ response = self.post_with_auth('/feed/v1/', **{ 'data': json.dumps({'data': 'bad_data'}), 'content_type': 'application/json'}) self.assertEqual(response.status_code, 422) @patch('distribution.models.feed.TaxonomyTopic.manager.get', autospec=True) @patch('distribution.models.feed.FeedItem.manager.purge', autospec=True) @patch('distribution.models.feed.FeedItem.manager.bulk_save', autospec=True) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_json_post_feed_is_saved(self, search_by_field_mock, bulk_save_mock, purge_mock, taxonomy_mock, verify_password_mock, role_mock, user_mock): """ Test: Posting a feed to /feed/create/ should save it """ data = utils.get_consumer_post_data() data['items'][0]['id'] = 'myid' response = self.post_with_auth('/feed/v1/', **{ 'data': json.dumps(data), 'content_type': 'application/json' }) self.assertEqual(response.status_code, 200) self.assertEqual(purge_mock.call_count, 1) self.assertEqual(bulk_save_mock.call_count, 1) result = json.loads(response.data) item = bulk_save_mock.call_args[0][0][0] self.assertEqual(item.key, data['items'][0]['id']) self.assertEqual(result['items'][0], data['items'][0]['id']) self.assertEqual(result['succeeded'], 1) self.assertEqual(result['failed'], 0) self.assertEqual(result['errors'], []) @patch('distribution.models.feed.TaxonomyTopic.manager.get', autospec=True) @patch('distribution.models.feed.FeedItem.manager.purge', autospec=True) @patch('distribution.models.feed.FeedItem.manager.bulk_save', autospec=True) @patch('distribution.models.feed.FeedItem.manager.search_by_field', side_effect=NotFound) def test_json_post_feed_not_saved(self, search_by_field_mock, bulk_save_mock, purge_mock, taxonomy_mock, verify_password_mock, role_mock, user_mock): """ Test: Posting a feed to /feed/create/ should save it """ data = utils.get_consumer_post_data() data['items'][0]['id'] = 'myid' del data['items'][0]['content_type'] response = self.post_with_auth('/feed/v1/', **{ 'data': json.dumps(data), 'content_type': 'application/json' }) self.assertEqual(response.status_code, 200) self.assertEqual(purge_mock.call_count, 1) self.assertEqual(bulk_save_mock.call_count, 1) result = json.loads(response.data) self.assertEqual(result['errors'][0]['id'], data['items'][0]['id']) self.assertEqual(result['errors'][0]['content_type'], 'N/A') self.assertEqual(result['errors'][0]['domain'], 'news') self.assertEqual(result['succeeded'], 0) self.assertEqual(result['failed'], 1) self.assertEqual(result['errors'][0]['error'], {u'content_type': [u'required field']}) def test_get_by_invalid_datetime_arguments(self, verify_password_mock, role_mock, user_mock): """ Test that the feed request can only have one datetime filter at the same time """ invalid_cases = ( 'publication_datetime=1d&meta.last_update=1d', 'publication_datetime__from=2015-08-13T18:30:02Z&last_modified_datetime=1d', 'meta.last_update__from=2015-08-13T18:30:02Z&last_modified_datetime__from=2015-08-13T18:30:02Z' ) for invalid_case in invalid_cases: response = self.get_with_auth( '/feed/v1/?{}'.format(invalid_case) ) self.assertEqual(response.status_code, 400) self.assertIn('Only one datetime parameter is valid', response.data) def test_invalid_datetime_request(self, verify_password_mock, role_mock, user_mock): """ Test the three valid datetime arguments """ datetime_filters = ['publication_datetime', 'last_modified_datetime', 'meta.last_update'] for datetime_filter in datetime_filters: self.invalid_from_filter(datetime_filter) self.invalid_to_filter(datetime_filter) self.invalid_hour_day_filter(datetime_filter) self.invalid_to_missing_from_filter(datetime_filter) @patch('distribution.models.feed.FeedItem.manager.search', autospec=True) def invalid_from_filter(self, datetime_filter, ext_search_mock): """ Should return code status 400 and Invalid format. Datetime format If the from date is not correctly formatted """ response = self.get_with_auth( '/feed/v1/?{}__from=05/05/2015'.format(datetime_filter) ) self.assertEqual(response.status_code, 400) self.assertIn('Invalid format. Datetime format RFC3339', response.data) @patch('distribution.models.feed.FeedItem.manager.extended_search') def invalid_to_filter(self, datetime_filter, ext_search_mock): """ Should return code status 400 and Invalid format. Datetime format If the to date is not correctly formatted """ response = self.get_with_auth( '/feed/v1/?{}__to=05/05/2015'.format(datetime_filter) ) self.assertEqual(response.status_code, 400) self.assertIn('Invalid format. Datetime format RFC3339', response.data) @patch('distribution.models.feed.FeedItem.manager.extended_search') def invalid_hour_day_filter(self, datetime_filter, ext_search_mock): """ Should return code status 400 and Invalid format. Range If the hour/day filter is not valid or the range is not appropriate """ invalid_filters = ['12hours', '13days', '25h', '104d'] for invalid_filter in invalid_filters: response = self.get_with_auth( '/feed/v1/?{}={}'.format(datetime_filter, invalid_filter) ) self.assertEqual(response.status_code, 400) self.assertIn('Invalid format', response.data) @patch('distribution.models.feed.FeedItem.manager.extended_search') def invalid_to_missing_from_filter(self, datetime_filter, ext_search_mock): """ Should return code status 400 and Missing query argument If the query have a __to argument but not a __from one """ response = self.get_with_auth( '/feed/v1/?{}__to=2015-08-13T18:30:02Z'.format(datetime_filter) ) self.assertEqual(response.status_code, 400) self.assertIn('Missing query argument', response.data) @patch('distribution.models.feed.FeedItem.manager.disable_many', autospec=True) @patch('distribution.adapters.models.AMPAdapter.get_amp_version_from_npm', return_value='1.3.0') def test_valid_json_schema_delete(self, amp_version_mock, delete_mock, verify_password_mock, role_mock, user_mock): """ Test: Should response with code 200 """ data = utils.get_consumer_delete_data() error_message = 'missing item' delete_mock.return_value = ([data['items'][0]], [error_message]) app.config['AMP_CACHE_ENABLED'] = False response = self.app.delete( '/feed/v1/', **{ 'data': json.dumps(data), 'content_type': 'application/json' } ) self.assertEqual(response.status_code, 200) self.assertEqual(delete_mock.call_count, 1) parsed_response = json.loads(response.data) self.assertEqual(parsed_response['failed'], 1) self.assertEqual(parsed_response['succeeded'], 1) self.assertEqual(parsed_response['errors'], [ {'domain': data['domain'], 'id': data['items'][0], 'error': error_message} ]) self.assertEqual(parsed_response['items'], [data['items'][1]]) @patch('distribution.models.feed.FeedItem.manager.disable_many', autospec=True) @patch('distribution.adapters.models.AMPAdapter.get_amp_version_from_npm', return_value='1.3.0') def test_validate_response_delete(self, amp_version_mock, delete_mock, verify_password_mock, role_mock, user_mock): """ Validate the response of a delete request to /feed/v1 """ data = utils.get_consumer_delete_data() error_message = 'missing item' delete_mock.return_value = ([data['items'][0], data['items'][1]], [error_message] * 2) app.config['AMP_CACHE_ENABLED'] = False response = self.app.delete( '/feed/v1/', **{ 'data': json.dumps(data), 'content_type': 'application/json' } ) self.assertEqual(response.status_code, 200) self.assertEqual(delete_mock.call_count, 1) parsed_response = json.loads(response.data) self.assertEqual(parsed_response['failed'], 2) self.assertEqual(parsed_response['succeeded'], 0) self.assertEqual(parsed_response['errors'], [ {'domain': data['domain'], 'id': data['items'][0], 'error': error_message}, {'domain': data['domain'], 'id': data['items'][1], 'error': error_message}, ]) self.assertEqual(parsed_response['items'], []) def test_invalid_json_schema_delete(self, verify_password_mock, role_mock, user_mock): """ Test: Should response with code 422 if the payload has an invalid schema """ self.assertEqual(self.delete_with_auth( '/feed/v1/', **{ 'data': json.dumps({'data': 'bad_data'}), 'content_type': 'application/json' }).status_code, 422) def test_invalid_limit_filter(self, verify_password_mock, role_mock, user_mock): """ Test: Should respond with code 400 if limit is not valid """ response = self.get_with_auth('/feed/v1/?limit=arnold') self.assertEqual(response.status_code, 400) def test_invalid_page_filter(self, verify_password_mock, role_mock, user_mock): """ Test: Should respond with code 400 if page is not valid """ response = self.get_with_auth('/feed/v1/?page=arnold') self.assertEqual(response.status_code, 400) def test_invalid_domain_filter(self, verify_password_mock, role_mock, user_mock): """ Test: Should respond with code 400 if domain is not valid """ response = self.get_with_auth('/feed/v1/?domain=1234') self.assertEqual(response.status_code, 400) @patch('distribution.managers.auth.APIUserManager.get', Mock(return_value=get_user_mock())) @patch('distribution.managers.auth.RoleManager.get', Mock(return_value=get_role_mock())) @patch('flask_api_auth.models.APIUser.verify_password', Mock(return_value=True)) class ValidateFeedTestCase(AuthResourceTestCaseMixin, ResourceTestCase): @patch('distribution.models.feed.TaxonomyTopic.manager.get', Mock(autospec=True)) def test_successful_valid_json(self): """ Test: Should return succeeded, failed, items and errors structure. Validates all the properties except alt_bodies. """ response = self.post_with_auth('/feed/v1/validate/', **{ 'data': json.dumps(utils.get_consumer_validate_data()), 'content_type': 'application/json' }) result = json.loads(response.data) self.assertIn('succeeded', result) self.assertEqual(result['succeeded'], 6) self.assertIn('failed', result) self.assertEqual(result['failed'], 1) self.assertIn('items', result) self.assertIn('errors', result) @patch('distribution.models.feed.TaxonomyTopic.manager.get', Mock(autospec=True)) def test_successful_valid_json_strict(self): """ Test: Should return succeeded, failed, items and errors structure. Validates all the properties, including alt_bodies. """ response = self.post_with_auth('/feed/v1/validate/?strict=true', **{ 'data': json.dumps(utils.get_consumer_validate_data()), 'content_type': 'application/json' }) result = json.loads(response.data) self.assertIn('succeeded', result) self.assertEqual(result['succeeded'], 5) self.assertIn('failed', result) self.assertEqual(result['failed'], 2) self.assertIn('items', result) self.assertIn('errors', result) def test_invalid_strict_parameter(self): """ Test: Should respond with code 400 if the strict parameter is not valid. """ response = self.post_with_auth('/feed/v1/validate/?strict=123', **{ 'data': None, 'content_type': 'application/json' }) self.assertEqual(response.status_code, 400) def test_unsuccessful_invalid_json(self): """ Test: Should return succeeded, failed, items and errors structure. Check the post schema validation. """ response = self.post_with_auth('/feed/v1/validate/', **{ 'data': json.dumps({ 'domain': 'load', 'model': 'article', 'sequence': 'validate' }), 'content_type': 'application/json' }) result = json.loads(response.data) self.assertIn('succeeded', result) self.assertEqual(result['succeeded'], 0) self.assertIn('failed', result) self.assertEqual(result['failed'], 1) self.assertIn('items', result) self.assertIn('errors', result) def test_unsuccessful_invalid_payload(self): """ Test: Should respond with code 422 if the request doesn't contain a valid payload """ response = self.post_with_auth('/feed/v1/validate/', **{ 'data': None, 'content_type': 'application/json' }) self.assertEqual(response.status_code, 422)