"""API Testing class and all its configs.""" from datetime import datetime, timedelta import json import random import string import unittest from unittest.mock import Mock, patch import pytest from parameterized import parameterized import requests import requests_mock from feed_status.models import ingestion_feed from feed_status.models.orm import feed_status from feed_status.resources.ingestion_feeds import IngestionFeedComment, \ IngestionFeedCommentList, IngestionFeedStatus from tests.connectors import config STATUSES = [ ingestion_feed.INGESTION_STATUS_POPULATED_RAW_TABLE, ingestion_feed.INGESTION_STATUS_INGESTED, ingestion_feed.INGESTION_STATUS_NOT_INGESTED] TEST_DATE = '2021-08-08' TEST_FEED_ID = 'itunes_tickets' def mock_get_ingestion_comment_list(date): """Mock FEED_INGESTION_TABLE query response.""" return {'Items': [{ 'date': date + random.choice(list(map( lambda s: '.' + s, config.units_data_feeds))), 'comment': random.choice(string.ascii_uppercase) + ( ''.join(random.choice(string.ascii_lowercase) for _ in range(20))) + '.'} for i in range(10)]} def mock_get_feed_comment(feed_id, date): """Mock FEED_INGESTION_TABLE query response.""" return {'Item': { 'date': date + '.' + feed_id, 'comment': random.choice(string.ascii_uppercase) + (''.join( random.choice(string.ascii_lowercase) for _ in range(10))) + '.' }} def mock_get_ingestion_status(date: datetime): """Mock FEED_INGESTION_TABLE query response.""" return {'Items': [{'status': random.choice(STATUSES), 'date': (date.today() - timedelta(days=i)).strftime( '%Y-%m-%d')} for i in range(119)]} def mock_get_feed_unit(feed_id, licensor): """Mock SF get() function for FeedUnits class.""" return [{'units': 0, 'feed_id': feed_id, 'licensor': licensor}] def mock_get_all_feeds_units(): """Mock SF get_units_for_all_feeds() function for AllFeedsUnits class.""" return [{'units': '67879809', 'a': 435, 'b': 2436}] class TestAPIEndpoints(unittest.TestCase): """API Testing Class.""" @parameterized.expand([config.comment_endpoints[0]]) def test_ingestion_comment_list(self, endpoint): """Test ingestion statuses endpoints.""" feed_status.FEED_INGESTION_TABLE.query = Mock() feed_status.FEED_INGESTION_TABLE.query.return_value = \ mock_get_ingestion_comment_list(datetime.today(). strftime('%Y-%m-%d')) final_endpoint = config.LINK + endpoint comment_list = IngestionFeedCommentList() mock_response = json.dumps(comment_list.get()[0]) with requests_mock.Mocker() as mocker: mocker.register_uri('GET', final_endpoint, text=mock_response) response = requests.get(final_endpoint) self.assertEqual(response.status_code, 200) self.assertIn(datetime.today().strftime('%Y-%m-%d'), response.content.decode('utf-8')) self.assertEqual(len(response.json().get('comments'). get(random.choice(config.units_data_feeds)). get(datetime.today(). strftime('%Y-%m-%d'))), 22) @parameterized.expand(config.status_endpoints) def test_status_endpoints(self, endpoint): """Test ingestion statuses endpoints.""" feed_status.FEED_INGESTION_TABLE.query = Mock() feed_status.FEED_INGESTION_TABLE.query.return_value = \ mock_get_ingestion_status(datetime.today()) feed_id = config.units_data_feeds[0] final_endpoint = config.LINK + endpoint.format(feed_id=feed_id) status = IngestionFeedStatus() mock_response = json.dumps(status.get(feed_id)) with requests_mock.Mocker() as mocker: mocker.register_uri('GET', final_endpoint, text=mock_response) response = requests.get(final_endpoint) self.assertEqual(response.status_code, 200) response_json = response.json() self.assertEqual(response_json.get('data')[0]. get('feed_id'), feed_id) self.assertIn(datetime.today().strftime('%Y-%m-%d'), response_json.get('data')[0].get('status')) self.assertIn(response_json.get('data')[0]. get('feed_id'), config.units_data_feeds) self.assertNotIn('youtube_asset_demographics', response_json.get('data')[0]. get('feed_id')) date_to_test = ((datetime.today() - timedelta(days=25)) .strftime('%Y-%m-%d')) self.assertIn( response_json['data'][0]['status'][date_to_test], STATUSES) @parameterized.expand([config.comment_endpoints[1]]) def test_feed_comment(self, endpoint): """Test ingestion feed comment endpoint.""" feed_status.FEED_INGESTION_TABLE.get_item = Mock() for feed_id in config.units_data_feeds: feed_status.FEED_INGESTION_TABLE.get_item.return_value = \ mock_get_feed_comment(feed_id, TEST_DATE) final_endpoint = config.LINK + endpoint.format(feed_id=feed_id, date=TEST_DATE) status = IngestionFeedComment() mock_response = json.dumps(status.get(feed_id, TEST_DATE)) with requests_mock.Mocker() as mocker: mocker.register_uri('GET', final_endpoint, text=mock_response) response = requests.get(final_endpoint) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()[0].get('feed_id'), feed_id) self.assertEqual(response.json()[0].get('date'), TEST_DATE) self.assertEqual(len(response.json()[0].get('comment')), 12) @parameterized.expand(config.units_data_feeds) @patch('feed_status.connectors.snowflake.get_units', mock_get_feed_unit) @pytest.mark.skip(reason='Efedorov: Snowflake turned off') def test_get_units(self, feed_id): """Test /feed_units/{feed_id}/{licensor} under the hood function. Test if the function provides correct data taken from SnowFlake. Args: feed_id (str): feed id. """ feeds = Mock() for licensor in config.licensors: response = feeds.get(feed_id, licensor) self.assertEqual(response[1], 200) self.assertEqual(response[0].get('data'). get('units')[0].get('units'), 0) @patch('feed_status.connectors.snowflake.get_units_for_all_feeds', mock_get_all_feeds_units) @pytest.mark.skip(reason='Snowflake turned off') def test_all_feeds_units(self): """Test /all_feeds_units under the hood function. Test if the function provides correct data taken from SnowFlake. """ feeds = Mock() response = feeds.get() self.assertEqual(response[1], 200) self.assertEqual(response[0].get('data')[0]. get('units')[0].get('units'), 0) if __name__ == '__main__': TestAPIEndpoints().run()