# Copyright The Cloud Custodian Authors.
# SPDX-License-Identifier: Apache-2.0
import logging
import unittest
import time

import datetime
from dateutil import tz
import jmespath
from mock import mock

from c7n.testing import mock_datetime_now
from c7n.exceptions import PolicyValidationError, ClientError
from c7n.resources import ec2
from c7n.resources.ec2 import actions, QueryFilter
from c7n import tags, utils

from .common import BaseTest


class TestEc2NetworkLocation(BaseTest):
    def test_ec2_network_location_terminated(self):
        factory = self.replay_flight_data("test_ec2_network_location")
        client = factory().client('ec2')
        resp = client.describe_instances()

        self.assertTrue(len(resp['Reservations'][0]['Instances']), 1)
        self.assertTrue(
            len(resp['Reservations'][0]['Instances'][0]['State']['Name']),
            'terminated'
        )

        policy = self.load_policy(
            {
                'name': 'ec2-network-location',
                'resource': 'ec2',
                'filters': [
                    {'State.Name': 'terminated'},
                    {'type': 'network-location',
                     "key": "tag:some-value"}
                ]
            },
            session_factory=factory
        )
        resources = policy.run()
        self.assertEqual(len(resources), 0)


class TestTagAugmentation(BaseTest):

    def test_tag_augment_empty(self):
        session_factory = self.replay_flight_data("test_ec2_augment_tag_empty")
        # recording was modified to be sans tags
        policy = self.load_policy(
            {"name": "ec2-tags", "resource": "ec2"}, session_factory=session_factory
        )
        resources = policy.run()
        self.assertEqual(len(resources), 0)

    def test_tag_augment(self):
        session_factory = self.replay_flight_data("test_ec2_augment_tags")
        # recording was modified to be sans tags
        policy = self.load_policy(
            {
                "name": "ec2-tags",
                "resource": "ec2",
                "filters": [{"tag:Env": "Production"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class TestInstanceAttrFilter(BaseTest):

    def test_attr_filter(self):
        session_factory = self.replay_flight_data("test_ec2_instance_attribute")
        policy = self.load_policy(
            {
                "name": "ec2-attr",
                "resource": "ec2",
                "filters": [
                    {
                        "type": "instance-attribute",
                        "attribute": "rootDeviceName",
                        "key": "Value",
                        "value": "/dev/sda1",
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(
            resources[0]["c7n:attribute-rootDeviceName"], {"Value": "/dev/sda1"}
        )


class TestSetMetadata(BaseTest):

    def test_set_metadata_server(self):
        output = self.capture_logging('custodian.actions')
        session_factory = self.replay_flight_data('test_ec2_set_md_access')
        policy = self.load_policy({
            'name': 'ec2-imds-access',
            'resource': 'aws.ec2',
            'actions': [
                {'type': 'set-metadata-access',
                 'tokens': 'required'},
            ]},
            session_factory=session_factory)
        resources = policy.run()
        if self.recording:
            time.sleep(2)
        results = session_factory().client('ec2').describe_instances(
            InstanceIds=[r['InstanceId'] for r in resources])
        self.assertJmes('[0].MetadataOptions.HttpTokens', resources, 'optional')
        self.assertJmes(
            'Reservations[].Instances[].MetadataOptions',
            results,
            [{'HttpEndpoint': 'enabled',
              'HttpPutResponseHopLimit': 1,
              'HttpTokens': 'required',
              'State': 'pending'},
             {'HttpEndpoint': 'enabled',
              'HttpPutResponseHopLimit': 1,
              'HttpTokens': 'required',
              'State': 'applied'}])
        self.assertEqual(len(resources), 2)
        self.assertEqual(
            output.getvalue(),
            ('set-metadata-access implicitly filtered 1 of 2 resources '
             'key:MetadataOptions.HttpTokens on optional\n'))


class TestMetricFilter(BaseTest):

    def test_metric_filter(self):
        session_factory = self.replay_flight_data("test_ec2_metric")
        policy = self.load_policy(
            {
                "name": "ec2-utilization",
                "resource": "ec2",
                "filters": [
                    {
                        "type": "metrics",
                        "name": "CPUUtilization",
                        "days": 3,
                        "value": 1.5,
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class TestPropagateSpotTags(BaseTest):

    def test_propagate_spot(self):
        session_factory = self.replay_flight_data("test_ec2_propagate_spot_tags")

        policy = self.load_policy(
            {
                "name": "ec2-spot",
                "resource": "ec2",
                "query": [{"instance-id": "i-01db165f1452ef5e4"}],
                "actions": [{"type": "propagate-spot-tags", "only_tags": ["Name"]}],
            },
            session_factory=session_factory,
        )

        policy.run()
        client = session_factory().client("ec2")
        tags = {
            t["Key"]: t["Value"]
            for t in client.describe_tags(
                Filters=[{"Name": "resource-id", "Values": ["i-01db165f1452ef5e4"]}]
            ).get(
                "Tags", []
            )
        }
        self.assertEqual(tags, {"Name": "Test"})


class TestDisableApiTermination(BaseTest):

    def test_term_prot_enabled(self):
        session_factory = self.replay_flight_data(
            "test_ec2_termination-protected_filter"
        )
        policy = self.load_policy(
            {
                "name": "ec2-termination-enabled",
                "resource": "ec2",
                "filters": [{"type": "termination-protected"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["InstanceId"], "i-092f500eaad726b71")

    def test_term_prot_not_enabled(self):
        session_factory = self.replay_flight_data(
            "test_ec2_termination-protected_filter"
        )
        policy = self.load_policy(
            {
                "name": "ec2-termination-NOT-enabled",
                "resource": "ec2",
                "filters": [{"not": [{"type": "termination-protected"}]}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 2)
        self.assertEqual(
            sorted([x["InstanceId"] for x in resources]),
            ["i-02117c13e1d21b229", "i-0718418de3bb4ae2a"],
        )

    def test_policy_permissions(self):
        session_factory = self.replay_flight_data(
            "test_ec2_termination-protected_filter"
        )
        policy = self.load_policy(
            {
                "name": "ec2-termination-enabled",
                "resource": "ec2",
                "filters": [{"type": "termination-protected"}],
            },
            session_factory=session_factory,
        )
        perms = policy.get_permissions()
        self.assertEqual(
            perms,
            {
                "ec2:DescribeInstances",
                "ec2:DescribeTags",
                "ec2:DescribeInstanceAttribute",
            },
        )


class TestEc2Permissions(BaseTest):

    def test_ec2_permissions(self):
        factory = self.replay_flight_data('test_ec2_permissions')
        policy = self.load_policy({
            'name': 'ec2-perm',
            'resource': 'aws.ec2',
            'filters': [{
                'type': 'check-permissions',
                'match': 'allowed',
                'actions': ['lambda:CreateFunction']}]},
            session_factory=factory, config={'region': 'us-west-2'})
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertTrue('c7n:perm-matches' in resources[0])


class TestSsm(BaseTest):

    def test_ssm_status(self):
        session_factory = self.replay_flight_data('test_ec2_ssm_filter')
        policy = self.load_policy({
            'name': 'ec2-ssm',
            'resource': 'aws.ec2',
            'filters': [
                {'type': 'ssm',
                 'key': 'PlatformName',
                 'value': 'Ubuntu'},
                {'type': 'ssm',
                 'key': 'PingStatus',
                 'value': 'Online'}]},
            session_factory=session_factory,
            config={'region': 'us-east-2'})
        resources = policy.run()
        self.assertEqual(len(resources), 2)
        self.assertTrue('c7n:SsmState' in resources[0])
        self.assertEqual(
            [r['InstanceId'] for r in resources],
            ['i-0dea82d960d56dc1d', 'i-0ba3874e85bb97244'])

    def test_ssm_compliance(self):
        session_factory = self.replay_flight_data('test_ec2_ssm_compliance_filter')
        policy = self.load_policy({
            'name': 'ec2-ssm-compliance',
            'resource': 'aws.ec2',
            'filters': [
                {'type': 'ssm-compliance',
                 'compliance_types': [
                     'Association',
                     'Patch'
                 ],
                 'severity': [
                     'CRITICAL',
                     'HIGH',
                     'MEDIUM',
                     'LOW',
                     'UNSPECIFIED'
                 ],
                 'states': ['NON_COMPLIANT']}]},
            session_factory=session_factory,
            config={'region': 'us-east-2'})
        resources = policy.run()
        self.assertEqual(len(resources), 2)
        self.assertTrue('c7n:ssm-compliance' in resources[0])
        self.assertEqual(
            [r['InstanceId'] for r in resources],
            ['i-0dea82d960d56dc1d', 'i-0ba3874e85bb97244'])


class TestHealthEventsFilter(BaseTest):

    def test_ec2_health_events_filter(self):
        session_factory = self.replay_flight_data("test_ec2_health_events_filter")
        policy = self.load_policy(
            {
                "name": "ec2-health-events-filter",
                "resource": "ec2",
                "filters": [{"type": "health-event"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class TestTagTrim(BaseTest):

    def test_ec2_tag_trim(self):
        self.patch(tags.TagTrim, "max_tag_count", 10)
        session_factory = self.replay_flight_data("test_ec2_tag_trim")
        ec2 = session_factory().client("ec2")
        start_tags = {
            t["Key"]: t["Value"]
            for t in ec2.describe_tags(
                Filters=[{"Name": "resource-id", "Values": ["i-fdb01920"]}]
            )[
                "Tags"
            ]
        }
        policy = self.load_policy(
            {
                "name": "ec2-tag-trim",
                "resource": "ec2",
                "filters": [{"type": "tag-count", "count": 10}],
                "actions": [
                    {
                        "type": "tag-trim",
                        "space": 1,
                        "preserve": [
                            "Name",
                            "Env",
                            "Account",
                            "Platform",
                            "Classification",
                            "Planet",
                        ],
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        end_tags = {
            t["Key"]: t["Value"]
            for t in ec2.describe_tags(
                Filters=[{"Name": "resource-id", "Values": ["i-fdb01920"]}]
            )[
                "Tags"
            ]
        }

        self.assertEqual(len(start_tags) - 1, len(end_tags))
        self.assertTrue("Containers" in start_tags)
        self.assertFalse("Containers" in end_tags)


class TestVolumeFilter(BaseTest):

    def test_ec2_attached_ebs_filter(self):
        session_factory = self.replay_flight_data("test_ec2_attached_ebs_filter")
        policy = self.load_policy(
            {
                "name": "ec2-unencrypted-vol",
                "resource": "ec2",
                "filters": [{"type": "ebs", "key": "Encrypted", "value": False}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    # DISABLED / Re-record flight data on public account
    def test_ec2_attached_volume_skip_block(self):
        session_factory = self.replay_flight_data("test_ec2_attached_ebs_filter")
        policy = self.load_policy(
            {
                "name": "ec2-unencrypted-vol",
                "resource": "ec2",
                "filters": [
                    {
                        "type": "ebs",
                        "skip-devices": ["/dev/sda1", "/dev/xvda", "/dev/sdb1"],
                        "key": "Encrypted",
                        "value": False,
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 0)


class TestResizeInstance(BaseTest):

    def test_ec2_resize(self):
        # preconditions - three instances (2 m4.4xlarge, 1 m4.1xlarge)
        # one of the instances stopped
        session_factory = self.replay_flight_data("test_ec2_resize")
        policy = self.load_policy(
            {
                "name": "ec2-resize",
                "resource": "ec2",
                "filters": [
                    {
                        "type": "value",
                        "key": "State.Name",
                        "value": ["running", "stopped"],
                        "op": "in",
                    },
                    {
                        "type": "value",
                        "key": "InstanceType",
                        "value": ["m4.2xlarge", "m4.4xlarge"],
                        "op": "in",
                    },
                ],
                "actions": [
                    {
                        "type": "resize",
                        "restart": True,
                        "default": "m4.large",
                        "type-map": {"m4.4xlarge": "m4.2xlarge"},
                    }
                ],
            },
            session_factory=session_factory,
        )

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

        stopped, running = [], []
        for i in resources:
            if i["State"]["Name"] == "running":
                running.append(i["InstanceId"])
            if i["State"]["Name"] == "stopped":
                stopped.append(i["InstanceId"])

        instances = utils.query_instances(
            session_factory(), InstanceIds=[r["InstanceId"] for r in resources]
        )

        cur_stopped, cur_running = [], []
        for i in instances:
            if i["State"]["Name"] == "running":
                cur_running.append(i["InstanceId"])
            if i["State"]["Name"] == "stopped":
                cur_stopped.append(i["InstanceId"])

        cur_running.sort()
        running.sort()

        self.assertEqual(cur_stopped, stopped)
        self.assertEqual(cur_running, running)
        instance_types = [i["InstanceType"] for i in instances]
        instance_types.sort()
        self.assertEqual(
            instance_types, list(sorted(["m4.large", "m4.2xlarge", "m4.2xlarge"]))
        )


class TestStateTransitionAgeFilter(BaseTest):

    def test_ec2_state_transition_age(self):
        session_factory = self.replay_flight_data(
            "test_ec2_state_transition_age_filter"
        )
        policy = self.load_policy(
            {
                "name": "ec2-state-transition-age",
                "resource": "ec2",
                "filters": [
                    {"State.Name": "running"}, {"type": "state-age", "days": 30}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        # compare stateTransition reason to expected
        self.assertEqual(len(resources), 1)
        self.assertEqual(
            resources[0]["StateTransitionReason"],
            "User initiated (2015-11-25 10:11:55 GMT)",
        )

    def test_date_parsing(self):
        instance = ec2.StateTransitionAge(None)

        # Missing key
        self.assertIsNone(instance.get_resource_date({}))

        # Bad date format
        self.assertRaises(
            ValueError,
            instance.get_resource_date,
            {"StateTransitionReason": "User initiated (201-02-06 17:77:00 GMT)"},
        )

        # Won't match regex
        self.assertIsNone(
            instance.get_resource_date(
                {"StateTransitionReason": "Server.InternalError"}
            )
        )

        # Test for success
        self.assertEqual(
            instance.get_resource_date(
                {"StateTransitionReason": "User initiated (2017-02-06 17:57:00 GMT)"}
            ),
            datetime.datetime(2017, 2, 6, 17, 57, tzinfo=tz.tzutc()),
        )


class TestImageAgeFilter(BaseTest):

    def test_ec2_image_age(self):
        session_factory = self.replay_flight_data("test_ec2_image_age_filter")
        policy = self.load_policy(
            {
                "name": "ec2-image-age",
                "resource": "ec2",
                "filters": [
                    {"State.Name": "running"}, {"type": "image-age", "days": 30}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class TestImageFilter(BaseTest):

    def test_ec2_image(self):
        session_factory = self.replay_flight_data("test_ec2_image_filter")
        policy = self.load_policy(
            {
                "name": "ec2-image",
                "resource": "ec2",
                "filters": [{"type": "image", "key": "Public", "value": True}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["InstanceId"], "i-039628786cabe8c16")


class TestInstanceAge(BaseTest):

    # placebo doesn't record tz information
    def test_ec2_instance_age(self):
        session_factory = self.replay_flight_data("test_ec2_instance_age_filter")
        policy = self.load_policy(
            {
                "name": "ec2-instance-age",
                "resource": "ec2",
                "filters": [
                    {"State.Name": "running"}, {"type": "instance-age", "days": 0}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)


class TestTag(BaseTest):

    def test_ec2_tag(self):
        session_factory = self.replay_flight_data("test_ec2_mark")
        policy = self.load_policy(
            {
                "name": "ec2-test-mark",
                "resource": "ec2",
                "filters": [{"State.Name": "running"}],
                "actions": [{"type": "tag", "key": "Testing", "value": "Testing123"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    def test_ec2_tag_errors(self):
        # Specifying both 'key' and 'tag' is an error
        policy = {
            "name": "ec2-tag-error",
            "resource": "ec2",
            "actions": [
                {"type": "tag", "key": "Testing", "tag": "foo", "value": "TestingError"}
            ],
        }
        self.assertRaises(PolicyValidationError, self.load_policy, policy)

        # Invalid op for 'mark-for-op' action
        policy = {
            "name": "ec2-tag-error",
            "resource": "ec2",
            "actions": [{"type": "mark-for-op", "op": "fake"}],
        }
        self.assertRaises(PolicyValidationError, self.load_policy, policy)

    def test_ec2_untag(self):
        session_factory = self.replay_flight_data("test_ec2_untag")
        policy = self.load_policy(
            {
                "name": "ec2-test-unmark",
                "resource": "ec2",
                "filters": [{"tag:Testing": "not-null"}],
                "actions": [{"type": "remove-tag", "tags": ["Testing"]}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    def test_ec2_untag_array(self):
        session_factory = self.replay_flight_data("test_ec2_untag_array")
        policy = self.load_policy(
            {
                "name": "ec2-test-unmark-array",
                "resource": "ec2",
                "filters": [{"tag:Testing": "not-null"}],
                "actions": [
                    {
                        "type": "remove-tag",
                        "tags": ["Testing", "TestingTwo", "TestingThree"],
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    def test_ec2_normalize_tag(self):
        session_factory = self.replay_flight_data("test_ec2_normalize_tag")

        policy = self.load_policy(
            {
                "name": "ec2-test-normalize-tag-lower",
                "resource": "ec2",
                "filters": [{"tag:Testing-lower": "not-null"}],
                "actions": [
                    {"type": "normalize-tag", "key": "Testing-lower", "action": "lower"}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

        policy = self.load_policy(
            {
                "name": "ec2-test-normalize-tag-upper",
                "resource": "ec2",
                "filters": [{"tag:Testing-upper": "not-null"}],
                "actions": [
                    {"type": "normalize-tag", "key": "Testing-upper", "action": "upper"}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

        policy = self.load_policy(
            {
                "name": "ec2-test-normalize-tag-title",
                "resource": "ec2",
                "filters": [{"tag:Testing-title": "not-null"}],
                "actions": [
                    {"type": "normalize-tag", "key": "Testing-title", "action": "title"}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

        policy = self.load_policy(
            {
                "name": "ec2-test-normalize-tag-strip",
                "resource": "ec2",
                "filters": [{"tag:Testing-strip": "not-null"}],
                "actions": [
                    {
                        "type": "normalize-tag",
                        "key": "Testing-strip",
                        "action": "strip",
                        "value": "blah",
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    def test_ec2_rename_tag(self):
        session_factory = self.replay_flight_data("test_ec2_rename_tag")

        policy = self.load_policy(
            {
                "name": "ec2-rename-start",
                "resource": "ec2",
                "filters": [{"tag:Testing": "present"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 3)

        policy = self.load_policy(
            {
                "name": "ec2-rename-tag",
                "resource": "ec2",
                "actions": [
                    {"type": "rename-tag", "old_key": "Testing", "new_key": "Testing1"}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 3)

        policy = self.load_policy(
            {
                "name": "ec2-rename-end",
                "resource": "ec2",
                "filters": [{"tag:Testing1": "present"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 3)

    def test_ec2_mark_zero(self):
        localtz = tz.gettz("America/New_York")
        dt = datetime.datetime.now(localtz)
        dt = dt.replace(year=2017, month=11, day=24, hour=7, minute=00)
        session_factory = self.replay_flight_data("test_ec2_mark_zero")
        session = session_factory(region="us-east-1")
        ec2 = session.client("ec2")
        resource = ec2.describe_instances(InstanceIds=["i-04d3e0630bd342566"])[
            "Reservations"
        ][
            0
        ][
            "Instances"
        ][
            0
        ]
        tags = [t["Value"] for t in resource["Tags"] if t["Key"] == "maid_status"]
        self.assertEqual(len(tags), 0)

        policy = self.load_policy(
            {
                "name": "ec2-mark-zero-days",
                "resource": "ec2",
                "filters": [{"tag:CreatorName": "joshuaroot"}],
                "actions": [{"type": "mark-for-op", "days": 0, "op": "terminate"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["InstanceId"], "i-04d3e0630bd342566")

        resource = ec2.describe_instances(InstanceIds=["i-04d3e0630bd342566"])[
            "Reservations"
        ][
            0
        ][
            "Instances"
        ][
            0
        ]
        tags = [t["Value"] for t in resource["Tags"] if t["Key"] == "maid_status"]
        result = datetime.datetime.strptime(
            tags[0].strip().split("@", 1)[-1], "%Y/%m/%d"
        ).replace(
            tzinfo=localtz
        )
        self.assertEqual(result.date(), dt.date())

    def test_ec2_mark_hours(self):
        localtz = tz.gettz("America/New_York")
        dt = datetime.datetime.now(localtz)
        dt = dt.replace(
            year=2018, month=2, day=20, hour=18, minute=00, second=0, microsecond=0
        )
        session_factory = self.replay_flight_data("test_ec2_mark_hours")
        session = session_factory(region="us-east-1")
        ec2 = session.client("ec2")

        policy = self.load_policy(
            {
                "name": "ec2-mark-5-hours",
                "resource": "ec2",
                "filters": [
                    {"tag:hourly-mark": "absent"}, {"tag:CreatorName": "joshuaroot"}
                ],
                "actions": [
                    {
                        "type": "mark-for-op",
                        "tag": "hourly-mark",
                        "hours": 3,
                        "op": "stop",
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

        resource = ec2.describe_instances(InstanceIds=[resources[0]["InstanceId"]])[
            "Reservations"
        ][
            0
        ][
            "Instances"
        ][
            0
        ]
        tags = [t["Value"] for t in resource["Tags"] if t["Key"] == "hourly-mark"]
        result = datetime.datetime.strptime(
            tags[0].strip().split("@", 1)[-1], "%Y/%m/%d %H%M %Z"
        ).replace(
            tzinfo=localtz
        )
        self.assertEqual(result, dt)

    def test_ec2_marked_hours(self):
        session_factory = self.replay_flight_data("test_ec2_marked_hours")
        policy = self.load_policy(
            {
                "name": "ec2-mark-5-hours",
                "resource": "ec2",
                "filters": [
                    {
                        "type": "marked-for-op",
                        "tag": "hourly-mark",
                        "op": "stop",
                        "skew_hours": 3,
                    }
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)
        self.assertEqual(resources[0]["InstanceId"], "i-098dae2615acb5809")


class TestStop(BaseTest):

    def test_ec2_stop(self):
        session_factory = self.replay_flight_data("test_ec2_stop")
        policy = self.load_policy(
            {
                "name": "ec2-test-stop",
                "resource": "ec2",
                "filters": [{"tag:Testing": "not-null"}],
                "actions": [{"type": "stop"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 1)

    def test_ec2_stop_hibernate(self):
        session_factory = self.replay_flight_data("test_ec2_stop_hibernate")
        policy = self.load_policy(
            {
                "name": "ec2-test-stop-hibernate",
                "resource": "ec2",
                "query": [{"tag-key": "Testing"}],
                "actions": [{"type": "stop", "hibernate": True}],
            },
            config={"region": "us-west-2"},
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 2)

        if self.recording:
            time.sleep(25)

        instances = utils.query_instances(
            session_factory(), InstanceIds=[r["InstanceId"] for r in resources]
        )

        stopped = [
            i
            for i in instances
            if i["StateReason"]["Code"] == "Client.UserInitiatedShutdown"
        ]
        hibernated = [
            i
            for i in instances
            if i["StateReason"]["Code"] == "Client.UserInitiatedHibernate"
        ]

        self.assertEqual(len(stopped), 1)
        self.assertEqual(len(hibernated), 1)


class TestReboot(BaseTest):

    def test_ec2_reboot(self):
        session_factory = self.replay_flight_data("test_ec2_reboot")
        policy = self.load_policy(
            {
                "name": "ec2-test-reboot",
                "resource": "ec2",
                "filters": [{"tag:Testing": "not-null"}],
                "actions": [{"type": "reboot"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 2)
        running = []
        for i in resources:
            if i["State"]["Name"] == "running":
                running.append(i["InstanceId"])
        if self.recording:
            time.sleep(25)
        instances = utils.query_instances(
            session_factory(), InstanceIds=[r["InstanceId"] for r in resources]
        )

        cur_running = []
        for i in instances:
            if i["State"]["Name"] == "running":
                cur_running.append(i["InstanceId"])

        cur_running.sort()
        running.sort()

        self.assertEqual(cur_running, running)


class TestStart(BaseTest):

    def test_invalid_state_extract(self):
        self.assertEqual(
            ec2.extract_instance_id(
                ("An error occurred (IncorrectInstanceState) when calling "
                 "the StartInstances operation: The instance 'i-abc123' is "
                 "not in a state from which it can be started.")),
            'i-abc123')
        self.assertRaises(
            ValueError,
            ec2.extract_instance_id,
            ("An error occurred (IncorrectInstanceState) when calling "
             "the StartInstances operation: The instance is "
             "not in a state from which it can be started."))

    def test_ec2_start_handle_invalid_state(self):
        policy = self.load_policy({
            "name": "ec2-test-start",
            "resource": "ec2",
            "filters": [],
            "actions": [{"type": "start"}],
        })

        client = mock.MagicMock()
        client.start_instances.side_effect = ClientError(
            {'Error': {
                'Code': 'IncorrectInstanceState',
                'Message': "The instance 'i-08270b9cfb568a1c4' is not in a state from which it can be started" # NOQA
            }}, 'StartInstances')

        start_action = policy.resource_manager.actions[0]
        self.assertEqual(
            start_action.process_instance_set(
                client, [{'InstanceId': 'i-08270b9cfb568a1c4'}], 'm5.xlarge', 'us-east-1a'),
            None)

        client2 = mock.MagicMock()
        client2.start_instances.side_effect = ValueError
        self.assertRaises(
            ValueError,
            start_action.process_instance_set,
            client2, [{'InstanceId': 'i-08270b9cfb568a1c4'}], 'm5.xlarge', 'us-east-1a')

    def test_ec2_start(self):
        session_factory = self.replay_flight_data("test_ec2_start")
        policy = self.load_policy(
            {
                "name": "ec2-test-start",
                "resource": "ec2",
                "filters": [],
                "actions": [{"type": "start"}],
            },
            session_factory=session_factory,
        )
        resources = policy.run()
        self.assertEqual(len(resources), 2)

    def test_ec2_start_fails(self):
        session_factory = self.replay_flight_data("test_ec2_start")
        policy = self.load_policy(
            {
                "name": "ec2-test-start",
                "resource": "ec2",
                "filters": [],
                "actions": [{"type": "start"}],
            },
            session_factory=session_factory,
        )
        output = self.capture_logging("custodian.actions", level=logging.DEBUG)
        with mock.patch.object(ec2.Start, "process_instance_set", return_value=True):
            try:
                policy.run()
            except RuntimeError:
                pass
            else:
                self.fail("should have raised error")

        log_output = output.getvalue()
        self.assertIn("Could not start 1 of 1 instances", log_output)
        self.assertIn("t2.micro us-west-2c", log_output)
        self.assertIn("i-08270b9cfb568a1c4", log_output)


class TestOr(BaseTest):

    def test_ec2_or_condition(self):
        session_factory = self.replay_flight_data("test_ec2_stop")
        policy = self.load_policy(
            {
                "name": "ec2-test-snapshot",
                "resource": "ec2",
                "filters": [
                    {"or": [{"tag:Name": "CompileLambda"}, {"tag:Name": "Spinnaker"}]}
                ],
            },
            session_factory=session_factory,
        )
        resources = policy.run()