import logging

import pytest
from streamsets.testframework.markers import aws, sdc_min_version
from streamsets.testframework.utils import get_random_string

logger = logging.getLogger(__name__)

S3_SANDBOX_PREFIX = 'sandbox'
LOG_FIELD_MAPPING = [{'fieldPath': '/date', 'group': 1},
                     {'fieldPath': '/time', 'group': 2},
                     {'fieldPath': '/timehalf', 'group': 3},
                     {'fieldPath': '/info', 'group': 4},
                     {'fieldPath': '/file', 'group': 5},
                     {'fieldPath': '/message', 'group': 6}]
REGULAR_EXPRESSION = r'(\S+) (\S+) (\S+) (\S+) (\S+) (.*)'
# log to be written int the file on s3
data_format_content = {
        'COMMON_LOG_FORMAT': '127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] '
                             '"GET /apache.gif HTTP/1.0" 200 232',
        'LOG4J': '200 [main] DEBUG org.StreamSets.Log4j unknown - This is sample log message',
        'APACHE_ERROR_LOG_FORMAT': '[Wed Oct 11 14:32:52 2000] [error] [client 127.0.0.1] client '
                                   'denied by server configuration:/export/home/live/ap/htdocs/test',
        'COMBINED_LOG_FORMAT': '127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache.gif'
                               ' HTTP/1.0" 200 2326 "http://www.example.com/strt.html" "Mozilla/4.08'
                               ' [en] (Win98; I ;Nav)"',
        'APACHE_CUSTOM_LOG_FORMAT': '10.185.248.71 - - [09/Jan/2015:9:12:06 +0000] "GET '
                                    '/inventoryServic/inventory/purchaseItem?userId=20253471&itemId=23434300 '
                                    'HTTP/1.1" 500 17 ',
        'CEF': '10.217.31.247 CEF:0|Citrix|NetScaler|NS10.0|APPFW|APPFW_STARTURL|6|src=10.217.253.78 '
               'spt=53743 method=GET request=http://vpx247.example.net/FFC/login.html msg=Disallow Illegal URL.',
        'LEEF': 'LEEF: 2.0|Trend Micro|Deep Security Agent|<DSA version>|4000030|cat=Anti-Malware '
                'name=HEU_AEGIS_CRYPT desc=HEU_AEGIS_CRYPT sev=6 cn1=241 msg=Realtime',
        'REGEX': '2019-04-30 08:23:53 AM [INFO] [streamsets.sdk.sdc_api] Pipeline Filewriterpipeline53'}
# data to verify the output of amazon s3 origin.
get_data_to_verify_output = {
        'LOG4J': {'severity': 'DEBUG', 'relativetime': '200', 'thread': 'main', 'category': 'org.StreamSets.Log4j',
                  'ndc': 'unknown', 'message': 'This is sample log message'},
        'COMMON_LOG_FORMAT': {'request': '/apache.gif', 'auth': 'frank', 'ident': '-', 'response': '200', 'bytes':
                              '232', 'clientip': '127.0.0.1', 'verb': 'GET', 'httpversion': '1.0', 'rawrequest': None,
                              'timestamp': '10/Oct/2000:13:55:36 -0700'},
        'APACHE_ERROR_LOG_FORMAT': {'message': 'client denied by server configuration:/export/home/live/ap/htdocs/'
                                               'test', 'timestamp': 'Wed Oct 11 14:32:52 2000', 'loglevel': 'error',
                                    'clientip': '127.0.0.1'},
        'COMBINED_LOG_FORMAT': {'request': '/apache.gif', 'agent': '"Mozilla/4.08 [en] (Win98; I ;Nav)"', 'auth':
                                'frank', 'ident': '-', 'verb': 'GET', 'referrer': '"http://www.example.com/strt.'
                                'html"', 'response': '200', 'bytes': '2326', 'clientip': '127.0.0.1',
                                'httpversion': '1.0', 'rawrequest': None, 'timestamp': '10/Oct/2000:13:55:36 -0700'},
        'APACHE_CUSTOM_LOG_FORMAT': {'remoteUser': '-', 'requestTime': '09/Jan/2015:9:12:06 +0000', 'request': 'GET '
                                     '/inventoryServic/inventory/purchaseItem?userId=20253471&itemId=23434300 HTTP/1.1',
                                     'logName': '-', 'remoteHost': '10.185.248.71', 'bytesSent': '17', 'status': '500'},
        'CEF': {'severity': '6', 'product': 'NetScaler', 'extensions': {'msg': 'Disallow Illegal URL.', 'request':
                'http://vpx247.example.net/FFC/login.html', 'method': 'GET', 'src': '10.217.253.78', 'spt': '53743'},
                'signature': 'APPFW', 'vendor': 'Citrix', 'cefVersion': 0, 'name': 'APPFW_STARTURL',
                'version': 'NS10.0'},
        'GROK': {'request': '/inventoryServic/inventory/purchaseItem?userId=20253471&itemId=23434300', 'auth': '-',
                 'ident': '-', 'response': '500', 'bytes': '17', 'clientip': '10.185.248.71', 'verb': 'GET',
                 'httpversion': '1.1', 'rawrequest': None, 'timestamp': '09/Jan/2015:9:12:06 +0000'},
        'LEEF': {'eventId': '4000030', 'product': 'Deep Security Agent', 'extensions': {'cat': 'Realtime'},
                 'leefVersion': 2.0, 'vendor': 'Trend Micro', 'version': '<DSA version>'},
        'REGEX': {'/time': '08:23:53', '/date': '2019-04-30', '/timehalf': 'AM',
                  '/info': '[INFO]', '/message': 'Pipeline Filewriterpipeline53', '/file': '[streamsets.sdk.sdc_api]'}}


@pytest.mark.skip('Not yet implemented')
def test_configuration_access_key_id(sdc_builder, sdc_executor):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_bucket(sdc_builder, sdc_executor):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_connection_timeout(sdc_builder, sdc_executor):
    pass


@pytest.mark.parametrize('task', ['CREATE_NEW_OBJECT'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_content(sdc_builder, sdc_executor, task):
    pass


@pytest.mark.parametrize('task', ['COPY_OBJECT'])
@pytest.mark.parametrize('delete_original_object', [False, True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_delete_original_object(sdc_builder, sdc_executor, task, delete_original_object):
    pass


@pytest.mark.parametrize('region', ['OTHER'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_endpoint(sdc_builder, sdc_executor, region):
    pass


@pytest.mark.parametrize('task', ['COPY_OBJECT'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_new_object_path(sdc_builder, sdc_executor, task):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_object(sdc_builder, sdc_executor):
    pass


@pytest.mark.parametrize('on_record_error', ['DISCARD', 'STOP_PIPELINE', 'TO_ERROR'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_on_record_error(sdc_builder, sdc_executor, on_record_error):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_preconditions(sdc_builder, sdc_executor):
    pass


@pytest.mark.parametrize('use_proxy', [True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_proxy_host(sdc_builder, sdc_executor, use_proxy):
    pass


@pytest.mark.parametrize('use_proxy', [True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_proxy_password(sdc_builder, sdc_executor, use_proxy):
    pass


@pytest.mark.parametrize('use_proxy', [True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_proxy_port(sdc_builder, sdc_executor, use_proxy):
    pass


@pytest.mark.parametrize('use_proxy', [True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_proxy_user(sdc_builder, sdc_executor, use_proxy):
    pass


@pytest.mark.parametrize('region', ['AP_NORTHEAST_1', 'AP_NORTHEAST_2', 'AP_NORTHEAST_3', 'AP_SOUTHEAST_1', 'AP_SOUTHEAST_2', 'AP_SOUTH_1', 'CA_CENTRAL_1', 'CN_NORTHWEST_1', 'CN_NORTH_1', 'EU_CENTRAL_1', 'EU_WEST_1', 'EU_WEST_2', 'EU_WEST_3', 'OTHER', 'SA_EAST_1', 'US_EAST_1', 'US_EAST_2', 'US_GOV_WEST_1', 'US_WEST_1', 'US_WEST_2'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_region(sdc_builder, sdc_executor, region):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_required_fields(sdc_builder, sdc_executor):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_retry_count(sdc_builder, sdc_executor):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_secret_access_key(sdc_builder, sdc_executor):
    pass


@pytest.mark.skip('Not yet implemented')
def test_configuration_socket_timeout(sdc_builder, sdc_executor):
    pass


@pytest.mark.parametrize('task', ['CHANGE_EXISTING_OBJECT'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_tags(sdc_builder, sdc_executor, task):
    pass


@pytest.mark.parametrize('task', ['CHANGE_EXISTING_OBJECT', 'COPY_OBJECT', 'CREATE_NEW_OBJECT'])
@pytest.mark.skip('Not yet implemented')
def test_configuration_task(sdc_builder, sdc_executor, task):
    pass


@pytest.mark.parametrize('use_proxy', [False, True])
@pytest.mark.skip('Not yet implemented')
def test_configuration_use_proxy(sdc_builder, sdc_executor, use_proxy):
    pass


@aws('s3')
@pytest.mark.parametrize('data_format', ['LOG'])
@pytest.mark.parametrize('log_format', ['COMMON_LOG_FORMAT', 'APACHE_ERROR_LOG_FORMAT', 'COMBINED_LOG_FORMAT',
                                        'APACHE_CUSTOM_LOG_FORMAT', 'REGEX', 'GROK', 'LOG4J', 'CEF', 'LEEF'])
def test_configurations_data_format_log(sdc_executor, sdc_builder, aws, data_format, log_format):
    """Check whether S3 origin can parse different log format or not. A log file is being created in s3 bucket
    mentioned below .S3 origin reads the log file and parse the same.

    Pipeline for the same-
    s3_origin >> trash
    s3_origin >= pipeline_finisher_executor
    """
    if log_format == 'GROK':
        file_content = data_format_content['APACHE_CUSTOM_LOG_FORMAT']
    else:
        file_content = data_format_content[log_format]
    client = aws.s3
    s3_key = f'{S3_SANDBOX_PREFIX}/{get_random_string()}'
    attributes = {'bucket': aws.s3_bucket_name,
                  'prefix_pattern': f'{s3_key}/*',
                  'number_of_threads': 1,
                  'read_order': 'LEXICOGRAPHICAL',
                  'data_format': data_format,
                  'log_format': log_format,
                  'custom_log_format': '%h %l %u [%t] "%r" %>s %b',
                  'regular_expression': REGULAR_EXPRESSION,
                  'field_path_to_regex_group_mapping': LOG_FIELD_MAPPING
                  }
    pipeline = get_aws_origin_to_trash_pipeline(sdc_builder, attributes, aws)
    s3_origin = pipeline.origin_stage
    try:
        client.put_object(Bucket=aws.s3_bucket_name, Key=f'{s3_key}/{get_random_string()}.log', Body=file_content)
        output_records = execute_pipeline_and_get_output(sdc_executor, s3_origin, pipeline)
        assert output_records[0].field == get_data_to_verify_output[log_format]
    finally:
        if sdc_executor.get_pipeline_status(pipeline).response.json().get('status') == 'RUNNING':
            sdc_executor.stop_pipeline(pipeline)
        # cleaning up s3 bucket
        delete_aws_objects(client, aws, s3_key)


def get_aws_origin_to_trash_pipeline(sdc_builder, attributes, aws):
    # Build pipeline.
    builder = sdc_builder.get_pipeline_builder()
    builder.add_error_stage('Discard')
    s3_origin = builder.add_stage('Amazon S3', type='origin')
    s3_origin.set_attributes(**attributes)
    trash = builder.add_stage('Trash')
    pipeline_finisher_executor = builder.add_stage('Pipeline Finisher Executor')
    pipeline_finisher_executor.set_attributes(stage_record_preconditions=["${record:eventType() == 'no-more-data'}"])
    s3_origin >> trash
    s3_origin >= pipeline_finisher_executor
    s3_origin_pipeline = builder.build().configure_for_environment(aws)
    s3_origin_pipeline.configuration['shouldRetry'] = False
    return s3_origin_pipeline


def delete_aws_objects(client, aws, s3_key):
    # Clean up S3.
    delete_keys = {'Objects': [{'Key': k['Key']}
                               for k in
                               client.list_objects_v2(Bucket=aws.s3_bucket_name, Prefix=s3_key)['Contents']]}
    client.delete_objects(Bucket=aws.s3_bucket_name, Delete=delete_keys)


def execute_pipeline_and_get_output(sdc_executor, s3_origin, pipeline):
    sdc_executor.add_pipeline(pipeline)
    snapshot = sdc_executor.capture_snapshot(pipeline, start_pipeline=True).snapshot
    output_records = snapshot[s3_origin].output
    return output_records

[(7, 'logging.getLogger', 'logging.getLogger', False, 'import logging\n'), (65, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (70, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (75, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (80, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (81, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (86, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (87, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (88, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (93, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (94, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (99, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (100, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (105, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (110, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (111, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (116, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (121, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (122, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (127, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (128, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (133, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (134, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (139, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (140, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (145, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (146, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (151, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (156, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (161, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (166, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (171, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (172, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (177, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (178, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (183, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (184, 'pytest.mark.skip', 'pytest.mark.skip', False, 'import pytest\n'), (189, 'streamsets.testframework.markers.aws', 'aws', False, 'from streamsets.testframework.markers import aws, sdc_min_version\n'), (190, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (191, 'pytest.mark.parametrize', 'pytest.mark.parametrize', False, 'import pytest\n'), (206, 'streamsets.testframework.utils.get_random_string', 'get_random_string', False, 'from streamsets.testframework.utils import get_random_string\n'), (220, 'streamsets.testframework.utils.get_random_string', 'get_random_string', False, 'from streamsets.testframework.utils import get_random_string\n')]