common.py

# Copyright The Cloud Custodian Authors.
# SPDX-License-Identifier: Apache-2.0
import gzip
import json
import logging
import os
import unittest
import uuid
from c7n.config import Bag

from c7n.testing import TestUtils, TextTestIO, functional # NOQA

from .zpill import PillTest, ACCOUNT_ID


logging.getLogger("placebo.pill").setLevel(logging.DEBUG)
logging.getLogger("botocore").setLevel(logging.WARNING)


C7N_VALIDATE = bool(os.environ.get("C7N_VALIDATE", ""))

skip_if_not_validating = unittest.skipIf(
    not C7N_VALIDATE, reason="We are not validating schemas."
)


# Set this so that if we run nose directly the tests will not fail
if "AWS_DEFAULT_REGION" not in os.environ:
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"


class BaseTest(TestUtils, PillTest):

    # custodian_schema = C7N_SCHEMA

    @property
    def account_id(self):
        return ACCOUNT_ID

    def _get_policy_config(self, **kw):
        if 'account_id' not in kw:
            kw['account_id'] = self.account_id
        if 'region' not in kw:
            kw['region'] = 'us-east-1'
        return super(BaseTest, self)._get_policy_config(**kw)


class ConfigTest(BaseTest):
    """Test base class for integration tests with aws config.

    To allow for integration testing with config.

     - before creating and modifying use the
       initialize_config_subscriber method to setup an sqs queue on
       the config recorder's sns topic. returns the sqs queue url.

     - after creating/modifying a resource, use the wait_for_config
       with the queue url and the resource id.
    """

    def wait_for_config(self, session, queue_url, resource_id=None):
        # lazy import to avoid circular
        from c7n.sqsexec import MessageIterator

        client = session.client("sqs")
        messages = MessageIterator(client, queue_url, timeout=20)
        results = []
        while True:
            for m in messages:
                msg = json.loads(m["Body"])
                change = json.loads(msg["Message"])
                messages.ack(m)
                if resource_id and change["configurationItem"]["resourceId"] != resource_id:
                    continue
                results.append(change["configurationItem"])
                break
            if results:
                break
        return results

    def initialize_config_subscriber(self, session):
        config = session.client("config")
        sqs = session.client("sqs")
        sns = session.client("sns")

        channels = config.describe_delivery_channels().get("DeliveryChannels", ())
        assert channels, "config not enabled"

        topic = channels[0]["snsTopicARN"]
        queue = "custodian-waiter-%s" % str(uuid.uuid4())
        queue_url = sqs.create_queue(QueueName=queue).get("QueueUrl")
        self.addCleanup(sqs.delete_queue, QueueUrl=queue_url)

        attrs = sqs.get_queue_attributes(
            QueueUrl=queue_url, AttributeNames=("Policy", "QueueArn")
        )
        queue_arn = attrs["Attributes"]["QueueArn"]
        policy = json.loads(
            attrs["Attributes"].get(
                "Policy",
                '{"Version":"2008-10-17","Id":"%s/SQSDefaultPolicy","Statement":[]}'
                % queue_arn,
            )
        )
        policy["Statement"].append(
            {
                "Sid": "ConfigTopicSubscribe",
                "Effect": "Allow",
                "Principal": "*",
                "Action": "sqs:SendMessage",
                "Resource": queue_arn,
                "Condition": {"ArnEquals": {"aws:SourceArn": topic}},
            }
        )
        sqs.set_queue_attributes(
            QueueUrl=queue_url, Attributes={"Policy": json.dumps(policy)}
        )
        subscription = sns.subscribe(
            TopicArn=topic, Protocol="sqs", Endpoint=queue_arn
        ).get(
            "SubscriptionArn"
        )
        self.addCleanup(sns.unsubscribe, SubscriptionArn=subscription)
        return queue_url


def placebo_dir(name):
    return os.path.join(os.path.dirname(__file__), "data", "placebo", name)


def data_path(*parts):
    return os.path.join(os.path.dirname(__file__), 'data', *parts)


def event_data(name, event_type="cwe"):
    with open(os.path.join(os.path.dirname(__file__), "data", event_type, name)) as fh:
        return json.load(fh)


def load_data(file_name, state=None, **kw):

    fopen = file_name.endswith('gz') and gzip.open or open
    data = json.loads(
        fopen(os.path.join(os.path.dirname(__file__), "data", file_name)).read()
    )
    if state:
        data.update(state)
    if kw:
        data.update(kw)
    return data


def instance(state=None, file="ec2-instance.json", **kw):
    return load_data(file, state, **kw)


class Instance(Bag):
    pass


class Reservation(Bag):
    pass


class Client:

    def __init__(self, instances):
        self.instances = instances
        self.filters = None

    def get_all_instances(self, filters=None):
        self.filters = filters
        return [Reservation({"instances": [i for i in self.instances]})]