""" Test actuals data logic output. This testing module sets up the flask application environment with enough data to run integration tests on actuals and analytics data transition logic. """ from collections import defaultdict from copy import copy from datetime import date from datetime import datetime from datetime import timedelta from itertools import repeat import json from unittest.mock import Mock from unittest.mock import patch from pytest import fixture from tests_integration import conftest from api.models import advances from api.models import cable_revenue from api.models import manual_adjustments from api.models import projections from api.models import release from api.models import term_license from api.utils import aurora from application import app import config START_DATE = date(2001, 1, 1) END_DATE = date(2001, 2, 28) JAN_DAYS = 31 JAN_TO_FEB_DAYS = 59 ACTUALS_UPC = '999999999991' ANALYTICS_UPC = '999999999992' TRANSITION_UPC = '999999999993' TRANSACTION_TYPE_ID = 42 COUNTRY_ID = 1 STORE_ID = 36 @fixture(scope='module') def client( actuals_accounting_revenue, analytics_accounting_revenue, transition_accounting_revenue, actuals_client_amount, analytics_client_amount, transition_client_amount): """Flask test client with database fixture setup. Yields: Flask: API flask application object. """ try: db_name = datetime.now().strftime('integrationtest_%Y%m%d_%H%M%S') table_names = ('accounting_revenue', 'client_amount', 'stores') table_data = { 'accounting_revenue': actuals_accounting_revenue + analytics_accounting_revenue + transition_accounting_revenue, 'client_amount': actuals_client_amount + analytics_client_amount + transition_client_amount} with aurora.context() as (cursor, connection): cursor.execute(f'CREATE DATABASE {db_name}') for table_name in table_names: cursor.execute( f""" CREATE TABLE {db_name}.{table_name} LIKE film_transparency.{table_name} """) if table_name == 'stores': continue rows = table_data[table_name] num_of_columns = len(rows[0]) query_placeholders = ','.join(repeat('%s', num_of_columns)) cursor.executemany( f""" INSERT INTO {db_name}.{table_name} VALUES ({query_placeholders}) """, rows) cursor.execute( f""" INSERT INTO {db_name}.stores SELECT * FROM film_transparency.stores """) original_config = copy(config.DATASTORE) config.DATASTORE['db'] = db_name app.testing = True test_app = app.test_client() yield test_app finally: config.DATASTORE = copy(original_config) with aurora.context() as (cursor, connection): cursor.execute(f'DROP DATABASE {db_name}') @fixture(scope='module') def actuals_accounting_revenue(): """Generate actuals accounting_revenue data.""" return generate_data(ACTUALS_UPC, JAN_TO_FEB_DAYS, 2) @fixture(scope='module') def actuals_client_amount(): """Generate actuals client_amount data.""" return generate_data(ACTUALS_UPC, JAN_TO_FEB_DAYS, 1, True) @fixture(scope='module') def analytics_accounting_revenue(): """Generate analytics accounting_revenue data.""" return generate_data(ANALYTICS_UPC, JAN_DAYS, 1) @fixture(scope='module') def analytics_client_amount(): """Generate analytics client_amount data.""" return generate_data(ANALYTICS_UPC, JAN_TO_FEB_DAYS, 1, True) @fixture(scope='module') def transition_accounting_revenue(): """Generate transition accounting_revenue data.""" return generate_data(TRANSITION_UPC, JAN_DAYS, 3) @fixture(scope='module') def transition_client_amount(): """Generate transition client_amount data.""" return generate_data(TRANSITION_UPC, JAN_TO_FEB_DAYS, 2, True) def generate_data(upc, days, amount, client_amount=False): """Create data based on days and starting amount. Args: upc (str): upc column value. days (list): list of dates to create rows with. amount (int): accumulating amount. client_amount (bool): indicating client_amount table to remove store ID column. Returns: list: rows of test data for accounting_revenue or client_amount table. """ mock_data = [] for i in range(days): mock_date = START_DATE + timedelta(days=i) mock_data.append( [upc, mock_date, float(amount), TRANSACTION_TYPE_ID, COUNTRY_ID] ) if not client_amount: mock_data[-1].append(STORE_ID) return mock_data def expected_actuals_growth(actuals_accounting_revenue): """Expected actuals growth data.""" growth = [] for i, row in enumerate(actuals_accounting_revenue, start=1): growth.append([str(row[1]), row[2] * i]) return growth def expected_analytics_growth(analytics_client_amount): """Expected analytics growth data.""" growth = [] for i, row in enumerate(analytics_client_amount, start=1): growth.append([str(row[1]), row[2] * i]) return growth def expected_transition_growth( transition_client_amount, transition_accounting_revenue): """Generate transition growth data from both transition tables.""" growth = [] client_rows = transition_client_amount accounting_rows = transition_accounting_revenue for i, client_row in enumerate(client_rows): row_date = str(client_row[1]) growth_value = client_row[2] if i < len(accounting_rows): growth_value = accounting_rows[i][2] if growth: growth.append([row_date, growth[-1][1] + growth_value]) else: growth.append([row_date, growth_value]) return growth @patch('api.utils.cms.get_client') def test_windowing_output( get_client, client, monkeypatch, windows_converted, window_model_raw): """Test windowing API output. This asserts only the windows output and doesn't test the output of windows and revenues together. """ def val(*args): return defaultdict(dict) def advances_val(*args): return (defaultdict(dict), False) monkeypatch.setattr(advances, 'fetch_advances', value=advances_val) monkeypatch.setattr( cable_revenue, 'fetch_all_time_buckets', value=lambda x: {'raw': tuple(), 'bucket': defaultdict(dict)}) monkeypatch.setattr( manual_adjustments, 'fetch_manual_adjustments', value=val) monkeypatch.setattr( projections, 'fetch_all_time_buckets', value=lambda *x: {'raw': tuple(), 'series': defaultdict(dict)}) monkeypatch.setattr( release, 'get_vendor_and_release', value=lambda x: ('12345', '4567890')) monkeypatch.setattr(term_license, 'fetch_term_licenses', value=val) film_room_client = Mock() film_room_client.get_windows.return_value = windows_converted get_client.return_value = film_room_client response = client.get(f'/release/{ANALYTICS_UPC}/profit-loss') response_data = json.loads(response.data, encoding='utf-8') assert response_data['windows'] == conftest.stringify_api_output( window_model_raw)