# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one or more§
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import socket
import sys
import ssl

from requests.exceptions import ConnectTimeout

from mock import Mock, patch

import requests_mock

import libcloud.common.base

from libcloud.test import unittest
from libcloud.common.base import Connection, CertificateConnection
from libcloud.common.exceptions import RateLimitReachedError
from libcloud.http import LibcloudBaseConnection
from libcloud.http import LibcloudConnection
from libcloud.http import SignedHTTPSAdapter
from libcloud.utils.retry import Retry
from libcloud.utils.retry import RetryForeverOnRateLimitError
from libcloud.utils.retry import RETRY_EXCEPTIONS
from libcloud.utils.py3 import assertRaisesRegex


class BaseConnectionClassTestCase(unittest.TestCase):
    def setUp(self):
        self.orig_proxy = os.environ.pop("http_proxy", None)

    def tearDown(self):
        if self.orig_proxy:
            os.environ["http_proxy"] = self.orig_proxy
        elif "http_proxy" in os.environ:
            del os.environ["http_proxy"]

        libcloud.common.base.ALLOW_PATH_DOUBLE_SLASHES = False

    @classmethod
    def tearDownClass(cls):
        if "http_proxy" in os.environ:
            del os.environ["http_proxy"]

        libcloud.common.base.ALLOW_PATH_DOUBLE_SLASHES = False

    def test_parse_proxy_url(self):
        conn = LibcloudBaseConnection()

        proxy_url = "http://127.0.0.1:3128"
        result = conn._parse_proxy_url(proxy_url=proxy_url)
        self.assertEqual(result[0], "http")
        self.assertEqual(result[1], "127.0.0.1")
        self.assertEqual(result[2], 3128)
        self.assertIsNone(result[3])
        self.assertIsNone(result[4])

        proxy_url = "https://127.0.0.2:3129"
        result = conn._parse_proxy_url(proxy_url=proxy_url)
        self.assertEqual(result[0], "https")
        self.assertEqual(result[1], "127.0.0.2")
        self.assertEqual(result[2], 3129)
        self.assertIsNone(result[3])
        self.assertIsNone(result[4])

        proxy_url = "http://user1:pass1@127.0.0.1:3128"
        result = conn._parse_proxy_url(proxy_url=proxy_url)
        self.assertEqual(result[0], "http")
        self.assertEqual(result[1], "127.0.0.1")
        self.assertEqual(result[2], 3128)
        self.assertEqual(result[3], "user1")
        self.assertEqual(result[4], "pass1")

        proxy_url = "https://user1:pass1@127.0.0.2:3129"
        result = conn._parse_proxy_url(proxy_url=proxy_url)
        self.assertEqual(result[0], "https")
        self.assertEqual(result[1], "127.0.0.2")
        self.assertEqual(result[2], 3129)
        self.assertEqual(result[3], "user1")
        self.assertEqual(result[4], "pass1")

        proxy_url = "http://127.0.0.1"
        expected_msg = "proxy_url must be in the following format"
        assertRaisesRegex(
            self, ValueError, expected_msg, conn._parse_proxy_url, proxy_url=proxy_url
        )

        proxy_url = "http://@127.0.0.1:3128"
        expected_msg = "URL is in an invalid format"
        assertRaisesRegex(
            self, ValueError, expected_msg, conn._parse_proxy_url, proxy_url=proxy_url
        )

        proxy_url = "http://user@127.0.0.1:3128"
        expected_msg = "URL is in an invalid format"
        assertRaisesRegex(
            self, ValueError, expected_msg, conn._parse_proxy_url, proxy_url=proxy_url
        )

    def test_constructor(self):
        proxy_url = "http://127.0.0.2:3128"
        os.environ["http_proxy"] = proxy_url
        conn = LibcloudConnection(host="localhost", port=80)
        self.assertEqual(conn.proxy_scheme, "http")
        self.assertEqual(conn.proxy_host, "127.0.0.2")
        self.assertEqual(conn.proxy_port, 3128)
        self.assertEqual(
            conn.session.proxies,
            {"http": "http://127.0.0.2:3128", "https": "http://127.0.0.2:3128"},
        )

        _ = os.environ.pop("http_proxy", None)
        conn = LibcloudConnection(host="localhost", port=80)
        self.assertIsNone(conn.proxy_scheme)
        self.assertIsNone(conn.proxy_host)
        self.assertIsNone(conn.proxy_port)

        proxy_url = "http://127.0.0.3:3128"
        conn.set_http_proxy(proxy_url=proxy_url)
        self.assertEqual(conn.proxy_scheme, "http")
        self.assertEqual(conn.proxy_host, "127.0.0.3")
        self.assertEqual(conn.proxy_port, 3128)
        self.assertEqual(
            conn.session.proxies,
            {"http": "http://127.0.0.3:3128", "https": "http://127.0.0.3:3128"},
        )

        proxy_url = "http://127.0.0.4:3128"
        conn = LibcloudConnection(host="localhost", port=80, proxy_url=proxy_url)
        self.assertEqual(conn.proxy_scheme, "http")
        self.assertEqual(conn.proxy_host, "127.0.0.4")
        self.assertEqual(conn.proxy_port, 3128)
        self.assertEqual(
            conn.session.proxies,
            {"http": "http://127.0.0.4:3128", "https": "http://127.0.0.4:3128"},
        )

        os.environ["http_proxy"] = proxy_url
        proxy_url = "http://127.0.0.5:3128"
        conn = LibcloudConnection(host="localhost", port=80, proxy_url=proxy_url)
        self.assertEqual(conn.proxy_scheme, "http")
        self.assertEqual(conn.proxy_host, "127.0.0.5")
        self.assertEqual(conn.proxy_port, 3128)
        self.assertEqual(
            conn.session.proxies,
            {"http": "http://127.0.0.5:3128", "https": "http://127.0.0.5:3128"},
        )

        os.environ["http_proxy"] = proxy_url
        proxy_url = "https://127.0.0.6:3129"
        conn = LibcloudConnection(host="localhost", port=80, proxy_url=proxy_url)
        self.assertEqual(conn.proxy_scheme, "https")
        self.assertEqual(conn.proxy_host, "127.0.0.6")
        self.assertEqual(conn.proxy_port, 3129)
        self.assertEqual(
            conn.session.proxies,
            {"http": "https://127.0.0.6:3129", "https": "https://127.0.0.6:3129"},
        )

    def test_connection_to_unusual_port(self):
        conn = LibcloudConnection(host="localhost", port=8080)
        self.assertIsNone(conn.proxy_scheme)
        self.assertIsNone(conn.proxy_host)
        self.assertIsNone(conn.proxy_port)
        self.assertEqual(conn.host, "http://localhost:8080")

        conn = LibcloudConnection(host="localhost", port=80)
        self.assertEqual(conn.host, "http://localhost")

    def test_connection_session_timeout(self):
        """
        Test that the connection timeout attribute is set correctly
        """
        conn = LibcloudConnection(host="localhost", port=8080)
        self.assertEqual(conn.session.timeout, 60)

        conn = LibcloudConnection(host="localhost", port=8080, timeout=10)
        self.assertEqual(conn.session.timeout, 10)

    def test_connection_timeout_raised(self):
        """
        Test that the connection times out
        """
        conn = LibcloudConnection(host="localhost", port=8080, timeout=0.1)
        # use a not-routable address to test that the connection timeouts
        host = "http://10.255.255.1"
        with self.assertRaises(ConnectTimeout):
            conn.request("GET", host)

    def test_connection_url_merging(self):
        """
        Test that the connection class will parse URLs correctly
        """
        conn = Connection(url="http://test.com/")
        conn.connect()
        self.assertEqual(conn.connection.host, "http://test.com")
        with requests_mock.mock() as m:
            m.get("http://test.com/test", text="data")
            response = conn.request("/test")
        self.assertEqual(response.body, "data")

    def test_morph_action_hook(self):
        conn = Connection(url="http://test.com")

        conn.request_path = ""
        self.assertEqual(conn.morph_action_hook("/test"), "/test")
        self.assertEqual(conn.morph_action_hook("test"), "/test")

        conn.request_path = "/v1"
        self.assertEqual(conn.morph_action_hook("/test"), "/v1/test")
        self.assertEqual(conn.morph_action_hook("test"), "/v1/test")

        conn.request_path = "/v1"
        self.assertEqual(conn.morph_action_hook("/test"), "/v1/test")
        self.assertEqual(conn.morph_action_hook("test"), "/v1/test")

        conn.request_path = "v1"
        self.assertEqual(conn.morph_action_hook("/test"), "/v1/test")
        self.assertEqual(conn.morph_action_hook("test"), "/v1/test")

        conn.request_path = "v1/"
        self.assertEqual(conn.morph_action_hook("/test"), "/v1/test")
        self.assertEqual(conn.morph_action_hook("test"), "/v1/test")

        conn.request_path = "/a"
        self.assertEqual(conn.morph_action_hook("//b/c.txt"), "/a/b/c.txt")

        conn.request_path = "/b"
        self.assertEqual(conn.morph_action_hook("/foo//"), "/b/foo/")

        libcloud.common.base.ALLOW_PATH_DOUBLE_SLASHES = True

        conn.request_path = "/"
        self.assertEqual(conn.morph_action_hook("/"), "//")

        conn.request_path = ""
        self.assertEqual(conn.morph_action_hook("/"), "/")

        conn.request_path = "/a"
        self.assertEqual(conn.morph_action_hook("//b/c.txt"), "/a//b/c.txt")

        conn.request_path = "/b"
        self.assertEqual(conn.morph_action_hook("/foo//"), "/b/foo//")

    def test_connect_with_prefix(self):
        """
        Test that a connection with a base path (e.g. /v1/) will
        add the base path to requests
        """
        conn = Connection(url="http://test.com/")
        conn.connect()
        conn.request_path = "/v1"
        self.assertEqual(conn.connection.host, "http://test.com")
        with requests_mock.mock() as m:
            m.get("http://test.com/v1/test", text="data")
            response = conn.request("/test")
        self.assertEqual(response.body, "data")

    def test_secure_connection_unusual_port(self):
        """
        Test that the connection class will default to secure (https) even
        when the port is an unusual (non 443, 80) number
        """
        conn = Connection(secure=True, host="localhost", port=8081)
        conn.connect()
        self.assertEqual(conn.connection.host, "https://localhost:8081")

        conn2 = Connection(url="https://localhost:8081")
        conn2.connect()
        self.assertEqual(conn2.connection.host, "https://localhost:8081")

    def test_secure_by_default(self):
        """
        Test that the connection class will default to secure (https)
        """
        conn = Connection(host="localhost", port=8081)
        conn.connect()
        self.assertEqual(conn.connection.host, "https://localhost:8081")

    def test_implicit_port(self):
        """
        Test that the port is not included in the URL if the protocol implies
        the port, e.g. http implies 80
        """
        conn = Connection(secure=True, host="localhost", port=443)
        conn.connect()
        self.assertEqual(conn.connection.host, "https://localhost")

        conn2 = Connection(secure=False, host="localhost", port=80)
        conn2.connect()
        self.assertEqual(conn2.connection.host, "http://localhost")

    def test_insecure_connection_unusual_port(self):
        """
        Test that the connection will allow unusual ports and insecure
        schemes
        """
        conn = Connection(secure=False, host="localhost", port=8081)
        conn.connect()
        self.assertEqual(conn.connection.host, "http://localhost:8081")

        conn2 = Connection(url="http://localhost:8081")
        conn2.connect()
        self.assertEqual(conn2.connection.host, "http://localhost:8081")


class ConnectionClassTestCase(unittest.TestCase):
    def setUp(self):
        self.originalConnect = Connection.connect
        self.originalResponseCls = Connection.responseCls

        Connection.connect = Mock()
        Connection.responseCls = Mock()
        Connection.allow_insecure = True

    def tearDown(self):
        Connection.connect = self.originalConnect
        Connection.responseCls = Connection.responseCls
        Connection.allow_insecure = True

    def test_dont_allow_insecure(self):
        Connection.allow_insecure = True
        Connection(secure=False)

        Connection.allow_insecure = False

        expected_msg = r"Non https connections are not allowed \(use " r"secure=True\)"
        assertRaisesRegex(self, ValueError, expected_msg, Connection, secure=False)

    def test_cache_busting(self):
        params1 = {"foo1": "bar1", "foo2": "bar2"}
        params2 = [("foo1", "bar1"), ("foo2", "bar2")]

        con = Connection()
        con.connection = Mock()
        con.pre_connect_hook = Mock()
        con.pre_connect_hook.return_value = {}, {}
        con.cache_busting = False

        con.request(action="/path", params=params1)
        args, kwargs = con.pre_connect_hook.call_args
        self.assertFalse("cache-busting" in args[0])
        self.assertEqual(args[0], params1)

        con.request(action="/path", params=params2)
        args, kwargs = con.pre_connect_hook.call_args
        self.assertFalse("cache-busting" in args[0])
        self.assertEqual(args[0], params2)

        con.cache_busting = True

        con.request(action="/path", params=params1)
        args, kwargs = con.pre_connect_hook.call_args
        self.assertTrue("cache-busting" in args[0])

        con.request(action="/path", params=params2)
        args, kwargs = con.pre_connect_hook.call_args
        self.assertTrue("cache-busting" in args[0][len(params2)])

    def test_context_is_reset_after_request_has_finished(self):
        context = {"foo": "bar"}

        def responseCls(connection, response):
            connection.called = True
            self.assertEqual(connection.context, context)

        con = Connection()
        con.called = False
        con.connection = Mock()
        con.responseCls = responseCls

        con.set_context(context)
        self.assertEqual(con.context, context)

        con.request("/")

        # Context should have been reset
        self.assertTrue(con.called)
        self.assertEqual(con.context, {})

        # Context should also be reset if a method inside request throws
        con = Connection(timeout=1, retry_delay=0.1)
        con.connection = Mock()

        con.set_context(context)
        self.assertEqual(con.context, context)
        con.connection.request = Mock(side_effect=ssl.SSLError())

        try:
            con.request("/")
        except ssl.SSLError:
            pass

        self.assertEqual(con.context, {})

        con.connection = Mock()
        con.set_context(context)
        self.assertEqual(con.context, context)

        con.responseCls = Mock(side_effect=ValueError())

        try:
            con.request("/")
        except ValueError:
            pass

        self.assertEqual(con.context, {})

    def _raise_socket_error(self):
        raise socket.gaierror("")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_with_sleep(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        mock_connect.side_effect = socket.gaierror("")
        retry_request = Retry(timeout=0.2, retry_delay=0.1, backoff=1)
        self.assertRaises(socket.gaierror, retry_request(con.request), action="/")

        self.assertGreater(mock_connect.call_count, 1, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_with_timeout(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        mock_connect.side_effect = socket.gaierror("")
        retry_request = Retry(timeout=0.2, retry_delay=0.1, backoff=1)
        self.assertRaises(socket.gaierror, retry_request(con.request), action="/")

        self.assertGreater(mock_connect.call_count, 1, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_with_backoff(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        mock_connect.side_effect = socket.gaierror("")
        retry_request = Retry(timeout=0.2, retry_delay=0.1, backoff=1)
        self.assertRaises(socket.gaierror, retry_request(con.request), action="/")
        self.assertGreater(mock_connect.call_count, 1, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_rate_limit_error_timeout(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        mock_connect.__name__ = "mock_connect"
        headers = {"retry-after": 0.2}
        mock_connect.side_effect = RateLimitReachedError(headers=headers)
        retry_request = Retry(timeout=0.4, retry_delay=0.1, backoff=1)
        self.assertRaises(RateLimitReachedError, retry_request(con.request), action="/")
        self.assertEqual(mock_connect.call_count, 2, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_rate_limit_error_forever_with_old_retry_class(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        self.retry_counter = 0

        def mock_connect_side_effect(*args, **kwargs):
            self.retry_counter += 1

            if self.retry_counter < 4:
                headers = {"retry-after": 0.1}
                raise RateLimitReachedError(headers=headers)

            return "success"

        mock_connect.__name__ = "mock_connect"
        mock_connect.side_effect = mock_connect_side_effect
        retry_request = RetryForeverOnRateLimitError(
            timeout=0.1, retry_delay=0.1, backoff=1
        )
        retry_request(con.request)(action="/")

        # We have waited longer the timeout but continue to retry
        result = retry_request(con.request)(action="/")
        self.assertEqual(result, "success")

        self.assertEqual(mock_connect.call_count, 5, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_should_not_retry_on_non_defined_exception(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        self.retry_counter = 0

        mock_connect.__name__ = "mock_connect"
        mock_connect.side_effect = ValueError("should not retry this " "error")
        retry_request = Retry(timeout=5, retry_delay=0.1, backoff=1)

        self.assertRaisesRegex(
            ValueError,
            "should not retry this error",
            retry_request(con.request),
            action="/",
        )
        self.assertEqual(mock_connect.call_count, 1, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_rate_limit_error_success_on_second_attempt(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        self.retry_counter = 0

        def mock_connect_side_effect(*args, **kwargs):
            self.retry_counter += 1

            if self.retry_counter < 2:
                headers = {"retry-after": 0.2}
                raise RateLimitReachedError(headers=headers)

            return "success"

        mock_connect.__name__ = "mock_connect"
        mock_connect.side_effect = mock_connect_side_effect
        retry_request = Retry(timeout=0.6, retry_delay=0.1, backoff=1)
        result = retry_request(con.request)(action="/")
        self.assertEqual(result, "success")

        self.assertEqual(mock_connect.call_count, 2, "Retry logic failed")

    @patch("libcloud.common.base.Connection.request")
    def test_retry_on_all_default_retry_exception_classes(self, mock_connect):
        con = Connection()
        con.connection = Mock()

        self.retry_counter = 0

        def mock_connect_side_effect(*args, **kwargs):
            self.retry_counter += 1

            if self.retry_counter < len(RETRY_EXCEPTIONS):
                raise RETRY_EXCEPTIONS[self.retry_counter]

            return "success"

        mock_connect.__name__ = "mock_connect"
        mock_connect.side_effect = mock_connect_side_effect
        retry_request = Retry(timeout=0.6, retry_delay=0.1, backoff=1)
        result = retry_request(con.request)(action="/")
        self.assertEqual(result, "success")

        self.assertEqual(
            mock_connect.call_count, len(RETRY_EXCEPTIONS), "Retry logic failed"
        )


class CertificateConnectionClassTestCase(unittest.TestCase):
    def setUp(self):
        self.connection = CertificateConnection(
            cert_file="test.pem", url="https://test.com/test"
        )
        self.connection.connect()

    def test_adapter_internals(self):
        adapter = self.connection.connection.session.adapters["https://"]
        self.assertTrue(isinstance(adapter, SignedHTTPSAdapter))
        self.assertEqual(adapter.cert_file, "test.pem")


if __name__ == "__main__":
    sys.exit(unittest.main())