# Copyright 2015-2017 Capital One Services, LLC
# Copyright The Cloud Custodian Authors.
# SPDX-License-Identifier: Apache-2.0
from .common import BaseTest

from c7n.exceptions import PolicyValidationError
from c7n.executor import MainThreadExecutor
from c7n.resources.elb import ELB, SetSslListenerPolicy


class ELBTagTest(BaseTest):

    def test_elb_tag_and_remove(self):
        self.patch(ELB, "executor_factory", MainThreadExecutor)
        session_factory = self.replay_flight_data("test_elb_tag_and_remove")
        client = session_factory().client("elb")

        policy = self.load_policy(
            {
                "name": "elb-tag",
                "resource": "elb",
                "filters": [{"LoadBalancerName": "CloudCustodian"}],
                "actions": [{"type": "tag", "key": "xyz", "value": "abdef"}],
            },
            config={"account_id": "644160558196"},
            session_factory=session_factory,
        )

        resources = policy.run()
        self.assertEqual(len(resources), 1)
        tags = client.describe_tags(LoadBalancerNames=["CloudCustodian"])[
            "TagDescriptions"
        ][
            0
        ][
            "Tags"
        ]
        tag_map = {t["Key"]: t["Value"] for t in tags}
        self.assertTrue("xyz" in tag_map)

        policy = self.load_policy(
            {
                "name": "elb-tag",
                "resource": "elb",
                "filters": [{"LoadBalancerName": "CloudCustodian"}],
                "actions": [{"type": "remove-tag", "tags": ["xyz"]}],
            },
            config={"account_id": "644160558196"},
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        tags = client.describe_tags(LoadBalancerNames=["CloudCustodian"])[
            "TagDescriptions"
        ][
            0
        ][
            "Tags"
        ]
        tag_map = {t["Key"]: t["Value"] for t in tags}
        self.assertFalse("xyz" in tag_map)

    def test_elb_tags(self):
        self.patch(ELB, "executor_factory", MainThreadExecutor)
        session_factory = self.replay_flight_data("test_elb_tags")
        policy = self.load_policy(
            {
                "name": "elb-mark",
                "resource": "elb",
                "filters": [{"tag:Platform": "ubuntu"}],
            },
            config={"account_id": "644160558196"},
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["LoadBalancerName"], "CloudCustodian")

    def test_mark_and_match(self):
        session_factory = self.replay_flight_data("test_elb_mark_and_match")
        policy = self.load_policy(
            {
                "name": "elb-mark",
                "resource": "elb",
                "filters": [{"LoadBalancerName": "CloudCustodian"}],
                "actions": [
                    {
                        "type": "mark-for-op",
                        "op": "delete",
                        "tag": "custodian_next",
                        "days": 1,
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()

        self.assertEqual(len(resources), 1)
        tags = session_factory().client("elb").describe_tags(
            LoadBalancerNames=["CloudCustodian"]
        )[
            "TagDescriptions"
        ][
            0
        ][
            "Tags"
        ]
        tag_map = {t["Key"]: t["Value"] for t in tags}
        self.assertTrue("custodian_next" in tag_map)

        policy = self.load_policy(
            {
                "name": "elb-mark-filter",
                "resource": "elb",
                "filters": [
                    {"type": "marked-for-op", "tag": "custodian_next", "op": "delete"}
                ],
            },
            config={"account_id": "644160558196"},
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class ELBInstance(BaseTest):

    def test_instance_filter(self):
        session_factory = self.replay_flight_data("test_elb_instance_filter")
        policy = self.load_policy(
            {
                "name": "elb-instance",
                "resource": "elb",
                "filters": [
                    {"type": "instance", "key": "ImageId", "value": "ami-40d28157"}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["LoadBalancerName"], "balanced")


class HealthCheckProtocolMismatchTest(BaseTest):

    def test_healthcheck_protocol_mismatch(self):
        session_factory = self.replay_flight_data("test_healthcheck_protocol_mismatch")
        policy = self.load_policy(
            {
                "name": "healthcheck-protocol-mismatch",
                "resource": "elb",
                "filters": [{"type": "healthcheck-protocol-mismatch"}],
            },
            session_factory=session_factory,
        )

        resources = policy.run()
        self.assertEqual(len(resources), 3)

        # make sure we matched the right load balcners
        elb_names = {elb["LoadBalancerName"] for elb in resources}
        self.assertEqual(
            elb_names,
            {
                "test-elb-no-listeners",
                "test-elb-protocol-matches",
                "test-elb-multiple-listeners",
            },
        )


class SSLPolicyTest(BaseTest):

    def test_ssl_ciphers(self):
        session_factory = self.replay_flight_data("test_ssl_ciphers")
        policy = self.load_policy(
            {
                "name": "test-ssl-ciphers",
                "resource": "elb",
                "filters": [{"type": "ssl-policy", "blacklist": ["Protocol-SSLv2"]}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["LoadBalancerName"], "test-elb-invalid-policy")

    def test_set_ssl_listener_policy_fail(self):
        session_factory = self.replay_flight_data("test_set_ssl_listener")
        self.patch(SetSslListenerPolicy, 'process_elb', lambda self, client, elb: elb.xyz)

        policy = self.load_policy({
            "name": "test-set-ssl-listerner",
            "resource": "elb",
            "filters": [{'LoadBalancerName': 'test-elb'}],
            "actions": [{
                "type": "set-ssl-listener-policy",
                "name": "testpolicy",
                "attributes": ["AES128-SHA256", "Protocol-TLSv1"]}]},
            session_factory=session_factory)
        self.assertRaises(AttributeError, policy.run)

    def test_set_ssl_listener_policy(self):
        session_factory = self.replay_flight_data("test_set_ssl_listener")
        client = session_factory().client("elb")
        policy = self.load_policy(
            {
                "name": "test-set-ssl-listerner",
                "resource": "elb",
                "filters": [
                    {
                        "type": "ssl-policy",
                        "whitelist": ["AES128-SHA256", "Protocol-TLSv1"],
                    },
                    {
                        "type": "value",
                        "key": "LoadBalancerName",
                        "value": "test-elb",
                        "op": "eq",
                    },
                ],
                "actions": [
                    {
                        "type": "set-ssl-listener-policy",
                        "name": "testpolicy",
                        "attributes": ["AES128-SHA256", "Protocol-TLSv1"],
                    }
                ],
            },
            session_factory=session_factory,
        )
        policy.run()
        response_pol = client.describe_load_balancers(LoadBalancerNames=["test-elb"])
        response_ciphers = client.describe_load_balancer_policies(
            LoadBalancerName="test-elb", PolicyNames=["testpolicy-1493768308000"]
        )
        curr_pol = response_pol["LoadBalancerDescriptions"][0]["ListenerDescriptions"][
            0
        ][
            "PolicyNames"
        ]

        curr_ciphers = []
        for x in response_ciphers["PolicyDescriptions"][0][
            "PolicyAttributeDescriptions"
        ]:
            curr_ciphers.append({str(k): str(v) for k, v in x.items()})
        active_ciphers = [
            x["AttributeName"] for x in curr_ciphers if x["AttributeValue"] == "true"
        ]
        self.assertEqual(
            curr_pol,
            [
                "AWSConsole-LBCookieStickinessPolicy-test-elb-1493748038333",
                "testpolicy-1493768308000",
            ],
        )
        self.assertEqual(active_ciphers, ["Protocol-TLSv1", "AES128-SHA256"])

    def test_ssl_matching(self):
        session_factory = self.replay_flight_data("test_ssl_ciphers")
        policy = self.load_policy(
            {
                "name": "test-ssl-matching",
                "resource": "elb",
                "filters": [
                    {
                        "type": "ssl-policy",
                        "matching": "^Protocol-",
                        "whitelist": [
                            "Protocol-TLSv1", "Protocol-TLSv1.1", "Protocol-TLSv1.2"
                        ],
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["LoadBalancerName"], "test-elb-invalid-policy")

    def test_filter_validation_no_blacklist(self):
        self.assertRaises(
            PolicyValidationError,
            self.load_policy,
            {
                "name": "test-ssl-ciphers",
                "resource": "elb",
                "filters": [{"type": "ssl-policy"}],
            },
            session_factory=None,
            validate=False,
        )

    def test_filter_validation_blacklist_not_iterable(self):
        self.assertRaises(
            PolicyValidationError,
            self.load_policy,
            {
                "name": "test-ssl-ciphers",
                "resource": "elb",
                "filters": [{"type": "ssl-policy", "blacklist": "single-value"}],
            },
            session_factory=None,
            validate=False,
        )


class TestDefaultVpc(BaseTest):

    def test_elb_default_vpc(self):
        session_factory = self.replay_flight_data("test_elb_default_vpc")
        p = self.load_policy(
            {
                "name": "elb-default-filters",
                "resource": "elb",
                "filters": [{"type": "default-vpc"}],
            },
            config={"region": "us-west-2"},
            session_factory=session_factory,
        )

        resources = p.run()
        self.assertEqual(len(resources), 1)