Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use is_aws method in localstack

Best Python code snippet using localstack_python

remote_run.py

Source:remote_run.py

1import json2from typing import Sequence3from botocore.exceptions import ClientError, NoCredentialsError4from flask import jsonify, make_response, request5from monkey_island.cc.resources.AbstractResource import AbstractResource6from monkey_island.cc.resources.request_authentication import jwt_required7from monkey_island.cc.services import AWSService8from monkey_island.cc.services.aws import AWSCommandResults9CLIENT_ERROR_FORMAT = (10    "ClientError, error message: '{}'. Probably, the IAM role that has been associated with the "11    "instance doesn't permit SSM calls. "12)13NO_CREDS_ERROR_FORMAT = (14    "NoCredentialsError, error message: '{}'. Probably, no IAM role has been associated with the "15    "instance. "16)17class RemoteRun(AbstractResource):18    # API Spec: POST request is an action, it's not updating/creating any resource.19    # GET makes sense. The resource should be split up since these two use cases don't20    # really go together.21    urls = ["/api/remote-monkey"]22    def __init__(self, aws_service: AWSService):23        self._aws_service = aws_service24    @jwt_required25    def get(self):26        action = request.args.get("action")27        if action == "list_aws":28            is_aws = self._aws_service.island_is_running_on_aws()29            resp = {"is_aws": is_aws}30            if is_aws:31                try:32                    resp["instances"] = self._aws_service.get_managed_instances()33                except NoCredentialsError as e:34                    # API Spec: HTTP status code should be 40135                    resp["error"] = NO_CREDS_ERROR_FORMAT.format(e)36                    return jsonify(resp)37                except ClientError as e:38                    # API Spec: HTTP status code should not be 20039                    resp["error"] = CLIENT_ERROR_FORMAT.format(e)40                    return jsonify(resp)41            return jsonify(resp)42        return {}43    @jwt_required44    def post(self):45        body = json.loads(request.data)46        if body.get("type") == "aws":47            results = self.run_aws_monkeys(body)48            # API Spec: POST should return identifier or updated/newly created resource, not some49            # kind of data. That's a GET thing.50            return RemoteRun._encode_results(results)51        # default action52        # API Spec: Why is this 500? 500 should be returned in case an exception occurs on the53        # server. 40x makes more sense.54        return make_response({"error": "Invalid action"}, 500)55    def run_aws_monkeys(self, request_body) -> Sequence[AWSCommandResults]:56        instances = request_body.get("instances")57        island_ip = request_body.get("island_ip")58        return self._aws_service.run_agents_on_managed_instances(instances, island_ip)59    @staticmethod60    def _encode_results(results: Sequence[AWSCommandResults]):61        result = list(map(RemoteRun._aws_command_results_to_encodable_dict, results))62        response = {"result": result}63        return jsonify(response)64    @staticmethod65    def _aws_command_results_to_encodable_dict(aws_command_results: AWSCommandResults):66        res_dict = aws_command_results.__dict__67        res_dict["status"] = res_dict["status"].name.lower()...

aws_lambda.py

Source:aws_lambda.py

1import logging2import sys3from process_rss import process4def lambda_handler(event, context):5    dry_run = "DRYRUN" in event6    is_aws = "IS_AWS" in event7    log_level = logging.DEBUG if "VERBOSE" in event else logging.INFO8    files = event["FILES"]9    logging.basicConfig(level=log_level)10    h = logging.StreamHandler(sys.stdout)11    h.setFormatter(logging.Formatter('%(asctime)s %(message)s'))12    logging.getLogger().addHandler(h)13    log = logging.getLogger(__name__)14    log.debug("In lambda handler -> %s %s %s" % (dry_run, is_aws, files))...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.