import io
import os
import json
import time
import unittest
import requests
from localstack import config
from localstack.utils.aws import aws_stack
from localstack.utils.common import short_uid, get_service_protocol, to_str, get_free_tcp_port
from localstack.utils.bootstrap import is_api_enabled
from localstack.services.generic_proxy import ProxyListener, run_proxy_server_http2


class TestEdgeAPI(unittest.TestCase):

    def test_invoke_apis_via_edge(self):
        edge_port = config.EDGE_PORT_HTTP or config.EDGE_PORT
        edge_url = '%s://localhost:%s' % (get_service_protocol(), edge_port)

        if is_api_enabled('s3'):
            self._invoke_s3_via_edge(edge_url)
            self._invoke_s3_via_edge_multipart_form(edge_url)
        if is_api_enabled('kinesis'):
            self._invoke_kinesis_via_edge(edge_url)
        if is_api_enabled('dynamodbstreams'):
            self._invoke_dynamodbstreams_via_edge(edge_url)
        if is_api_enabled('firehose'):
            self._invoke_firehose_via_edge(edge_url)
        if is_api_enabled('stepfunctions'):
            self._invoke_stepfunctions_via_edge(edge_url)

    def _invoke_kinesis_via_edge(self, edge_url):
        client = aws_stack.connect_to_service('kinesis', endpoint_url=edge_url)
        result = client.list_streams()
        self.assertIn('StreamNames', result)

    def _invoke_dynamodbstreams_via_edge(self, edge_url):
        client = aws_stack.connect_to_service('dynamodbstreams', endpoint_url=edge_url)
        result = client.list_streams()
        self.assertIn('Streams', result)

    def _invoke_firehose_via_edge(self, edge_url):
        client = aws_stack.connect_to_service('firehose', endpoint_url=edge_url)
        result = client.list_delivery_streams()
        self.assertIn('DeliveryStreamNames', result)

    def _invoke_stepfunctions_via_edge(self, edge_url):
        client = aws_stack.connect_to_service('stepfunctions', endpoint_url=edge_url)
        result = client.list_state_machines()
        self.assertIn('stateMachines', result)

    def _invoke_s3_via_edge(self, edge_url):
        client = aws_stack.connect_to_service('s3', endpoint_url=edge_url)
        bucket_name = 'edge-%s' % short_uid()

        client.create_bucket(Bucket=bucket_name)
        result = client.head_bucket(Bucket=bucket_name)
        self.assertEqual(result['ResponseMetadata']['HTTPStatusCode'], 200)
        client.delete_bucket(Bucket=bucket_name)

        bucket_name = 'edge-%s' % short_uid()
        object_name = 'testobject'
        bucket_url = '%s/%s' % (edge_url, bucket_name)
        result = requests.put(bucket_url, verify=False)
        self.assertEqual(result.status_code, 200)
        result = client.head_bucket(Bucket=bucket_name)
        self.assertEqual(result['ResponseMetadata']['HTTPStatusCode'], 200)
        headers = {'Content-Type': 'application/x-www-form-urlencoded'}
        result = requests.post(bucket_url, data='key=%s&file=file_content_123' % object_name,
            headers=headers, verify=False)
        self.assertEqual(result.status_code, 204)

        bucket_url = '%s/example' % (bucket_url)
        result = requests.put(bucket_url, data='hello', verify=False)
        self.assertEqual(result.status_code, 200)

        result = io.BytesIO()
        client.download_fileobj(bucket_name, object_name, result)
        self.assertEqual('file_content_123', to_str(result.getvalue()))

    def _invoke_s3_via_edge_multipart_form(self, edge_url):
        client = aws_stack.connect_to_service('s3', endpoint_url=edge_url)
        bucket_name = 'edge-%s' % short_uid()
        object_name = 'testobject'
        object_data = b'testdata'

        client.create_bucket(Bucket=bucket_name)
        presigned_post = client.generate_presigned_post(bucket_name, object_name)

        files = {'file': object_data}
        r = requests.post(presigned_post['url'], data=presigned_post['fields'], files=files, verify=False)
        self.assertEqual(r.status_code, 204)

        result = io.BytesIO()
        client.download_fileobj(bucket_name, object_name, result)
        self.assertEqual(to_str(object_data), to_str(result.getvalue()))

        client.delete_object(Bucket=bucket_name, Key=object_name)
        client.delete_bucket(Bucket=bucket_name)

    def test_http2_traffic(self):
        port = get_free_tcp_port()

        class MyListener(ProxyListener):
            def forward_request(self, method, path, data, headers):
                return {
                    'method': method, 'path': path, 'data': data
                }

        url = 'https://localhost:%s/foo/bar' % port

        listener = MyListener()
        run_proxy_server_http2(port, listener=listener, asynchronous=True, use_ssl=True)
        time.sleep(1)
        response = requests.post(url, verify=False)
        self.assertEqual({'method': 'POST', 'path': '/foo/bar', 'data': ''}, json.loads(to_str(response.content)))

    def test_invoke_sns_sqs_integration_using_edge_port(self):
        edge_port = config.EDGE_PORT_HTTP or config.EDGE_PORT
        region_original = os.environ.get('DEFAULT_REGION')
        os.environ['DEFAULT_REGION'] = 'us-southeast-2'
        edge_url = '%s://localhost:%s' % (get_service_protocol(), edge_port)
        sns_client = aws_stack.connect_to_service('sns', endpoint_url=edge_url)
        sqs_client = aws_stack.connect_to_service('sqs', endpoint_url=edge_url)

        topic = sns_client.create_topic(Name='test_topic3')
        topic_arn = topic['TopicArn']
        test_queue = sqs_client.create_queue(QueueName='test_queue3')

        queue_url = test_queue['QueueUrl']
        sqs_client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=['QueueArn'])
        sns_client.subscribe(TopicArn=topic_arn, Protocol='sqs', Endpoint=queue_url)
        sns_client.publish(TargetArn=topic_arn, Message='Test msg')

        response = sqs_client.receive_message(
            QueueUrl=queue_url,
            AttributeNames=['SentTimestamp'],
            MaxNumberOfMessages=1,
            MessageAttributeNames=['All'],
            VisibilityTimeout=2,
            WaitTimeSeconds=2,
        )
        self.assertEqual(len(response['Messages']), 1)

        os.environ.pop('DEFAULT_REGION')
        if region_original is not None:
            os.environ['DEFAULT_REGION'] = region_original
