Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.