KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_free_tcp_port method in localstack

Best Python code snippet using localstack_python

server_sockets_test.py

Source:server_sockets_test.py

...45		server.terminate()46	def test_default_socket(self):47		self._test_connect([], "allow", [], "hello", ":", EXIT_OK)48	def test_tcp_socket(self):49		port = get_free_tcp_port()50		self._test_connect(["--bind-tcp=0.0.0.0:%i" % port], "allow", [], "hello", "tcp/127.0.0.1:%i/" % port, EXIT_OK)51		self._test_connect(["--bind-tcp=0.0.0.0:%i" % port], "allow", [], "hello", "ws/127.0.0.1:%i/" % port, EXIT_OK)52	def test_ws_socket(self):53		port = get_free_tcp_port()54		self._test_connect(["--bind-ws=0.0.0.0:%i" % port], "allow", [], "hello", "ws/127.0.0.1:%i/" % port, EXIT_OK)55	def test_ssl(self):56		server = None57		display_no = self.find_free_display_no()58		display = ":%s" % display_no59		tcp_port = get_free_tcp_port()60		ws_port = get_free_tcp_port()61		wss_port = get_free_tcp_port()62		ssl_port = get_free_tcp_port()63		try:64			tmpdir = tempfile.mkdtemp(suffix='ssl-xpra')65			certfile = os.path.join(tmpdir, "self.pem")66			openssl_command = [67				"openssl", "req", "-new", "-newkey", "rsa:4096", "-days", "2", "-nodes", "-x509",68				"-subj", "/C=US/ST=Denial/L=Springfield/O=Dis/CN=localhost",69				"-keyout", certfile, "-out", certfile,70				]71			openssl = self.run_command(openssl_command)72			assert pollwait(openssl, 10)==0, "openssl certificate generation failed"73			cert_data = load_binary_file(certfile)74			log("generated cert data: %s", repr_ellipsized(cert_data))75			if not cert_data:76				#cannot run openssl? (happens from rpmbuild)...

server.py

Source:server.py

...18from user_srv.settings import settings19def on_exit(signo, frame):20    logger.info("è¿ç¨ä¸æ")21    sys.exit(0)22def get_free_tcp_port():23    tcp = socket.socket(socket.AF_INET, socket.SOCK_STREAM)24    tcp.bind(("", 0))25    _, port = tcp.getsockname()26    tcp.close()27    return port28def serve():29    parser = argparse.ArgumentParser()30    parser.add_argument('--ip',31                        nargs="?",32                        type=str,33                        default="192.168.245.1",34                        help="binding ip"35                        )36    parser.add_argument('--port',37                        nargs="?",38                        type=int,39                        default=0,40                        help="the listening port"41                        )42    args = parser.parse_args()43    if args.port == 0:44        port = get_free_tcp_port()45    else:46        port = args.port47    logger.add("logs/user_srv_{time}.log")48    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))49    # æ³¨åç¨æ·æå¡50    user_pb2_grpc.add_UserServicer_to_server(UserServicer(), server)51    # æ³¨åå¥åº·æ£æ¥52    health_servicer = health.HealthServicer()53    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)54    server.add_insecure_port(f'{args.ip}:{port}')55    """56        windowsä¸æ¯æçä¿¡å·æ¯æéçï¼57            SIGINT ctrl+c ä¸æ58            SIGTERM killååºçè½¯ä»¶ç»æ¢59    """60    # ä¸»è¿ç¨éåºä¿¡å·çå¬61    signal.signal(signal.SIGINT, on_exit)62    signal.signal(signal.SIGTERM, on_exit)63    logger.info(f"å¯å¨æå¡ï¼{args.ip}:{port}")64    server.start()65    logger.info(f"æå¡æ³¨åå¼å§")66    register = consul.ConsulRegister(settings.CONSUL_HOST, settings.CONSUL_PORT)67    # æ²¡ææ³¨åæå68    if not register.register(name=settings.SERVICE_NAME, id=settings.SERVICE_NAME,69                             address=args.ip, port=port, tags=settings.SERVICE_TAGS, check=None):70        logger.info(f"æå¡æ³¨åå¤±è´¥")71        sys.exit(0)72    logger.info(f"æå¡æ³¨åæå")73    server.wait_for_termination()74# @logger.catch75# def my_function(x, y, z):76#     # An error? It's caught anyway!77#     return 1 / (x + y + z)78if __name__ == "__main__":79    # my_function(0,0,0)80    # logger.debug("è°è¯ä¿¡æ¯")81    # logger.info("æ®éä¿¡æ¯")82    # logger.warning("è¦åä¿¡æ¯")83    # logger.error("éè¯¯ä¿¡æ¯")84    # logger.critical("ä¸¥ééè¯¯ä¿¡æ¯")85    # print(get_free_tcp_port())86    logging.basicConfig()...

base_unit_test.py

Source:base_unit_test.py

...3    def setUp(self):4        # generate random docker container name5        self.container_name = self.generate_random_string(15)6        # get a free TCP port for the container7        self.port = self.get_free_tcp_port()8    def tearDown(self):9        from os import system10        system("docker kill {}".format(self.container_name))11    @staticmethod12    def generate_random_string(k: int):13        from random import choice14        from string import ascii_lowercase, digits15        return ''.join(choice(ascii_lowercase + digits) for _ in range(k))16    @staticmethod17    def get_free_tcp_port():18        from socket import socket, AF_INET, SOCK_STREAM19        tcp = socket(AF_INET, SOCK_STREAM)20        tcp.bind(('', 0))21        addr, port = tcp.getsockname()22        tcp.close()23        return port24if __name__ == '__main__':...

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.