Best Python code snippet using fMBT_python
python3share-client
Source:python3share-client  
1#!/usr/bin/env python32# fMBT, free Model Based Testing tool3# Copyright (c) 2013-2019, Intel Corporation.4#5# Author: antti.kervinen@intel.com6#7# This program is free software; you can redistribute it and/or modify it8# under the terms and conditions of the GNU Lesser General Public License,9# version 2.1, as published by the Free Software Foundation.10#11# This program is distributed in the hope it will be useful, but WITHOUT12# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or13# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for14# more details.15#16# You should have received a copy of the GNU Lesser General Public License along with17# this program; if not, write to the Free Software Foundation, Inc.,18# 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.19# This executable implements a commandline interface for executing20# Python code on pythonshare servers.21"""Pythonshare client - connect to shared distributed namespaces22Usage: pythonshare-client [options]23Options:24  -h, --help25          Print this help.26  -C, --connect=hostspec27          Connect to pythonshare-server at hostspec. Options28          when connected will be executed on this connection.29  -n, --namespace=ns30          Set namespace to be used in for code/eval/interact/drop.31  -A, --async32          Following python-code executions and evaluations33          will be asynchronous.34  -S, --sync35          Following python-code executions and evaluations36          will be synchronous. This is the default.37  -o, --output=filename38          Output from following successful python-code evaluations39          will be written to filename. The default is standard output.40  -f, --format=format41          Evaluation return values will be converted to string with42          given formatter. Valid options are str, repr, pickle, print43          and raw.44  --password=password45          Authenticate to pythonshare server with password. The password46          must be plain text.47  --password-file=filename48          Authenticate to pythonshare server with password from49          filename. The password must be in filename in plain text.50Options when connected:51  -c, --code=python-code52          Execute python-code in ns.53  -e, --eval=python-expression54          Evaluate python-expression in ns. Print return55          value to standard output.56  -i, --interact57          Open interactive console for running code in ns.58  --ls-local59          List local namespaces.60  --ls-remote61          List remote namespaces.62  --ls-remote-ip63          List remote namespaces with namespace registerer's IP64          address and port.65  --ns-type=ns66          Returns type of a namespace ("local", "remote" or "" if it67          does not exist).68  --drop69          Drop namespace ns from the server.70  --unlock71          Unlock locked ns on the server.72  -D, --disconnect73          Disconnect from the pythonshare-server.74  -k, --kill75          Shutdown pythonshare-server.76  -P, --poll77          Poll which asynchronously return values are available for78          reading.79Example:80  pythonshare-client -C socket://localhost:8089 -n myns -c 'a=41' -e 'a+1'81"""82import python3share83import code84import pickle85import getopt86import parser87import sys88import traceback89opt_output_file = sys.stdout90def error(msg, exit_status=1):91    sys.stderr.write("pythonshare-client: %s\n" % (msg,))92    sys.exit(1)93def output(data):94    opt_output_file.write(data)95    opt_output_file.flush()96formatter_str    = lambda o: str(o)97formatter_repr   = lambda o: repr(o)98formatter_raw    = lambda o: o99formatter_pickle = lambda o: pickle.dumps(o)100formatter_print  = lambda o: str(o) + "\n"101def _pythonCode(code):102    try:103        parser.expr(code)104        return "expression"105    except SyntaxError:106        try:107            parser.suite(code.encode("utf8"))108            return "suite"109        except SyntaxError:110            return None111class PythonshareConsole(code.InteractiveConsole):112    def __init__(self, *args, **kwargs):113        try:114            import readline115        except:116            pass117        self._conn = kwargs.pop("conn")118        self._ns = kwargs.pop("namespace", None)119        if self._ns == None:120            self._ns = conn.namespace()121        code.InteractiveConsole.__init__(self, *args, **kwargs)122        self.__class__.__name__ = "pythonshare namespace %s at %s" % (123            self._ns, getattr(self._conn, "hostspec", "N/A"))124        self._codelines = ""125    def runsource(self, source, *args):126        self._codelines = source127        try:128            need_more = code.InteractiveConsole.runsource(self, source, *args)129        except:130            self._codelines = ""131            raise132        return need_more133    def runcode(self, code):134        source = self._codelines135        self._codelines = ""136        code_type = _pythonCode(source)137        try:138            if code_type == "expression":139                print(self._conn.eval_in(self._ns, source))140            elif code_type == "suite":141                self._conn.exec_in(self._ns, source)142            else:143                return144        except python3share.PythonShareError as e:145            tb_lines = traceback.format_exc().splitlines()146            try:147                i = tb_lines.index("RemoteEvalError: Traceback (most recent call last):")148                print(tb_lines[i])149            except ValueError:150                try:151                    i = tb_lines.index("RemoteExecError: Traceback (most recent call last):")152                    print(tb_lines[i])153                except ValueError:154                    print("i=-2")155                    i = -2156            print("\n".join(tb_lines[i+3:]))157    def showsyntaxerror(self, *args):158        self._codelines = ""159        return code.InteractiveConsole.showsyntaxerror(self, *args)160if __name__ == "__main__":161    try:162        opts, remainder = getopt.gnu_getopt(163            sys.argv[1:], "C:DAhSs:Pp:n:c:e:r:o:f:ik",164            ["connect=", "disconnect",165             "async", "sync", "poll",166             "help", "server=", "port=",167             "password=", "password-file=",168             "namespace=", "code=", "eval=", "read=",169             "output=", "format=", "interact",170             "ls-local", "ls-remote", "ls-remote-ip", "ns-type=",171             "drop", "unlock", "kill"])172    except getopt.GetoptError as e:173        error(str(e))174    opt_server = "localhost"175    opt_port = python3share.default_port176    opt_namespace = None177    opt_async = False178    opt_formatter = formatter_str179    opt_password = None180    conn = None181    if len(remainder) > 0:182        error('unknown parameter(s): "%s"' %183              '", "'.join(remainder))184    for opt, arg in opts:185        if opt in ["-h", "--help"]:186            print(__doc__)187            sys.exit(0)188        elif opt in ["-s", "--server"]:189            opt_server = arg190        elif opt in ["-p", "--port"]:191            try:192                opt_port = int(arg)193            except ValueError:194                error('invalid port "%s", integer expected.' % (arg,))195        elif opt in ["--password"]:196            opt_password = arg197        elif opt in ["--password-file"]:198            try:199                opt_password = file(arg).read().strip()200            except IOError as err:201                error('error reading password file "%s": %s' % (arg, err))202        elif opt in ["-C", "--connect"]:203            hostspec = arg204            try:205                conn = python3share.connection(hostspec, password=opt_password)206            except python3share.socket.error as e:207                error('cannot connect to "%s": %s' % (hostspec, e))208        elif opt in ["-D", "--disconnect"]:209            conn.close()210            conn = None211        elif opt in ["-n", "--namespace"]:212            opt_namespace = arg213            if conn:214                conn.set_namespace(opt_namespace)215        elif opt in ["-c", "--code"]:216            if conn == None:217                conn = python3share.connection("socket://%s:%s" %218                                              (opt_server, opt_port),219                                              password=opt_password)220            try:221                if opt_namespace:222                    conn.set_namespace(opt_namespace)223                conn.exec_(arg, async_=opt_async)224            except python3share.PythonShareError as e:225                print(type(e))226                print(e)227        elif opt in ["-e", "--eval"]:228            if conn == None:229                conn = python3share.connection("socket://%s:%s" %230                                              (opt_server, opt_port),231                                              password=opt_password)232            try:233                if opt_namespace:234                    conn.set_namespace(opt_namespace)235                output(opt_formatter(236                    conn.eval_(arg, async_=opt_async)))237            except python3share.PythonShareError as e:238                print(type(e))239                print(e)240        elif opt in ["-i", "--interact"]:241            if conn == None:242                conn = python3share.connection("socket://%s:%s" %243                                              (opt_server, opt_port),244                                              password=opt_password)245            console = PythonshareConsole(conn=conn, namespace=opt_namespace)246            console.interact()247        elif opt in ["--ls-local"]:248            if conn == None:249                error("cannot list namespaces - not connected")250            local_nss = conn.ls_local()251            print("\n".join(local_nss))252        elif opt in ["--ls-remote"]:253            if conn == None:254                error("cannot list namespaces - not connected")255            remote_nss = conn.ls_remote()256            print("\n".join(remote_nss))257        elif opt in ["--ls-remote-ip"]:258            if conn == None:259                error("cannot list namespaces - not connected")260            remote_nss = conn.ls_remote(ip=True)261            for name in sorted(remote_nss.keys()):262                print(name, remote_nss[name][0], remote_nss[name][1])263        elif opt in ["--ns-type"]:264            if conn == None:265                error("cannot query namespace - not connected")266            ns_type = conn.ns_type(arg)267            if ns_type != None:268                print(ns_type)269        elif opt in ["--drop"]:270            if conn == None:271                error("cannot drop namespace - not connected")272            if not opt_namespace:273                opt_namespace = conn.namespace()274            try:275                conn.drop_ns(opt_namespace)276            except Exception as e:277                error("cannot drop namespace: %s" % (278                    str(e).strip().splitlines()[-1].split(":")[-1].strip(),))279        elif opt in ["--unlock"]:280            if conn == None:281                error("cannot unlock namespace - not connected")282            if not opt_namespace:283                opt_namespace = conn.namespace()284            try:285                conn.unlock_ns(opt_namespace)286            except Exception as e:287                error("cannot unlock namespace: %s" % (288                    str(e).strip().splitlines()[-1].split(":")[-1].strip(),))289        elif opt in ["-k", "--kill"]:290            if conn == None:291                error("cannot kill server - not connected")292            if not opt_namespace:293                opt_namespace = conn.namespace()294            try:295                conn.kill_server(opt_namespace)296            except Exception as e:297                error("error on kill: %s" % (e,))298        elif opt in ["-r", "--read"]:299            try:300                try:301                    if not opt_namespace:302                        opt_namespace = conn.namespace()303                    index = int(arg)304                    # arg is an index to the list of poll_rvs305                    l = conn.poll_rvs(opt_namespace)306                    print(conn.read_rv(l[index]))307                except ValueError: # arg is a Async_rv string308                    print(conn.read_rv(arg))309            except python3share.PythonShareError as e:310                print(type(e))311                print(e)312        elif opt in ["-A", "--async"]:313            opt_async = True314        elif opt in ["-S", "--sync"]:315            opt_async = False316        elif opt in ["-P", "--poll"]:317            if not opt_namespace:318                opt_namespace = conn.namespace()319            async_rvs = conn.poll_rvs(opt_namespace)320            print("\n".join([str(arv) for arv in async_rvs]))321        elif opt in ["-o", "--output"]:322            opt_output_file = file(arg, "w")323        elif opt in ["-f", "--format"]:324            try:325                opt_formatter = globals()["formatter_%s" % (arg,)]326            except KeyError:...pythonshare-client
Source:pythonshare-client  
1#!/usr/bin/env python22# fMBT, free Model Based Testing tool3# Copyright (c) 2013, Intel Corporation.4#5# Author: antti.kervinen@intel.com6#7# This program is free software; you can redistribute it and/or modify it8# under the terms and conditions of the GNU Lesser General Public License,9# version 2.1, as published by the Free Software Foundation.10#11# This program is distributed in the hope it will be useful, but WITHOUT12# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or13# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for14# more details.15#16# You should have received a copy of the GNU Lesser General Public License along with17# this program; if not, write to the Free Software Foundation, Inc.,18# 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.19# This executable implements a commandline interface for executing20# Python code on pythonshare servers.21"""Pythonshare client - connect to shared distributed namespaces22Usage: pythonshare-client [options]23Options:24  -h, --help25          Print this help.26  -C, --connect=hostspec27          Connect to pythonshare-server at hostspec. Options28          when connected will be executed on this connection.29  -n, --namespace=ns30          Set namespace to be used in for code/eval/interact/drop.31  -A, --async32          Following python-code executions and evaluations33          will be asynchronous.34  -S, --sync35          Following python-code executions and evaluations36          will be synchronous. This is the default.37  -o, --output=filename38          Output from following successful python-code evaluations39          will be written to filename. The default is standard output.40  -f, --format=format41          Evaluation return values will be converted to string with42          given formatter. Valid options are str, repr, pickle, print43          and raw.44  --password=password45          Authenticate to pythonshare server with password. The password46          must be plain text.47  --password-file=filename48          Authenticate to pythonshare server with password from49          filename. The password must be in filename in plain text.50Options when connected:51  -c, --code=python-code52          Execute python-code in ns.53  -e, --eval=python-expression54          Evaluate python-expression in ns. Print return55          value to standard output.56  -i, --interact57          Open interactive console for running code in ns.58  --ls-local59          List local namespaces.60  --ls-remote61          List remote namespaces.62  --ns-type=ns63          Returns type of a namespace ("local", "remote" or "" if it64          does not exist).65  --drop66          Drop namespace ns from the server.67  --unlock68          Unlock locked ns on the server.69  -D, --disconnect70          Disconnect from the pythonshare-server.71  -k, --kill72          Shutdown pythonshare-server.73  -P, --poll74          Poll which asynchronously return values are available for75          reading.76Example:77  pythonshare-client -C socket://localhost:8089 -n myns -c 'a=41' -e 'a+1'78"""79import pythonshare80import code81import cPickle82import getopt83import parser84import sys85import traceback86opt_output_file = sys.stdout87def error(msg, exit_status=1):88    sys.stderr.write("pythonshare-client: %s\n" % (msg,))89    sys.exit(1)90def output(data):91    opt_output_file.write(data)92    opt_output_file.flush()93formatter_str    = lambda o: str(o)94formatter_repr   = lambda o: repr(o)95formatter_raw    = lambda o: o96formatter_pickle = lambda o: cPickle.dumps(o)97formatter_print  = lambda o: str(o) + "\n"98def _pythonCode(code):99    try:100        parser.expr(code)101        return "expression"102    except SyntaxError:103        try:104            parser.suite(code.encode("utf8"))105            return "suite"106        except SyntaxError:107            return None108class PythonshareConsole(code.InteractiveConsole):109    def __init__(self, *args, **kwargs):110        try:111            import readline112        except:113            pass114        self._conn = kwargs.pop("conn")115        self._ns = kwargs.pop("namespace", None)116        if self._ns == None:117            self._ns = conn.namespace()118        code.InteractiveConsole.__init__(self, *args, **kwargs)119        self.__class__.__name__ = "pythonshare namespace %s at %s" % (120            self._ns, getattr(self._conn, "hostspec", "N/A"))121        self._codelines = ""122    def runsource(self, source, *args):123        self._codelines = source124        try:125            need_more = code.InteractiveConsole.runsource(self, source, *args)126        except:127            self._codelines = ""128            raise129        return need_more130    def runcode(self, code):131        source = self._codelines132        self._codelines = ""133        code_type = _pythonCode(source)134        try:135            if code_type == "expression":136                print self._conn.eval_in(self._ns, source)137            elif code_type == "suite":138                self._conn.exec_in(self._ns, source)139            else:140                return141        except pythonshare.PythonShareError, e:142            tb_lines = traceback.format_exc().splitlines()143            try:144                i = tb_lines.index("RemoteEvalError: Traceback (most recent call last):")145                print tb_lines[i]146            except ValueError:147                try:148                    i = tb_lines.index("RemoteExecError: Traceback (most recent call last):")149                    print tb_lines[i]150                except ValueError:151                    print "i=-2"152                    i = -2153            print "\n".join(tb_lines[i+3:])154    def showsyntaxerror(self, *args):155        self._codelines = ""156        return code.InteractiveConsole.showsyntaxerror(self, *args)157if __name__ == "__main__":158    try:159        opts, remainder = getopt.gnu_getopt(160            sys.argv[1:], "C:DAhSs:Pp:n:c:e:r:o:f:ik",161            ["connect=", "disconnect",162             "async", "sync", "poll",163             "help", "server=", "port=",164             "password=", "password-file=",165             "namespace=", "code=", "eval=", "read=",166             "output=", "format=", "interact",167             "ls-local", "ls-remote", "ns-type=",168             "drop", "unlock", "kill"])169    except getopt.GetoptError, e:170        error(str(e))171    opt_server = "localhost"172    opt_port = pythonshare.default_port173    opt_namespace = None174    opt_async = False175    opt_formatter = formatter_str176    opt_password = None177    conn = None178    if len(remainder) > 0:179        error('unknown parameter(s): "%s"' %180              '", "'.join(remainder))181    for opt, arg in opts:182        if opt in ["-h", "--help"]:183            print __doc__184            sys.exit(0)185        elif opt in ["-s", "--server"]:186            opt_server = arg187        elif opt in ["-p", "--port"]:188            try:189                opt_port = int(arg)190            except ValueError:191                error('invalid port "%s", integer expected.' % (arg,))192        elif opt in ["--password"]:193            opt_password = arg194        elif opt in ["--password-file"]:195            try:196                opt_password = file(arg).read().strip()197            except IOError as err:198                error('error reading password file "%s": %s' % (arg, err))199        elif opt in ["-C", "--connect"]:200            hostspec = arg201            try:202                conn = pythonshare.connection(hostspec, password=opt_password)203            except pythonshare.socket.error, e:204                error('cannot connect to "%s": %s' % (hostspec, e))205        elif opt in ["-D", "--disconnect"]:206            conn.close()207            conn = None208        elif opt in ["-n", "--namespace"]:209            opt_namespace = arg210            if conn:211                conn.set_namespace(opt_namespace)212        elif opt in ["-c", "--code"]:213            if conn == None:214                conn = pythonshare.connection("socket://%s:%s" %215                                              (opt_server, opt_port),216                                              password=opt_password)217            try:218                if opt_namespace:219                    conn.set_namespace(opt_namespace)220                conn.exec_(arg, async=opt_async)221            except pythonshare.PythonShareError, e:222                print type(e)223                print e224        elif opt in ["-e", "--eval"]:225            if conn == None:226                conn = pythonshare.connection("socket://%s:%s" %227                                              (opt_server, opt_port),228                                              password=opt_password)229            try:230                if opt_namespace:231                    conn.set_namespace(opt_namespace)232                output(opt_formatter(233                    conn.eval_(arg, async=opt_async)))234            except pythonshare.PythonShareError, e:235                print type(e)236                print e237        elif opt in ["-i", "--interact"]:238            if conn == None:239                conn = pythonshare.connection("socket://%s:%s" %240                                              (opt_server, opt_port),241                                              password=opt_password)242            console = PythonshareConsole(conn=conn, namespace=opt_namespace)243            console.interact()244        elif opt in ["--ls-local"]:245            if conn == None:246                error("cannot list namespaces - not connected")247            local_nss = conn.ls_local()248            print "\n".join(local_nss)249        elif opt in ["--ls-remote"]:250            if conn == None:251                error("cannot list namespaces - not connected")252            remote_nss = conn.ls_remote()253            print "\n".join(remote_nss)254        elif opt in ["--ns-type"]:255            if conn == None:256                error("cannot query namespace - not connected")257            ns_type = conn.ns_type(arg)258            if ns_type != None:259                print ns_type260        elif opt in ["--drop"]:261            if conn == None:262                error("cannot drop namespace - not connected")263            if not opt_namespace:264                opt_namespace = conn.namespace()265            try:266                conn.drop_ns(opt_namespace)267            except Exception, e:268                error("cannot drop namespace: %s" % (269                    str(e).strip().splitlines()[-1].split(":")[-1].strip(),))270        elif opt in ["--unlock"]:271            if conn == None:272                error("cannot unlock namespace - not connected")273            if not opt_namespace:274                opt_namespace = conn.namespace()275            try:276                conn.unlock_ns(opt_namespace)277            except Exception, e:278                error("cannot unlock namespace: %s" % (279                    str(e).strip().splitlines()[-1].split(":")[-1].strip(),))280        elif opt in ["-k", "--kill"]:281            if conn == None:282                error("cannot kill server - not connected")283            if not opt_namespace:284                opt_namespace = conn.namespace()285            try:286                conn.kill_server(opt_namespace)287            except Exception, e:288                error("error on kill: %s" % (e,))289        elif opt in ["-r", "--read"]:290            try:291                try:292                    if not opt_namespace:293                        opt_namespace = conn.namespace()294                    index = int(arg)295                    # arg is an index to the list of poll_rvs296                    l = conn.poll_rvs(opt_namespace)297                    print conn.read_rv(l[index])298                except ValueError: # arg is a Async_rv string299                    print conn.read_rv(arg)300            except pythonshare.PythonShareError, e:301                print type(e)302                print e303        elif opt in ["-A", "--async"]:304            opt_async = True305        elif opt in ["-S", "--sync"]:306            opt_async = False307        elif opt in ["-P", "--poll"]:308            if not opt_namespace:309                opt_namespace = conn.namespace()310            async_rvs = conn.poll_rvs(opt_namespace)311            print "\n".join([str(arv) for arv in async_rvs])312        elif opt in ["-o", "--output"]:313            opt_output_file = file(arg, "w")314        elif opt in ["-f", "--format"]:315            try:316                opt_formatter = globals()["formatter_%s" % (arg,)]317            except KeyError:...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
