"""Lambda test module."""
import base64
from unittest.mock import patch
import datetime
import decimal

import pytest
import pandas as pd
import config
from src import app


@pytest.fixture
def kinesis_record():
    """Fixture that returns MySQL log record from Kinesis stream."""
    return [
        {
            'id': 2845,
            'entry_date': pd.Timestamp('2010-05-25 11:45:54'),
            'publisher_id': 1515,
            'track_id': None,
            'cut_date': datetime.date(2010, 5, 20),
            'cash_date': None,
            'check_payable': 'test',
            'check_no': '12499',
            'check_amt': decimal.Decimal('145.000000'),
            'comments': None,
            'paidfor_type': None,
            'check_amt_OLD': None,
            'paidfor_period_id': 127,
            'last_modified_by': 0,
            'user_type': None,
            'log_action': 'update',
            'ts': 1546419205,
            'log_timestamp': pd.Timestamp('2019-01-02 03:53:25')
        }
    ]


@pytest.fixture
def event():
    """Fixture that returns lambda event."""
    return {'Records': [{
            'kinesis': {
                'kinesisSchemaVersion': '1.0',
                'partitionKey': 'art_relations',
                'sequenceNumber':
                '49591584225074164406951083979870858879182018665261301778',
                'data': 'eyJkYXRhYmFzZSI6ImFydF9yZWxhdGlvbnMiLCJ0YWJsZSI6InB1Y'
                'mxpc2hlcnNfY2hlY2tzcGFpZCIsInR5cGUiOiJ1cGRhdGUiLCJ0cyI6MTU0Nj'
                'QxOTIwNSwieGlkIjoxMDUzMTM3LCJjb21taXQiOnRydWUsImRhdGEiOnsiaWQ'
                'iOjI4NDUsImVudHJ5X2RhdGUiOiIyMDEwLTA1LTI1IDExOjQ1OjU0IiwicHVi'
                'bGlzaGVyX2lkIjoxNTE1LCJ0cmFja19pZCI6bnVsbCwiY3V0X2RhdGUiOiIyM'
                'DEwLTA1LTIwIiwiY2FzaF9kYXRlIjpudWxsLCJjaGVja19wYXlhYmxlIjoidG'
                'VzdCIsImNoZWNrX25vIjoiMTI0OTkiLCJjaGVja19hbXQiOjE0NSwiY29tbWV'
                'udHMiOm51bGwsInBhaWRmb3JfdHlwZSI6bnVsbCwiY2hlY2tfYW10X09MRCI6'
                'bnVsbCwicGFpZGZvcl9wZXJpb2RfaWQiOjEyNywibGFzdF9tb2RpZmllZF9ie'
                'SI6bnVsbCwidXNlcl90eXBlIjpudWxsfSwib2xkIjp7ImNoZWNrX2FtdCI6MT'
                'AwMDAwMDAwMDAwMH19',
                'approximateArrivalTimestamp': 1546419330.043},
            'eventSource': 'aws:kinesis',
            'eventVersion': '1.0',
            'eventID': 'shardId-000000000001:495915842250741644069510839798708'
            '58879182018665261301778',
            'eventName': 'aws:kinesis:record',
            'invokeIdentityArn': 'arn:aws:iam::408829442480:role/lambda',
            'awsRegion': 'us-east-2',
            'eventSourceARN': 'eventsrcARN'
            }]}


@patch('src.app.os')
@patch('src.app.upload_to_s3')
@patch('src.app.create_parquet_file')
def test_handler(
        mock_create_parquet, mock_upload_to_s3, mock_os, kinesis_record,
        event):
    """Test handler function."""
    app.handler(event, None)

    mock_create_parquet.assert_called_with(kinesis_record)
    mock_upload_to_s3.assert_called_with(
        mock_create_parquet.return_value,
        'art_relations',
        'publishers_checkspaid')
    mock_os.remove.assert_called_once_with(
        config.TEMP_FILE_PATH.format(mock_create_parquet.return_value))

    assert mock_create_parquet.return_value


@pytest.fixture
def kinesis_record_for_release_status():
    """Fixture that returns MySQL log record from Kinesis stream."""
    return [
        {
            'release_status_id': 6571265,
            'release_id': 123,
            'status': 'transfer_to_content',
            'date': datetime.date(2019, 11, 5),
            'changed_by': 45,
            'changed_by_type': 'oa',
            'upc_REMOVED': 123456,
            'updated_timestamp': pd.Timestamp('2019-11-08 08:45:51'),
            'ts': 1573220751,
            'log_action': 'insert',
            'log_timestamp': pd.Timestamp('2019-11-08 08:45:51')
        }
    ]


@pytest.fixture
def event_for_release_status():
    """Fixture that returns lambda event."""
    return {'Records': [{
            'kinesis': {
                'kinesisSchemaVersion': '1.0',
                'partitionKey': 'art_relations',
                'sequenceNumber': '49601197027725910978478934357733'
                                  '188575951325242413023554',
                'data': 'eyJkYXRhYmFzZSI6ImFydF9yZWxhdGlvbnMiLCJ0YW'
                        'JsZSI6InJlbGVhc2Vfc3RhdHVzIiwidHlwZSI6Imlu'
                        'c2VydCIsInRzIjoxNTczMjIwNzUxLCJ4aWQiOjMxNT'
                        'c2LCJjb21taXQiOnRydWUsImRhdGEiOnsicmVsZWFz'
                        'ZV9zdGF0dXNfaWQiOjY1NzEyNjUsInJlbGVhc2VfaW'
                        'QiOjEyMywic3RhdHVzIjoidHJhbnNmZXJfdG9fY29u'
                        'dGVudCIsImRhdGUiOiIyMDE5LTExLTA1IiwiY2hhbm'
                        'dlZF9ieSI6NDUsImNoYW5nZWRfYnlfdHlwZSI6Im9h'
                        'IiwidXBjX1JFTU9WRUQiOjEyMzQ1NiwidXBkYXRlZF'
                        '90aW1lc3RhbXAiOiIyMDE5LTExLTA4IDEzOjQ1OjUx'
                        'In19',
                'approximateArrivalTimestamp': 1573220752.071
            },
            'eventSource': 'aws:kinesis',
            'eventVersion': '1.0',
            'eventID': 'shardId-000000000020:4960119702772591097847'
                       '8934357733188575951325242413023554',
            'eventName': 'aws:kinesis:record',
            'invokeIdentityArn': 'arn:aws:iam::408829442480:role/lambda',
            'awsRegion': 'us-east-2',
            'eventSourceARN': 'eventsrcARN'
            }]}


@patch('src.app.os')
@patch('src.app.upload_to_s3')
@patch('src.app.create_parquet_file')
def test_handler_for_release_status(
        mock_create_parquet_file_for_release_status, mock_upload_to_s3,
        mock_os, kinesis_record_for_release_status,
        event_for_release_status):
    """Test handler function."""
    app.handler(event_for_release_status, None)

    mock_create_parquet_file_for_release_status.\
        assert_called_with(kinesis_record_for_release_status)
    mock_upload_to_s3.assert_called_with(
        mock_create_parquet_file_for_release_status.return_value,
        'art_relations',
        'release_status')
    mock_os.remove.assert_called_once_with(
        config.TEMP_FILE_PATH.format
        (mock_create_parquet_file_for_release_status.return_value))
    assert mock_create_parquet_file_for_release_status.return_value


@patch('src.app.s3_client')
def test_upload_to_s3(mock_boto_client):
    """Test to upload to S3."""
    app.upload_to_s3('test.parquet', 'db', 'table')
    mock_boto_client.upload_file.assert_called_once_with(
        config.TEMP_FILE_PATH.format('test.parquet'),
        config.AWS_S3_BUCKET_NAME,
        'db/table/test.parquet')


@patch('src.app.generate_file_name')
@patch('src.app.pq')
@patch('src.app.pa.Table')
@patch('src.app.pd')
def test_create_parquet_file(
        mock_pandas, mock_pyarrow_table, mock_pyarrow_parquet,
        mock_generate_filename, event):
    """Test to create parquet file."""
    table_records = base64.b64decode(
        event['Records'][0]['kinesis']['data']).decode('utf-8')
    app.create_parquet_file(table_records)
    mock_pandas.DataFrame.assert_called_once_with(
        table_records, index=list(range(1, len(table_records) + 1)))
    mock_pyarrow_table.from_pandas.assert_called_once_with(
        mock_pandas.DataFrame.return_value)
    mock_generate_filename.assert_called_with()
    file_name = mock_generate_filename.return_value
    mock_pyarrow_parquet.ParquetWriter.assert_called_once_with(
        config.TEMP_FILE_PATH.format(file_name),
        mock_pyarrow_table.from_pandas.return_value.schema,
        use_deprecated_int96_timestamps=True)
    assert file_name


@patch('src.app.generate_file_name')
@patch('src.app.pq')
@patch('src.app.pa.Table')
@patch('src.app.pd')
def test_create_parquet_file_for_release_status(
        mock_pandas, mock_pyarrow_table, mock_pyarrow_parquet,
        mock_generate_filename, event_for_release_status):
    """Test to create parquet file."""
    table_records = base64.b64decode(
        event_for_release_status['Records'][0]['kinesis']
        ['data']).decode('utf-8')
    app.create_parquet_file(table_records)
    mock_pandas.DataFrame.assert_called_once_with(
        table_records, index=list(range(1, len(table_records) + 1)))
    mock_pyarrow_table.from_pandas.assert_called_once_with(
        mock_pandas.DataFrame.return_value)
    mock_generate_filename.assert_called_with()
    file_name = mock_generate_filename.return_value
    mock_pyarrow_parquet.ParquetWriter.assert_called_once_with(
        config.TEMP_FILE_PATH.format(file_name),
        mock_pyarrow_table.from_pandas.return_value.schema,
        use_deprecated_int96_timestamps=True)
    assert file_name


@pytest.mark.parametrize(
    ('description', 'sanitize_input', 'sanitized_output', 'table_name'),
    [('All valid values of datetime and decimal',
      {'entry_date': '2010-05-25 11:45:54', 'cut_date': '2010-05-20',
          'cash_date': None, 'check_amt': 145},
      {'entry_date': pd.Timestamp('2010-05-25 11:45:54'),
          'cut_date': datetime.date(2010, 5, 20),
          'cash_date': None,
          'check_amt': decimal.Decimal(145)}, 'publishers_checkspaid'),
     ('All fields have 0 values',
      {'entry_date': '0000-00-00 00:00:00', 'cut_date': '0000-00-00',
          'cash_date': None, 'check_amt': 0.00},
      {'entry_date': None, 'cut_date': None, 'cash_date': None,
          'check_amt': None}, 'publishers_checkspaid'),
     ('Decimal values auto-padded upto 6 digits for Athena',
      {'entry_date': '2010-05-25 11:45:54', 'cut_date': '2010-05-20',
          'cash_date': None, 'check_amt': 145.12},
      {'entry_date': pd.Timestamp('2010-05-25 11:45:54'),
          'cut_date': datetime.date(2010, 5, 20),
          'cash_date': None,
          'check_amt': decimal.Decimal('145.120000')},
      'publishers_checkspaid'),
     ('Default 0 for NULL integer column values',
      {'key_id': None, 'changed_by': None,
       'date_added': '0000-00-00 00:00:00'},
      {'key_id': 0, 'changed_by': 0, 'date_added': None}, 'rejection_notes'),
     ])
def test_sanitize_column_values_for_athena(
        description, sanitize_input, sanitized_output, table_name):
    """Test to santize column values for Athena."""
    table_records = app.sanitize_column_values_for_athena(
        sanitize_input, table_name)
    assert table_records == sanitized_output


@pytest.mark.parametrize(
    ('description', 'sanitize_input', 'sanitized_output'),
    [('Invalid value of datetime',
      {'entry_date': '2010-35-65 45:58:88', 'cut_date': '2010-05-20',
          'cash_date': None, 'check_amt': 145.12},
      {'entry_date': None,
          'cut_date': datetime.date(2010, 5, 20),
          'cash_date': None,
          'check_amt': decimal.Decimal('145.120000')}),
     ('Invalid value of date',
      {'entry_date': '2010-05-25 11:45:54', 'cut_date': '20109-34-56',
          'cash_date': None, 'check_amt': 145.12},
      {'entry_date': pd.Timestamp('2010-05-25 11:45:54'),
          'cut_date': None,
          'cash_date': None,
          'check_amt': decimal.Decimal('145.120000')})
     ])
def test_sanitize_invalid_column_values_for_athena(
        description, sanitize_input, sanitized_output):
    """Test to santize column values for Athena."""
    table_records = app.sanitize_column_values_for_athena(
        sanitize_input, 'publishers_checkspaid')
    assert table_records == sanitized_output


@pytest.mark.parametrize(
    ('table_name', 'raw_record', 'expected_output'),
    [('release_approval_queue',
      {'release_approval_id': 1947184, 'release_id': 2731994,
       'release_correction_id': None, 'checked_out_by': 1271,
       'admin_approval': 'N', 'status': 'checked_out',
       'approved_by': None, 'last_updated': '2019-11-15 02:35:23',
       'date_submitted': '2019-11-04 00:00:00',
       'initiate_errorcorrection': 0, 'submitted_by': 32932,
       'changed_by_type': 'oa', 'changed_by': 562},
      True),
     ('manual_adjustment',
      {'id': 494008, 'parent_type': 'vendor', 'parent_id': 7934,
       'date_added': '2007-05-08 10:44:13', 'created_by': 156,
       'amount': -548.43, 'adjust_for_year': 2006,
       'adjust_for_quarter': 4, 'apply_to_year': 2007,
       'apply_to_quarter': 1,
       'comment': 'To mark 4Q06 as paid via wire transfer.',
       'category': None, 'category_id': 21, 'adjust_for_period_id': 96,
       'apply_to_period_id': 99, 'currencies_id': None,
       'amount_in_original_currency': -548.43, 'attachment_location': None,
       'updated_timestamp': '2015-09-28 05:48:30', 'last_modified_by': None,
       'user_type': None},
      True),
     ('release_approval_queue',
      {'release_approval_id': 1947184, 'release_id': 2731994,
       'release_correction_id': None, 'checked_out_by': 1271,
       'admin_approval': 'N', 'status': 'checked_in',
       'approved_by': None, 'last_updated': '2019-11-15 02:35:23',
       'date_submitted': '2019-11-04 00:00:00',
       'initiate_errorcorrection': 0, 'submitted_by': 32932,
       'changed_by_type': 'alw', 'changed_by': 43807},
      False),
     ('release_correction',
      {'release_correction_id': 223126, 'release_id': 2176241,
       'status': 'submitted', 'last_updated': '2019-11-14 17:06:40',
       'last_updated_by': 7123, 'last_updated_type': 'oa'},
      True)
     ])
def test_table_should_log(table_name, raw_record, expected_output):
    """Test table_should_log."""
    table_should_log_response = app.table_should_log(table_name, raw_record)
    assert table_should_log_response == expected_output


@pytest.mark.parametrize(
    ('record', 'expected_output'),
    [({
        'database': 'art_relations',
        'table': 'project',
        'type': 'update',
        'data': {
            'project_id': 1899441,
            'project_code': '80_KOZLOV_VSPOMNI',
            'project_name': 'test',
            'created_date_utc': '2016-11-04 16:43:52.000000',
            'updated_date_utc': '2016-11-04 16:43:52.000000',
            'artist_id': 594576,
            'last_modified_by': 0,
            'user_type': 'oa'
        },
        'old': {
            'artist_id': 12345,
            'last_modified_by': '123',
            'user_type': 'alw',
            'updated_timestamp': '2020-02-03 06:54:34'
        }},
      True),
     ({
         'database': 'art_relations',
         'table': 'project',
         'type': 'update',
         'data': {
             'project_id': 1899441,
             'project_code': '80_KOZLOV_VSPOMNI',
             'project_name': 'test',
             'created_date_utc': '2016-11-04 16:43:52.000000',
             'updated_date_utc': '2016-11-04 16:43:52.000000',
             'artist_id': 594576,
             'last_modified_by': 0,
             'user_type': 'oa'
         },
         'old': {
             'project_name': 'xyz',
             'last_modified_by': '123',
             'user_type': 'alw',
             'updated_timestamp': '2020-02-03 06:54:34'
         }},
      None)
     ])
def test_project_table_log(record, expected_output):
    """Test project_table_log."""
    project_table_log_response = app.project_table_log(record)
    assert project_table_log_response == expected_output