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))...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!