Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.