KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use is_port_available method in avocado
Best Python code snippet using avocado_python
experiment_server_controller.py
Source:experiment_server_controller.py 
...11#!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!12# Supporting functions13# general purpose helper functions used by the dashboard server and controller14#!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!15def is_port_available(ip, port):16 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)17 s.settimeout(1)18 try:19 s.connect((ip, int(port)))20 s.shutdown(2)21 return False22 except socket.timeout:23 print "*** Failed to test port availability. Check that host\nis set properly in config.txt"24 return True25 except socket.error,e:26 return True27def wait_until_online(function, ip, port):28 """29 Uses Wait_For_State to wait for the server to come online, then runs the given function.30 """31 awaiting_service = Wait_For_State(lambda: not is_port_available(ip, port), function)32 awaiting_service.start()33 return awaiting_service34def launch_browser(host, port, route):35 launchurl = "http://{host}:{port}/{route}".format(host=host, port=port, route=route)36 webbrowser.open(launchurl, new=1, autoraise=True)37def launch_browser_when_online(ip, port, route):38 return wait_until_online(lambda: launch_browser(ip, port, route), ip, port)39#----------------------------------------------------------------40# handles waiting for processes which we don't control (e.g.,41# browser requests)42#----------------------------------------------------------------43class Wait_For_State(Thread):44 """45 Waits for a state-checking function to return True, then runs a given46 function. For example, this is used to launch the browser once the server is47 started up.48 Example:49 t = Wait_For_State(lambda: server.check_port_state(), lambda: print "Server has started!")50 t.start()51 t.cancel() # Cancels thread52 """53 def __init__(self, state_function, function, pollinterval=1):54 Thread.__init__(self)55 self.function = function56 self.state_function = state_function57 self.pollinterval = pollinterval58 self.finished = Event()59 self.final = lambda: ()60 def cancel(self):61 self.finished.set()62 def run(self):63 while not self.finished.is_set():64 if self.state_function():65 self.function()66 self.finished.set()67 else:68 self.finished.wait(self.pollinterval)69#----------------------------------------------70# vanilla exception handler71#----------------------------------------------72class ExperimentServerControllerException(Exception):73 def __init__(self, value):74 self.value = value75 def __str__(self):76 return repr(self.value)77#----------------------------------------------78# simple wrapper class to control the79# starting/stopping of experiment server80#----------------------------------------------81class ExperimentServerController:82 def __init__(self, config):83 self.config = config84 self.server_running = False85 def get_ppid(self):86 if not self.is_port_available():87 url = "http://{hostname}:{port}/ppid".format(hostname=self.config.get("Server Parameters", "host"), port=self.config.getint("Server Parameters", "port"))88 ppid_request = urllib2.Request(url)89 ppid = urllib2.urlopen(ppid_request).read()90 return ppid91 else:92 raise ExperimentServerControllerException("Cannot shut down experiment server, server not online")93 def restart(self):94 self.shutdown()95 self.startup()96 def shutdown(self, ppid=None):97 if not ppid:98 ppid = self.get_ppid()99 print("Shutting down experiment server at pid %s..." % ppid)100 try:101 os.kill(int(ppid), signal.SIGKILL)102 self.server_running = False103 except ExperimentServerControllerException:104 print ExperimentServerControllerException105 else:106 self.server_running = False107 def kill_child_processes(self, parent_pid, sig=signal.SIGTERM):108 if os.uname()[0] is 'Linux':109 ps_command = subprocess.Popen('pstree -p %d | perl -ne \'print "$1 "\110 while /\((\d+)\)/g\'' % parent.pid,111 shell=True, stdout=subprocess.PIPE)112 ps_output = ps_command.stdout.read()113 retcode = ps_command.wait()114 assert retcode == 0, "ps command returned %d" % retcode115 for pid_str in ps_output.split("\n")[:-1]:116 os.kill(int(pid_str), sig)117 if os.uname()[0] is 'Darwin':118 child_pid = parent.get_children(recursive=True)119 for pid in child_pid:120 pid.send_signal(signal.SIGTERM)121 def is_server_running(self):122 PROCNAME = "psiturk_experiment_server"123 cmd = "ps -o pid,command | grep '"+ PROCNAME + "' | grep -v grep | awk '{print $1}'"124 psiturk_exp_processes = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)125 output = psiturk_exp_processes.stdout.readlines()126 parent = psutil.Process(psiturk_exp_processes.pid)127 self.kill_child_processes(parent.pid)128 129 if output:130 is_psiturk_using_port = True131 else:132 is_psiturk_using_port = False133 is_port_open = self.is_port_available()134 #print self.server_running, " ", portopen135 if is_port_open and is_psiturk_using_port: # This should never occur136 return 'maybe'137 elif not is_port_open and not is_psiturk_using_port:138 return 'blocked'139 elif is_port_open and not is_psiturk_using_port:140 return 'no'141 elif not is_port_open and is_psiturk_using_port:142 return 'yes'143 def is_port_available(self):144 return is_port_available(self.config.get("Server Parameters", "host"), self.config.getint("Server Parameters", "port"))145 def startup(self):146 server_command = "{python_exec} '{server_script}'".format(147 python_exec = sys.executable,148 server_script = os.path.join(os.path.dirname(__file__), "experiment_server.py")149 )150 server_status = self.is_server_running()151 if server_status == 'no':152 #print "Running experiment server with command:", server_command153 subprocess.Popen(server_command, shell=True, close_fds=True)154 print "Experiment server launching..."155 self.server_running = True156 elif server_status == 'maybe':157 print "Error: Not sure what to tell you..."158 elif server_status == 'yes':...
bash_utils.py
Source:bash_utils.py
...33 for f in files:34 copy_file(f, destination)35def find_free_port(base_port=7000):36 port = base_port37 available = is_port_available(port)38 while not available:39 port += 140 available = is_port_available(port)41 return port42def is_port_available(port=7000):43 import socket, errno44 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)45 try:46 s.bind(("0.0.0.0", port))47 except socket.error as e:48 if e.errno == errno.EADDRINUSE:49 return False50 else:51 traceback.print_exc()52 raise Exception(e)53 s.close()54 return True55def get_ip_address():56 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)...
ports.py
Source:ports.py
1# -*- coding: utf-8 -*-2#3# Copyright (C) 2016-2020 SysMedOs_team @ AG Bioanalytik, University of Leipzig:4#5# LipidLynxX is using GPL V3 License6#7# Please cite our publication in an appropriate form.8# LipidLynxX preprint on bioRxiv.org9# Zhixu Ni, Maria Fedorova.10# "LipidLynxX: a data transfer hub to support integration of large scale lipidomics datasets"11# DOI: 10.1101/2020.04.09.03389412#13# For more info please contact:14# Developer Zhixu Ni zhixu.ni@uni-leipzig.de15import socket16def try_port(port: int) -> bool:17 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)18 result = False19 try:20 sock.bind(("0.0.0.0", port))21 result = True22 except Exception as e:23 print(f"{e}")24 print(f"Port: [{port}] already in use, try to find next available port.")25 sock.close()26 return result27def check_port(port: int, task_name: str = None) -> int:28 is_port_available = False29 tmp_port = port30 if task_name:31 task_info = f" for {task_name}"32 else:33 task_info = ""34 while is_port_available is False:35 is_port_available = try_port(tmp_port)36 if is_port_available:37 if tmp_port == port:38 print(f"Port: [{port}] is available{task_info}.")39 else:40 print(41 f"Port: [{port}] already in use, switch to next available port: [{tmp_port}]{task_info}."42 )43 else:44 tmp_port += 345 return tmp_port46if __name__ == "__main__":47 usr_port = 139948 available_port = check_port(usr_port, "LipidLynxX")...
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 avocado 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?
