import datetime
from pathlib import Path
from unittest import mock
from unittest.mock import patch, call, MagicMock, PropertyMock

import pytest
from airflow.exceptions import AirflowFailException

from flows.awal_salesforce import tasks
from tests.testing_utils import SubstringMatcher


@pytest.mark.freeze_time('2022-12-21 12:33:21')
def test_bootstrap():
    dag_run_mock = MagicMock()
    date_obj = datetime.datetime(2022, 12, 21, 12, 32, 59, tzinfo=datetime.timezone.utc)
    type(dag_run_mock).logical_date = PropertyMock(return_value=date_obj)

    result = tasks.bootstrap(
        dag_run=dag_run_mock,
        ds=date_obj.strftime('%Y-%m-%d')
    )

    assert result == {
        'archive_s3_bucket': 'dev-cucumbers',
        'archive_s3_key': 'AWAL/salesforce/2022-12-21/',
        'archive_s3_path': 's3://dev-cucumbers/AWAL/salesforce/2022-12-21/'
    }


@pytest.mark.freeze_time('2022-12-21 12:33:21')
def test_bootstrap_negative():
    dag_run_mock = MagicMock()
    date_obj = datetime.datetime(2022, 12, 20, 18, 56, 59, tzinfo=datetime.timezone.utc)
    type(dag_run_mock).logical_date = PropertyMock(return_value=date_obj)

    with pytest.raises(AirflowFailException):
        tasks.bootstrap(
            dag_run=dag_run_mock,
            ds=date_obj.strftime('%Y-%m-%d')
        )


@patch.object(tasks, 'S3Hook')
@patch.object(tasks, 'api')
def test_fetch(api_mock, s3hook_mock):
    api_mock.describe_salesforce_object.return_value = {'desc_sf_object': True}

    result = tasks.fetch(
        salesforce_entity='Account',
        s3_bucket='abucket',
        s3_key='s3/key/',
    )

    assert result == {'data_file': 'account_data.csv', 'structure_file': 'account_describe.json'}
    assert s3hook_mock.return_value.load_file.call_args_list == [
        call(filename=mock.ANY, replace=True, bucket_name='abucket', key='s3/key/account_describe.json'),
        call(filename=mock.ANY, replace=True, bucket_name='abucket', key='s3/key/account_data.csv'),
    ]


@patch.object(tasks, 'SnowflakeHook')
@patch.object(tasks, 'S3Hook')
@patch.object(tasks, 'api')
def test_create_table(api_mock, s3hook_mock, snowflakehook_mock, tmp_path):
    filename = 'account_desc.json'

    def download_file(local_path, **kwargs):
        filename = Path(local_path) / 'tmp_file'
        filename.write_text('{"key": "value"}')
        return str(filename)

    s3hook_mock.return_value.download_file.side_effect = download_file

    result = tasks.create_table(
        s3_bucket='abucket',
        s3_key='s3/key/',
        structure_file=filename,
        table_name='temp_table',
    )

    assert snowflakehook_mock.return_value.run.call_args_list == [
        call(sql=api_mock.gen_create_table_dds(), parameters={})
    ]
    assert result == snowflakehook_mock().run()


@patch.object(tasks, 'SnowflakeHook')
@patch.object(tasks, 'AwsBaseHook')
def test_copy_data(awsbasehook_mock, snowflakehook_mock):
    result = tasks.copy_data(
        data_file='file1.csv',
        table_name='temp_table',
        s3_path='s3://abucket/some/key/'
    )

    assert snowflakehook_mock.return_value.run.call_args_list == [
        call(
            sql=SubstringMatcher(
                containing=[
                    'COPY INTO temp_table'
                ]
            ),
            parameters={
                'table_name': 'temp_table',
                'aws_key_id': awsbasehook_mock().get_session().get_credentials().access_key,
                'aws_secret_key': awsbasehook_mock().get_session().get_credentials().secret_key,
                's3_path': 's3://abucket/some/key/',
                'files': 'file1.csv'}
        )
    ]
    assert result == snowflakehook_mock().run()