# Copyright 2016-2017 Capital One Services, LLC
# Copyright The Cloud Custodian Authors.
# SPDX-License-Identifier: Apache-2.0
import jmespath
from .common import BaseTest
from c7n.utils import local_session
from unittest.mock import MagicMock


class CloudFrontWaf(BaseTest):

    def test_waf(self):
        factory = self.replay_flight_data("test_distribution_waf")
        p = self.load_policy(
            {
                "name": "waf-cfront",
                "resource": "distribution",
                "filters": [{"type": "waf-enabled", "web-acl": "test", "state": False}],
                "actions": [{"type": "set-waf", "web-acl": "test", "state": True}],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 2)

        p = self.load_policy(
            {
                "name": "waf-cfront",
                "resource": "distribution",
                "filters": [{"type": "waf-enabled", "web-acl": "test", "state": False}],
            },
            session_factory=factory,
        )
        self.assertEqual(p.run(), [])


class CloudFront(BaseTest):

    def test_shield_metric_filter(self):
        factory = self.replay_flight_data("test_distribution_shield_metrics")
        p = self.load_policy(
            {
                "name": "ddos-filter",
                "resource": "distribution",
                "filters": [
                    {
                        "type": "shield-metrics",
                        "name": "DDoSDetected",
                        "value": 1,
                        "op": "ge",
                    }
                ],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 0)

    def test_distribution_metric_filter(self):
        factory = self.replay_flight_data("test_distribution_metric_filter")
        p = self.load_policy(
            {
                "name": "requests-filter",
                "resource": "distribution",
                "filters": [
                    {"type": "metrics", "name": "Requests", "value": 3, "op": "ge"}
                ],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(resources[0]["DomainName"], "d32plmcrnvwzrd.cloudfront.net")

    def test_distribution_set_ssl(self):
        factory = self.replay_flight_data("test_distrbution_set_ssl")

        k = "DefaultCacheBehavior.ViewerProtocolPolicy"

        p = self.load_policy(
            {
                "name": "distribution-set-ssl",
                "resource": "distribution",
                "filters": [
                    {"type": "value", "key": k, "value": "allow-all", "op": "contains"}
                ],
                "actions": [
                    {"type": "set-protocols", "ViewerProtocolPolicy": "https-only"}
                ],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 1)
        expr = jmespath.compile(k)
        r = expr.search(resources[0])
        self.assertTrue("allow-all" in r)

        client = local_session(factory).client("cloudfront")
        resp = client.list_distributions()
        self.assertEqual(
            resp["DistributionList"]["Items"][0]["DefaultCacheBehavior"][
                "ViewerProtocolPolicy"
            ],
            "https-only",
        )

    def test_distribution_custom_origin(self):
        factory = self.replay_flight_data("test_distrbution_custom_origin")

        k = "Origins.Items[].CustomOriginConfig.OriginSslProtocols.Items[]"

        p = self.load_policy(
            {
                "name": "distribution-set-ssl",
                "resource": "distribution",
                "filters": [
                    {"type": "value", "key": k, "value": "TLSv1", "op": "contains"}
                ],
                "actions": [
                    {
                        "type": "set-protocols",
                        "OriginSslProtocols": ["TLSv1.1", "TLSv1.2"],
                        "OriginProtocolPolicy": "https-only",
                    }
                ],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 1)
        expr = jmespath.compile(k)
        r = expr.search(resources[0])
        self.assertTrue("TLSv1" in r)

        client = local_session(factory).client("cloudfront")
        resp = client.list_distributions()
        self.assertEqual(
            resp["DistributionList"]["Items"][0]["Origins"]["Items"][0][
                "CustomOriginConfig"
            ][
                "OriginProtocolPolicy"
            ],
            "https-only",
        )
        self.assertTrue(
            "TLSv1.2" in resp["DistributionList"]["Items"][0]["Origins"]["Items"][0][
                "CustomOriginConfig"
            ][
                "OriginSslProtocols"
            ][
                "Items"
            ]
        )
        self.assertFalse(
            "TLSv1" in resp["DistributionList"]["Items"][0]["Origins"]["Items"][0][
                "CustomOriginConfig"
            ][
                "OriginSslProtocols"
            ][
                "Items"
            ]
        )

    def test_distribution_disable(self):
        factory = self.replay_flight_data("test_distrbution_disable")

        p = self.load_policy(
            {
                "name": "distribution-disable",
                "resource": "distribution",
                "filters": [
                    {
                        "type": "value",
                        "key": "DefaultCacheBehavior.ViewerProtocolPolicy",
                        "value": "allow-all",
                        "op": "contains",
                    }
                ],
                "actions": [{"type": "disable"}],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["Enabled"], True)

        client = local_session(factory).client("cloudfront")
        resp = client.list_distributions()
        self.assertEqual(resp["DistributionList"]["Items"][0]["Enabled"], False)

    def test_distribution_check_s3_origin_missing_bucket(self):
        factory = self.replay_flight_data("test_distribution_check_s3_origin_missing_bucket")

        p = self.load_policy(
            {
                "name": "test_distribution_check_s3_origin_missing_bucket",
                "resource": "distribution",
                "filters": [
                    {
                        "type": "mismatch-s3-origin",
                    }
                ]
            },
            session_factory=factory,
        )

        resources = p.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]['c7n:mismatched-s3-origin'][0], 'c7n-idontexist')

    def test_distribution_check_logging_enabled(self):
        factory = self.replay_flight_data("test_distribution_check_logging_enabled")

        p = self.load_policy(
            {
                "name": "test_distribution_logging_enabled",
                "resource": "distribution",
                "filters": [
                    {
                        "type": "distribution-config",
                        "key": "Logging.Enabled",
                        "value": True
                    }
                ]
            },
            session_factory=factory,
        )

        resources = p.run()

        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]['c7n:distribution-config']['Logging']['Enabled'], True)

    def test_distribution_check_logging_enabled_error(self):
        factory = self.replay_flight_data("test_distribution_check_logging_enabled")

        client = factory().client("cloudfront")
        mock_factory = MagicMock()
        mock_factory.region = 'us-east-1'
        mock_factory().client(
            'cloudfront').exceptions.NoSuchDistribution = (
                client.exceptions.NoSuchDistribution)

        mock_factory().client('cloudfront').get_distribution_config.side_effect = (
            client.exceptions.NoSuchDistribution(
                {'Error': {'Code': 'xyz'}},
                operation_name='get_distribution_config'))
        p = self.load_policy(
            {
                "name": "test_distribution_logging_enabled",
                "resource": "distribution",
                "filters": [
                    {
                        "type": "distribution-config",
                        "key": "Logging.Enabled",
                        "value": True
                    }
                ]
            },
            session_factory=mock_factory,
        )

        try:
            p.resource_manager.filters[0].process(
                [{'Id': 'abc'}])
        except client.exceptions.NoSuchDistribution:
            self.fail('should not raise')
        mock_factory().client('cloudfront').get_distribution_config.assert_called_once()

    def test_streaming_distribution_check_logging_enabled(self):
        factory = self.replay_flight_data("test_streaming_distribution_check_logging_enabled")

        p = self.load_policy(
            {
                "name": "test_streaming_distribution_logging_enabled",
                "resource": "streaming-distribution",
                "filters": [
                    {
                        "type": "distribution-config",
                        "key": "Logging.Enabled",
                        "value": True
                    }
                ]
            },
            session_factory=factory,
        )

        resources = p.run()

        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]['c7n:distribution-config']['Logging']['Enabled'],
            True)

    def test_streaming_distribution_check_logging_enabled_error(self):
        factory = self.replay_flight_data("test_streaming_distribution_check_logging_enabled")

        client = factory().client("cloudfront")
        mock_factory = MagicMock()
        mock_factory.region = 'us-east-1'
        mock_factory().client(
            'cloudfront').exceptions.NoSuchStreamingDistribution = (
                client.exceptions.NoSuchStreamingDistribution)

        mock_factory().client('cloudfront').get_streaming_distribution_config.side_effect = (
            client.exceptions.NoSuchStreamingDistribution(
                {'Error': {'Code': 'xyz'}},
                operation_name='get_streaming_distribution_config'))
        p = self.load_policy(
            {
                "name": "test_streaming_distribution_logging_enabled",
                "resource": "streaming-distribution",
                "filters": [
                    {
                        "type": "distribution-config",
                        "key": "Logging.Enabled",
                        "value": True
                    }
                ]
            },
            session_factory=mock_factory,
        )

        try:
            p.resource_manager.filters[0].process(
                [{'Id': 'abc'}])
        except client.exceptions.NoSuchDistribution:
            self.fail('should not raise')
        mock_factory().client('cloudfront').get_streaming_distribution_config.assert_called_once()

    def test_distribution_tag(self):
        factory = self.replay_flight_data("test_distrbution_tag")

        p = self.load_policy(
            {
                "name": "distribution-tag",
                "resource": "distribution",
                "filters": [{"tag:abc": "present"}],
                "actions": [{"type": "tag", "key": "123", "value": "456"}],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 1)

        client = local_session(factory).client("cloudfront")
        resp = client.list_tags_for_resource(Resource=resources[0]["ARN"])
        self.assertEqual(len(resp["Tags"]["Items"]), 2)

    def test_streaming_distribution_disable(self):
        factory = self.replay_flight_data("test_streaming_distrbution_disable")

        p = self.load_policy(
            {
                "name": "streaming-distribution-disable",
                "resource": "streaming-distribution",
                "filters": [
                    {
                        "type": "value",
                        "key": "S3Origin.OriginAccessIdentity",
                        "value": "",
                    }
                ],
                "actions": [{"type": "disable"}],
            },
            session_factory=factory,
        )
        resources = p.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["Enabled"], True)

        client = local_session(factory).client("cloudfront")
        resp = client.list_streaming_distributions()
        self.assertEqual(
            resp["StreamingDistributionList"]["Items"][0]["Enabled"], False
        )

    def test_streaming_distribution_tag(self):
        factory = self.replay_flight_data("test_streaming_distrbution_tag")

        p = self.load_policy(
            {