KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use port_exposed method in localstack
Best Python code snippet using localstack_python
container_actions.py
Source:container_actions.py 
1import os2import subprocess3import config4import json5import time6def start_container(args):7 8 '''9 This function is used in order to start a specific container.10 11 @param container_id: The id of the container.12 @return Success/Failure.13 @return log_msg: A list containing the log messages that occurred during the process.14 ''' 15 16 t1 = time.time()17 log_msg = []18 flag_failure = False19 20 try:21 args = json.dumps(args)22 args = json.loads(args)23 container_id = args['container_id']24 except:25 flag_failure = True26 log_msg.append('Wrong arguments provided!')27 log_msg.append(args)28 t2 = time.time()29 timeElapsed = t2 - t130 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})31 32 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker start ' + container_id33 34 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)35 (out, err) = proc.communicate()36 out.decode("utf-8").split('\r\n')37 if(proc.returncode == 0):38 log_msg.append('Container started successfully')39 else:40 flag_failure = True41 log_msg.append('Container failed to start')42 log_msg.append(str(out))43 44 t2 = time.time()45 timeElapsed = t2 - t146 47 if(flag_failure):48 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})49 else:50 return json.dumps({"exitStatus":"Success", "message": log_msg, "timeElapsed": timeElapsed})51def stop_container(args):52 53 '''54 This function is used in order to stop a specific container.55 56 @param container_id: The id of the container.57 @return Success/Failure.58 @return log_msg: A list containing the log messages that occurred during the process.59 ''' 60 61 t1 = time.time()62 log_msg = []63 flag_failure = False64 65 try:66 args = json.dumps(args)67 args = json.loads(args)68 container_id = args['container_id']69 except:70 flag_failure = True71 log_msg.append('Wrong arguments provided!')72 log_msg.append(args)73 t2 = time.time()74 timeElapsed = t2 - t175 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})76 77 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker stop ' + container_id78 79 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)80 (out, err) = proc.communicate()81 out.decode("utf-8").split('\r\n')82 if(proc.returncode == 0):83 log_msg.append('Container stopped successfully')84 else:85 flag_failure = True86 log_msg.append('Container failed to stop')87 log_msg.append(str(out))88 89 t2 = time.time()90 timeElapsed = t2 - t191 92 if(flag_failure):93 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})94 else:95 return json.dumps({"exitStatus":"Success", "message": log_msg, "timeElapsed": timeElapsed})96def deploy_container(args):97 98 '''99 This function is used in order to deploy a container based on a given predefined image.100 101 @param image_name: The name of the predefined image.102 @param container_name: The name of the container.103 @param port_exposed: The port number to be exposed by the container.104 @return Success/Failure.105 @return log_msg: A list containing the log messages that occurred during the process.106 ''' 107 108 t1 = time.time()109 log_msg = []110 flag_failure = False111 112 try:113 args = json.dumps(args)114 args = json.loads(args)115 image_name = args['image_name']116 container_name = args['container_name']117 port_exposed = args['port_exposed']118 except:119 flag_failure = True120 log_msg.append('Wrong arguments provided!')121 log_msg.append(args)122 t2 = time.time()123 timeElapsed = t2 - t1124 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})125 126 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker run -d -t --name ' + container_name + ' --publish ' + port_exposed + ':' + port_exposed + ' ' + image_name 127 128 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)129 (out, err) = proc.communicate()130 out.decode("utf-8").split('\r\n')131 132 if(proc.returncode == 0):133 log_msg.append('Container deployment successful')134 else:135 flag_failure = True136 log_msg.append('Container deployment failed')137 log_msg.append(str(out))138 139 t2 = time.time()140 timeElapsed = t2 - t1141 142 if(flag_failure):143 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})144 else:145 return json.dumps({"exitStatus":"Success", "message": log_msg, "timeElapsed": timeElapsed, "container_id": out.replace('\n','')})146def delete_container(args):147 148 '''149 This function is used in order to delete a specific container.150 151 @param container_id: The id of the container.152 @return Success/Failure.153 @return log_msg: A list containing the log messages that occurred during the process.154 ''' 155 156 t1 = time.time()157 log_msg = []158 flag_failure = False159 160 try:161 args = json.dumps(args)162 args = json.loads(args)163 container_id = args['container_id']164 except:165 flag_failure = True166 log_msg.append('Wrong arguments provided!')167 log_msg.append(args)168 t2 = time.time()169 timeElapsed = t2 - t1170 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})171 172 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker rm ' + container_id 173 174 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)175 (out, err) = proc.communicate()176 out.decode("utf-8").split('\r\n')177 if(proc.returncode == 0):178 log_msg.append('Container removed successfully.')179 else:180 flag_failure = True181 log_msg.append('Container removal failure. Maybe the container is still active')182 log_msg.append(str(out))183 184 t2 = time.time()185 timeElapsed = t2 - t1186 187 if(flag_failure):188 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})189 else:190 return json.dumps({"exitStatus":"Success", "message": log_msg, "timeElapsed": timeElapsed})191def get_container_details(args):192 '''193 This function is used in order to get all the details a specific container.194 195 @param container_id: The id of the container.196 @return Success/Failure.197 @return When Success it returns a dictionary containing the information of the container,198 Null in case of Failure 199 ''' 200 201 t1 = time.time()202 203 try:204 args = json.dumps(args)205 args = json.loads(args)206 container_id = args['container_id']207 except:208 t2 = time.time()209 timeElapsed = t2 - t1210 return json.dumps({"exitStatus":"Failure", "message": "NULL", "timeElapsed": timeElapsed})211 212 213 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker inspect ' + container_id214 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)215 (out, err) = proc.communicate()216 out.decode("utf-8").split('\r\n')217 218 t2 = time.time()219 timeElapsed = t2 - t1220 221 if(proc.returncode == 0):222 try:223 container_info = json.loads(out)[0]224 return json.dumps({"exitStatus":"Success", "message": container_info, "timeElapsed": timeElapsed})225 except:226 return json.dumps({"exitStatus":"Failure", "message": "NULL", "timeElapsed": timeElapsed})227 else:228 return json.dumps({"exitStatus":"Failure", "message": "NULL", "timeElapsed": timeElapsed})229 230def check_if_active(args):231 232 '''233 This function is used in order to check whether a specific container is active or not.234 235 @param container_id: The id of the container.236 @return True/False/Failure.237 @return log_msg: A list containing the log messages that occurred during the process.238 ''' 239 240 t1 = time.time()241 242 try:243 arguments = json.dumps(args)244 arguments = json.loads(arguments)245 container_id = args['container_id']246 except:247 t2 = time.time()248 timeElapsed = t2 - t1249 return json.dumps({"exitStatus":"Failure", "message": "NULL", "timeElapsed": timeElapsed})250 251 info = json.loads(get_container_details(args))252 if (info['exitStatus'] == "Success"):253 return json.dumps({"exitStatus":"Success", "message": info['message']['State']['Running'], "timeElapsed": info['timeElapsed']})254 else:255 return json.dumps({"exitStatus":"Failure", "message": "NULL", "timeElapsed": info['timeElapsed']})256 257def perform_action_inside_container(args):258 '''259 This function is used in order to perform an action inside a specific container.260 261 @param container_id: The id of the container.262 @param function_details: A dictionary containing all the details of the function to be called.263 e.x.:{264 "function_name": "the name of the function",265 "params": {266 "var_1": "value_1",267 ...,268 "var_N": "value_N",269 }270 }271 @return Success/Failure.272 @return When Success it returns a dictionary containing the information of the container,273 Null in case of Failure 274 ''' 275 276 t1 = time.time()277 278 log_msg = []279 flag_failure = False280 281 try:282 arguments = json.dumps(args)283 arguments = json.loads(arguments)284 container_id = args['container_id']285 function_details = arguments['function_details']286 except:287 flag_failure = True288 log_msg.append('Wrong arguments provided!')289 log_msg.append(args)290 t2 = time.time()291 timeElapsed = t2 - t1292 return json.dumps({"exitStatus":"Failure", "message": log_msg, "timeElapsed": timeElapsed})293 294 function_call = '"' + 'import malib.ma as malib; malib.' + function_details["function_name"] + '('295 args = function_details["params"].keys()296 if (len(args) > 0):297 for i in range(0, len(args) - 1):298 function_call = function_call + "'" + function_details["params"][args[i]] + "', "299 300 function_call = function_call + "'" + function_details["params"][args[(len(args) - 1)]] + "')" + '"'301 else:302 function_call = function_call + ")" + '"'303 command = 'echo "' + config.sudoer_password + '" | sudo --prompt="" -S docker exec ' + container_id + ' python3 -c ' + function_call304 proc = subprocess.Popen([command], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell = True)305 (out, err) = proc.communicate()306 out.decode("utf-8").split('\r\n')307 308 t2 = time.time()309 timeElapsed = t2 - t1310 311 info = json.loads(out)312 if(proc.returncode == 0):313 return json.dumps({"exitStatus":info['exitStatus'], "message": info['message'], "timeElapsed": timeElapsed})314 else:...
main.py
Source:main.py
1# https://levelup.gitconnected.com/understanding-grpc-a-practical-application-in-go-and-python-f3003c9158ef2# imports for functionality3from src.undirected import Undirected4from src.dag import Dag5from src.grid import Grid6from src.graph import Graph7from src.tree import Tree8# for fun9import emoji10# imports for the server11import grpc12from src.generated.query_servicer_pb2 import ToyGraphDBResponse, Pong, Status, Success13from src.generated.query_servicer_pb2_grpc import (14 ToyGraphDBServicer,15 add_ToyGraphDBServicer_to_server,16)17from src.generated import query_servicer_pb2 as query__servicer__pb218import logging19from concurrent import futures20PORT_EXPOSED = 909021class ToyGraphDBServicerProxy(ToyGraphDBServicer):22 def __init__(self):23 pass24 def ping(self, request, context):25 return Pong(message="ALl hail GRPC")26 def read_graph(self, request, context):27 pass28 def create_graph(self, request, context):29 type_of_graph = request.type30 response = None31 print(request.type)32 try:33 # can be better34 graph = (35 Grid(request.grid.data)36 if type_of_graph == query__servicer__pb2.Type_of_graphs.GRID37 else None38 )39 graph = (40 Dag(request.grid.data)41 if type_of_graph == query__servicer__pb2.Type_of_graphs.DAG42 else None43 )44 graph = (45 Undirected(request.grid.data)46 if type_of_graph == query__servicer__pb2.Type_of_graphs.UNDIRECTED47 else None48 )49 response = ToyGraphDBResponse(50 status=None,51 error_message=None,52 success_message=Success(53 code_number=query__servicer__pb2.Success.Code.ALL_GOOD,54 message="yay",55 ),56 )57 except ValueError as e:58 response = ToyGraphDBResponse(59 status=None, error_message=None, success_message=None,60 )61 return response62 def call_functionality_in_graph(self, request, context):63 pass64def serve():65 server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))66 add_ToyGraphDBServicer_to_server(ToyGraphDBServicerProxy(), server)67 server.add_insecure_port(f"[::]:{PORT_EXPOSED}")68 print(emoji.emojize(f"All systems go :rocket:"))69 server.start()70 server.wait_for_termination()71if __name__ == "__main__":72 logging.basicConfig()73 print(emoji.emojize(f"Running on port {PORT_EXPOSED} :thumbs_up:"))...
create_grid.py
Source:create_grid.py
1import grpc2from src.generated import query_servicer_pb2, query_servicer_pb2_grpc3from src.generated.query_servicer_pb2 import (4 Empty,5 ToyGraphDBRequest,6 Type_of_graphs,7 grid,8 List_of_integer,9 List_of_list_of_integer,10)11PORT_EXPOSED = 909012SAMPLE_GRID = [[0, 1, 1], [0, 1, 1], [0, 1, 1]]13DATABASE_NAME = "create_grid"14TABLE_NAME = "simple_grid"15# open a gRPC channel16channel = grpc.insecure_channel(f"localhost:{PORT_EXPOSED}")17# create a stub (client)18stub = query_servicer_pb2_grpc.ToyGraphDBStub(channel)19# make the call20response = stub.ping(Empty())21print(response)22# https://www.freecodecamp.org/news/googles-protocol-buffers-in-python/23# convert sample grid to data24list_of_integer = List_of_integer()25first_item = list_of_integer.add()26first_item.data.append([1,2,3])27grid_data = List_of_list_of_integer(data=list_of_integer)28grid_request_object = grid(data=grid_data)29request = ToyGraphDBRequest(30 type=Type_of_graphs.GRID,31 grid=grid_request_object,32 database=DATABASE_NAME,33 table=TABLE_NAME,34)35response = stub.create_graph(request)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
