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

How to use cmd2py method in fMBT

Best Python code snippet using fMBT_python

pycosh.py

Source:pycosh.py

...94        out = out.decode(sys.stdout.encoding).encode("utf-8")95    if err != None and sys.stderr.encoding:96        err = err.decode(sys.stderr.encoding).encode("utf-8")97    return status, out, err98def cmd2py(cmdline):99    if "|" in cmdline:100        cmd_left, cmd_right = cmdline.split("|", 1)101        funccall = "pipe(%s, %s)" % (repr(cmd2py(cmd_left)),102                                     repr(cmd2py(cmd_right)))103    elif ">" in cmdline:104        cmd_left, cmd_right = cmdline.split(">", 1)105        filenames = shlex.split(cmd_right)106        if len(filenames) != 1:107            raise ValueError('right side of > must have single filename')108        funccall = "pipe(%s, %s)" % (109            repr(cmd2py(cmd_left)),110            repr("redir(%r)" % (filenames[0],)))111    else:112        cmdline_list = shlex.split(cmdline.strip())113        funcname = cmdline_list[0]114        args = cmdline_list[1:]115        funccall = funcname + repr(tuple(args))116    return funccall117if not recb is None:118    _parsed_call = []119    _parsed_calls = []120    _parsed_oper = []121    def _prog_args_parsed(*args):122        _parsed_calls.append("%s%s" % (123            _parsed_call[0], repr(tuple(_parsed_call[1:]))))124        _parsed_call[:] = []125    def _call_seq_parsed(*args):126        if _parsed_oper:127            for oper in _parsed_oper:128                if oper == ";":129                    _parsed_calls[0:2] = ["cat_eval(%s, %s)" % (repr(_parsed_calls[0]), repr(_parsed_calls[1]))]130                elif oper == "|":131                    _parsed_calls[0:2] = ["pipe(%s, %s)" % (repr(_parsed_calls[0]), repr(_parsed_calls[1]))]132        _parsed_oper[:] = []133    _PROG = recb.pattern(r"(?P<func>[a-zA-Z_0-9]+)\s*",134                         cb=lambda _1, m, _2: _parsed_call.append(m.groupdict()['func']))135    _ARG_QUOTED = recb.pattern(r"'(?P<arg>[^']*)'\s*")136    _ARG_DOUBLE_QUOTED = recb.pattern(r'"(?P<arg>[^"]*)"\s*')137    _ARG_SPACE_SEP = recb.pattern(r"(?P<arg>[^\s;|()]+)\s*")138    _CMD_CALL = (139        _PROG + recb.many(140            recb.any(_ARG_QUOTED, _ARG_DOUBLE_QUOTED, _ARG_SPACE_SEP).set_cb(141                lambda _1, m, _2: _parsed_call.append(m[1].groupdict()['arg'])))142    ).set_ca(_prog_args_parsed)143    _CMD_CONT = (recb.pattern(r"(?P<oper>[|;])\s*").set_cb(144        lambda _1, m, _2: _parsed_oper.append(m.groupdict()['oper']))145                 + _CMD_CALL)146    _CMD_PIPE_SEQ = (_CMD_CALL + recb.many(_CMD_CONT)).set_ca(_call_seq_parsed)147    _CMD_GROUP = (recb.pattern(r"\(\s*") +148                  _CMD_PIPE_SEQ +149                  recb.pattern(r"\)\s*") + recb.many(_CMD_CONT))150    _CMD = (_CMD_GROUP | _CMD_PIPE_SEQ).set_patterns({'GROUP': _CMD_GROUP})151def _test_cmd2py_newparser(cmdline):152    if recb is None:153        raise ImportError('recb required')154    _CMD.debug(interactive=False).parse("")155    _CMD.debug(interactive=False).parse("ls 'i'")156    _CMD.debug(interactive=False).parse('ls I "I"')157    _CMD.debug(interactive=False).parse('echo I "i I" | grep i')158    _CMD.debug(interactive=False).parse('echo I "i I" | grep I | grep i')159    _CMD.debug(interactive=False).parse("(echo test.txt; ls) | grep '1 2 3'")160def cmd2py_newparser(cmdline):161    _parsed_call[:] = []162    _parsed_calls[:] = []163    _parsed_oper[:] = []164    _, unparsed = _CMD.parse(cmdline)165    if unparsed:166        view_chars = 10 # how many chars around syntax error is shown in error message167        error_pos = len(cmdline) - len(unparsed) + 1168        str_at_pos = (repr(cmdline[max(error_pos-view_chars, 0):error_pos]) +169                      "<--[error pos]" +170                      repr(cmdline[error_pos:min(len(cmdline)-1, error_pos+view_chars)]))171        raise ValueError('syntax error at pos %s: %s' % (172            len(cmdline) - len(unparsed),173            str_at_pos))174    if len(_parsed_calls) == 1:175        return _parsed_calls[0]176    else:177        raise ValueError('parse error')178def prompt():179    """prompt180    print prompt"""181    try:182        user = getpass.getuser()183    except Exception:184        user = ""185    try:186        hostname = socket.gethostname()187    except Exception:188        hostname = ""189    return (user + "@" +190            hostname + ":" +191            os.getcwd().replace("\\", "/") + ": ")192def awk(prog, *args):193    """awk PROG [FILE...]194    PROG syntax: [/REGEXP/]{print $N...}"""195    filenames = expand(*args, accept_pipe=True).splitlines()196    if not filenames:197        raise ValueError("missing input")198    rv = []199    awk_syntax = re.compile('(/([^/]*)/)?\{([^}]*)\}')200    parsed_prog = awk_syntax.match(prog)201    if not parsed_prog:202        raise ValueError('syntax error in awk program')203    awk_pattern = parsed_prog.group(2)204    if not awk_pattern is None:205        awk_pattern_re = re.compile(awk_pattern)206    else:207        awk_pattern_re = re.compile("")208    awk_statements = [s.strip() for s in parsed_prog.group(3).split(";")]209    awk_fieldno_re = re.compile("\$([0-9]+)")210    awk_fieldsep_re = re.compile("[ \n\t\r]*")211    for filename in filenames:212        for line in open(filename).xreadlines():213            if awk_pattern_re.search(line):214                for stmt in awk_statements:215                    if stmt.startswith("print"):216                        what = stmt[5:].strip()217                        if not what:218                            # plain "print" results in full line219                            what = "$0"220                        else:221                            # no variable handling for now...222                            what = what.replace('"', '')223                        fields = [int(n) for n in awk_fieldno_re.findall(what)]224                        translate = {}225                        if fields:226                            line_fields = [line.splitlines()[0]] + [227                                l for l in awk_fieldsep_re.split(line) if l]228                            for field in fields:229                                if field < len(line_fields):230                                    translate["$" + str(field)] = line_fields[field]231                                else:232                                    translate["$" + str(field)] = ""233                            for rep in reversed(sorted(translate.keys())):234                                # if not reversed, might replace $1 before $10235                                what = what.replace(rep, translate[rep])236                        rv.append(what)237    return "\n".join(rv)238def cd(dirname):239    """cd DIRNAME240    change current working directory"""241    d = expand(dirname, accept_pipe=False, min=1, exist=True).splitlines()242    if len(d) > 1:243        raise ValueError("ambiguous directory name")244    os.chdir(os.path.join(os.getcwd(), d[0]))245    return ""246def curl(*args):247    """curl [-x P][-o FILE] URL248    download URL (use proxy P), save to FILE"""249    opts, urls = _getopts(args, "x:o:")250    rv = []251    if not urls:252        raise ValueError("missing URL(s)")253    if "-x" not in opts and "http_proxy" in _g_pyenv:254        opts["-x"] = _g_pyenv.get("http_proxy")255    if "-x" in opts:256        proxy = urllib2.ProxyHandler({257            'http': opts["-x"],258            'https': opts["-x"],259            'ftp': opts["-x"]})260    else:261        proxy = urllib2.ProxyHandler({})262    opener = urllib2.build_opener(proxy)263    urllib2.install_opener(opener)264    for url in urls:265        data = urllib2.urlopen(url).read()266        if "-o" in opts:267            _file(opts["-o"], "a").write(data)268        else:269            rv.append(data)270    return "".join(rv)271def find(*args):272    """find [-n NAME][-i][-t T][-p a] DIR...273    find under DIR(s), see help(find)"""274    # -n NAME: search entries matching wildcard pattern NAME275    # -i: ignore case276    # -t f: (type=file) match only files277    # -t d: (type=dir) match only directories278    # -p a: print absolute paths279    opts, remainder = _getopts(args, "n:t:ip:")280    dirnames = expand(*remainder, exist=True, accept_pipe=False).splitlines()281    if not dirnames:282        raise ValueError("missing DIR")283    if "-n" in opts:284        findname = opts["-n"]285    else:286        findname = "*"287    if "-p" in opts:288        if opts["-p"] == "a":289            print_absolute_names = True290        else:291            raise ValueError("invalid print option -p %r, supported: 'a'" %292                             (opts["-p"],))293    else:294        print_absolute_names = False295    if "-i" in opts:296        ignore_case = True297    else:298        ignore_case = False299    if "-t" in opts:300        findtype = opts["-t"].lower()301        if findtype not in ["f", "d"]:302            raise ValueError("find type must be 'f' (file) or 'd' (directory)")303    else:304        findtype = None305    rv = []306    for dirname in dirnames:307        dirname_ends_with_sep = dirname[-1] in ["/", "\\"]308        slash_only = not "\\" in dirname309        if slash_only:310            sep = "/"311        else:312            sep = os.path.sep313        # DIR + NAME forms a path without duplicate path separators:314        # if (and only if) DIR ends with /, then NAME does not start with /315        for root, dirs, files in os.walk(dirname):316            if slash_only:317                root = root.replace("\\", "/")318            if findtype:319                dirs_set = set(dirs)320                files_set = set(files)321            for name in dirs + files:322                if ((ignore_case == False and fnmatch.fnmatch(name, findname)) or323                     (ignore_case == True and fnmatch.fnmatch(name.lower(), findname.lower()))):324                    if (findtype == "f" and name not in files_set):325                        continue # skip not-a-file from find -t f ...326                    elif (findtype == "d" and name not in dirs_set):327                        continue # skip not-a-dir from find -t d ...328                    if print_absolute_names:329                        rv.append(os.path.abspath(root + sep + name).replace('\\','/'))330                    else:331                        if root == dirname:332                            if dirname_ends_with_sep:333                                rv.append(name)334                            else:335                                rv.append(sep + name)336                        else:337                            rv.append(root[len(dirname):] + sep + name)338    return "\n".join(rv)339def date():340    """date341    print current date and time"""342    t = datetime.datetime.now()343    return t.strftime("%Y-%m-%d %H:%M:%S.") + str(t.microsecond)344def diff(*args):345    """diff FILE1 FILE2346    print differences between two files"""347    files = expand(*args).splitlines()348    try:349        file1, file2 = files350    except Exception:351        raise ValueError("exactly two files required")352    lines1 = _file(file1).readlines()353    lines2 = _file(file2).readlines()354    udiff = difflib.unified_diff(lines1, lines2, file1, file2)355    # append endlines to lines where it is missing356    difflines = [l + ["", "\n"][l[-1] != "\n"] for l in udiff]357    return "".join(difflines)358def du(*args):359    """du [-h] FILE...360    print [human readable] disk usage of FILEs"""361    opts, filenames = _getopts(args, "h")362    if "-h" in opts:363        size_formatter = _human_readable_size364    else:365        size_formatter = lambda size: str(size)366    filenames = expand(*filenames, accept_pipe=False, min=1).splitlines()367    total_size = 0368    retval = []369    for direntry in filenames:370        size = None371        if os.path.isdir(direntry):372            for root, dirs, filelist in os.walk(direntry):373               for filename in filelist:374                   fullname = os.path.join(root, filename)375                   size = os.stat(fullname).st_size376                   retval.append("%-8s %s" % (size_formatter(size), fullname))377                   total_size += size378        elif os.path.isfile(direntry):379            size = os.stat(direntry).st_size380            total_size += size381            retval.append("%-8s %s" % (size_formatter(size), direntry))382    retval.append(size_formatter(total_size))383    return "\n".join(retval)384def echo(*args):385    return " ".join(args)386def env():387    """env388    print environment variables"""389    rv = []390    for key in sorted(_g_pyenv.keys()):391        rv.append("%s=%s" % (key, repr(_g_pyenv[key])))392    return "\n".join(rv)393def expand(*filenames, **kwargs):394    accept_pipe = kwargs.get("accept_pipe", True)395    min_count = kwargs.get("min", 0)396    must_exist = kwargs.get("exist", False)397    rv = []398    if not filenames:399        if accept_pipe and _g_pipe_has_data:400            rv.append(_g_pipe_filename)401    else:402        for pattern in filenames:403            extends_to = glob.glob(pattern)404            if extends_to:405                for filepath in extends_to:406                    if "/" in filepath and "\\" in filepath:407                        filepath = filepath.replace('\\', '/')408                    rv.append(filepath)409            elif not must_exist:410                rv.append(pattern)411    if not min_count <= len(rv):412        raise ValueError("expected at least %s file(s), got %s" %413                         (min_count, len(rv)))414    return "\n".join(rv)415def export(assignment):416    """export VAR=VALUE417    assign VALUE to environment variable VAR"""418    if not "=" in assignment or not assignment.split("=")[0].strip():419        raise ValueError("expected VAR=VALUE")420    _g_pyenv.__setitem__(*assignment.split("=", 1))421    return ""422def grep(*args):423    """grep [-iH] PATTERN [FILE...]424    show matching lines in file(s)"""425    opts, pattern_filenames = _getopts(args, "iH")426    ignore_case = "-i" in opts427    always_print_filename = "-H" in opts428    try:429        pattern = pattern_filenames[0]430        filenames = pattern_filenames[1:]431    except:432        raise ValueError("grep pattern missing")433    if ignore_case:434        pattern = pattern.lower()435    matching_lines = []436    all_files = expand(*filenames).splitlines()437    if len(all_files) > 1:438           always_print_filename = True439    prefix = ""440    for filename in all_files:441        if always_print_filename:442            prefix = filename.replace("\\", "/") + ": "443        if os.path.isdir(filename):444           matching_lines.append("grep: %s: is a directory\n" % (filename,))445           continue446        try:447            for line in file(filename).xreadlines():448                if ((not ignore_case and pattern in line) or449                    (ignore_case and pattern in line.lower())):450                    matching_lines.append(prefix + line)451        except IOError, e:452           matching_lines.append("grep: %s: %s\n" % (filename, e))453    return "".join(matching_lines)454def head(*args):455    """head [-n NUM] [FILE...]456    show first NUM lines in file(s)"""457    opts, filenames = _getopts(args, "n:")458    all_files = expand(*filenames).splitlines()459    if "-n" in opts:460        lines = int(opts["-n"])461    else:462        lines = 10463    rv = []464    for filename in all_files:465        line_count = 0466        for line in file(filename).xreadlines():467            line_count += 1468            if line_count > lines:469                break470            rv.append(line)471    return "".join(rv)472def help(func=None):473    """help [COMMAND]474    print help (on COMMAND)"""475    if not func:476        rv = []477        for c in globals().keys():478            if c.startswith("_"):479                continue480            if not isinstance(globals()[c], types.FunctionType):481                continue482            if not globals()[c].__doc__:483                continue484            if len(globals()[c].__doc__.splitlines()) != 2:485                continue486            rv.append("%-26s%s" %487                      tuple([l.strip() for l in488                             globals()[c].__doc__.splitlines()]))489        rv = sorted(rv)490    elif isinstance(globals().get(func, None), types.FunctionType):491        rv = inspect.getsource(globals().get(func)).splitlines()492    return "\n".join(rv)493def kill(*pids):494    """kill PID...495    terminate processes"""496    for pid in pids:497        os.kill(int(pid), signal.SIGTERM)498    return ""499def ls(*args):500    """ls [-ld]501    list files on current working directory"""502    opts, filenames = _getopts(args, "ld")503    files = []504    if filenames:505        for filename in expand(*filenames, exist=True).splitlines():506            if os.path.isdir(filename) and not "-d" in opts:507                root, subdirs, subfiles = os.walk(filename).next()508                root = root.replace('\\', '/')509                files.extend(sorted([root + "/" + d + "/" for d in subdirs]) +510                             sorted([root + "/" + f for f in subfiles]))511            else:512                files.append(filename)513    else:514        _, subdirs, files = os.walk(".").next()515        files = sorted([d + "/" for d in subdirs]) + sorted(files)516    files_outnames = []517    for f in files:518        if f.endswith("/"):519            outname = os.path.basename(f[:-1]) + "/"520        else:521            outname = os.path.basename(f)522        files_outnames.append((f, outname))523    if "-l" in opts:524        rv = []525        for f, o in files_outnames:526            fstat = os.stat(f)527            rv.append("%10s  %s  %s" % (528                fstat.st_size,529                time.strftime("%Y-%m-%d %H:%M", time.localtime(fstat.st_mtime)),530                o))531    else:532        rv = [o for f, o in files_outnames]533    return "\n".join(rv)534def nl(*filenames):535    """nl FILE...536    number lines"""537    all_files = expand(*filenames).splitlines()538    rv = []539    line_no = 0540    for filename in all_files:541        for line in file(filename).xreadlines():542            line_no += 1543            rv.append("%5s  %s" % (line_no, line))544    return "".join(rv)545def mkdir(*args):546    """mkdir [-p] DIRNAME...547    make directories, -p: intermediates if necessary"""548    args, dirnames = _getopts(args, "-p")549    for dirname in dirnames:550        if "-p" in args:551            os.makedirs(dirname)552        else:553            os.mkdir(dirname)554    return ""555def redir(dst_filename):556    # redirect data from input pipe to a file557    src_filename = expand(accept_pipe=True)558    if src_filename:559        file(dst_filename, "wb").write(560            file(src_filename, "rb").read())561    return ""562def rm(*args):563    """rm [-r] FILE...564    remove file"""565    args, filenames = _getopts(args, "rf")566    filenames = expand(*filenames, accept_pipe=False, min=1).splitlines()567    for filename in filenames:568        if "-r" in args and os.path.isdir(filename):569            shutil.rmtree(filename)570        else:571            os.remove(filename)572    return ""573def rmdir(dirname):574    """rmdir DIRNAME575    remove directory"""576    os.rmdir(dirname)577    return ""578def cat(*filenames):579    """cat FILE...580    concatenate contents of listed files"""581    return "".join([_file(f).read() for f in expand(*filenames).splitlines()])582def df(*args):583    """df [-h] DIRNAME584    print [human readable] free space on DIRNAME"""585    args, dirnames = _getopts(args, "-h")586    if "-h" in args:587        human_readable = True588    else:589        human_readable = False590    try:591        dirname = dirnames[0]592    except IndexError:593        raise Exception("directory name missing")594    if os.name == "nt": # Windows595        cfree = ctypes.c_ulonglong(0)596        ctypes.windll.kernel32.GetDiskFreeSpaceExW(597            ctypes.c_wchar_p(dirname), None, None,598            ctypes.byref(cfree))599        free = cfree.value600    else:  # Posix601        st = os.statvfs(dirname)602        free = st.f_bavail * st.f_frsize603    if human_readable:604        retval = _human_readable_size(free)605    else:606        retval = str(free)607    return retval608def md5sum(*filenames):609    """md5sum FILE...610    print MD5 (128-bit) checksums."""611    rv = []612    for filename in expand(*filenames).splitlines():613        rv.append("%-34s%s" %614                  (md5.md5(file(filename, "rb").read()).hexdigest(),615                   filename))616    return "\n".join(rv)617def mv(src, dst):618    """mv SOURCE DEST619    move file or directory to destination"""620    shutil.move(src, dst)621    return ""622def cp(src, dst):623    """cp SOURCE DEST624    copy file or directory to destination"""625    shutil.copy(src, dst)626    return ""627def pipe(expr_left, expr_right):628    global _g_pipe_has_data629    try:630        pipe_in_data = eval(expr_left)631        file(_g_pipe_filename, "wb").write(pipe_in_data)632        del pipe_in_data633        _g_pipe_has_data = True634        rv = eval(expr_right)635    finally:636        try:637            os.remove(_g_pipe_filename)638        except Exception:639            pass640        _g_pipe_has_data = False641    return rv642def cat_eval(expr_left, expr_right):643    left_data = eval(expr_left)644    right_data = eval(expr_right)645    return left_data + right_data646def osenv():647    """osenv648    print operating system environment variables"""649    rv = []650    for key in sorted(os.environ.keys()):651        rv.append("%s=%r" % (key, os.environ[key]))652    return "\n".join(rv)653def ps(*args):654    """ps [-v] [PID...]655    list processes (-v virtual memory)"""656    args, pids = _getopts(args, "-v")657    rv = []658    pids = set(pids)659    if os.name == "nt":660        if "-v" in args:661            opt_field = "PageFileUsage"662        else:663            opt_field = "parentprocessid"664        _, o, _ = _shell_soe(665            "wmic process get %s,processid,description,commandline" %666            (opt_field,))667        for line in o.splitlines():668            try:669                cmd_desc_almostlast, lastfield = line.rstrip().rsplit(" ", 1)670                cmd_desc, almostlast = (671                    cmd_desc_almostlast.rstrip().rsplit(" ", 1))672                cmd, desc = cmd_desc.rstrip().rsplit(" ", 1)673                if opt_field == "PageFileUsage":674                    pid = lastfield675                    pdata = almostlast # PageFileUsage676                else:677                    pid = lastfield678                    pdata = almostlast # parent pid679                if not desc.lower() in cmd.strip().lower():680                    cmd = "[%s] %s" % (desc.strip(), cmd.strip())681                if not pids or pid in pids:682                    rv.append("%8s %8s %s" %683                              (pid.strip(), pdata.strip(), cmd.strip()))684            except Exception:685                pass686    else:687        if "-v" in args:688            opt_field = "size"689        else:690            opt_field = "ppid"691        _, o, _ = _shell_soe("ps ax -o%s,pid,cmd" % (opt_field,))692        for line in o.splitlines():693            pdata, pid, cmd = line.strip().split(None, 2)694            if not pids or pid in pids:695                rv.append("%8s %8s %s" % (pid, pdata, cmd.strip()))696    return "\n".join(rv)697def psh(*cmd):698    """psh COMMAND699    run COMMAND in powershell (Windows)"""700    _, o, e = _shell_soe(701        ("powershell.exe",) + cmd)702    return o + e703_g_pspycosh_conn = None704def pspycosh(psconn, *cmdlines):705    """pspycosh CONNSPEC [CMD...]706    open remote pycosh shell or run CMDs on it"""707    global _g_pspycosh_conn708    if isinstance(psconn, pythonshare.client.Connection):709        _g_pspycosh_conn = psconn710        close_connection = False711    else:712        _g_pspycosh_conn = pythonshare.connect(psconn)713        close_connection = True714    _g_pspycosh_conn.exec_(_g_pycosh_source)715    if cmdlines:716        rv = []717        try:718            for cmdline in cmdlines:719                rv.append(pycosh_eval(cmdline))720        finally:721            if close_connection:722                _g_pspycosh_conn.close()723                _g_pspycosh_conn = None724        return "".join(rv)725    return ""726def _psput_file(conn, src_filename, dst_filename):727    data = file(src_filename, "rb").read()728    conn.eval_('file(%s, "wb").write(base64.b64decode(%s))' %729               (repr(dst_filename),730                repr(base64.b64encode(data))))731def _psput_dir(conn, dirname, dest_dir):732    rv = []733    dirname = dirname.replace("\\", "/")734    dir_dest_dir = dest_dir.replace("\\", "/") + "/" + os.path.basename(dirname)735    for root, dirs, files in os.walk(dirname):736        file_src_dir = root.replace('\\', '/')737        if file_src_dir[len(dirname):]:738            file_dest_dir = (dir_dest_dir + "/" + file_src_dir[len(dirname):])739        else:740            file_dest_dir = dir_dest_dir741        try:742            conn.eval_('os.makedirs(%r)' % (file_dest_dir,))743        except:744            pass745        for f in files:746            _psput_file(conn,747                        file_src_dir + "/" + f,748                        file_dest_dir + "/" + f)749            rv.append(file_src_dir + "/" + f)750    return rv751def psput(psconn, pattern):752    """psput CONNSPEC[//DEST] FILE...753    upload files to pythonshare server"""754    # Examples:755    # Put files to current working directory on host:756    #     psput passwd@host:port files757    # Put localdir under cwd/relative/path on host:758    #     psput passwd@host:port//relative/path localdir759    # Put localdir under /abs/path on host on Linux host:760    #     psput passwd@host:port///abs/path localdir761    # Put localdir under c:/abs/winpath on Windows host:762    #     psput passwd@host:port//c:/abs/winpath localdir763    # Put localdir to /abs/path on Linux host via hub/namespace:764    #     psput passwd@hub:port/namespace///abs/path localdir765    # Check cwd on host:766    #     pspycosh passwd@host:port pwd767    if isinstance(psconn, pythonshare.client.Connection):768        dest_dir = "."769        conn = psconn770        close_connection = False771    else:772        if "//" in psconn:773            psconn, dest_dir = psconn.split("//", 1)774        else:775            dest_dir = "."776        conn = pythonshare.connect(psconn)777        close_connection = True778    conn.exec_("import base64, os")779    rv = []780    for filename in expand(pattern, accept_pipe=False).splitlines():781        if os.path.isdir(filename):782            rv.extend(_psput_dir(conn, filename, dest_dir))783        else:784            _psput_file(conn, filename, dest_dir + "/" + os.path.basename(filename))785            rv.append(filename)786    if close_connection:787        conn.close()788    return "\n".join(rv)789def psget(psconn, pattern):790    """psget CONNSPEC FILE...791    download files from pythonshare server"""792    # Get *.txt from host to current working directory:793    #     psget passwd@host:port *.txt794    # Get * from host via hub to current working directory:795    #     psget passwd@hub/host *796    # Get * from HOSTDIR on host, via hub, to current working directory:797    #     psget passwd@hub/host//HOSTDIR *798    if isinstance(psconn, pythonshare.client.Connection):799        conn = psconn800        remotedir = ""801        close_connection = False802    elif "//" in psconn:803        hostspec, remotedir = psconn.split("//", 1)804        conn = pythonshare.connect(hostspec)805        close_connection = True806    else:807        remotedir = ""808        conn = pythonshare.connect(psconn)809        close_connection = True810    conn.exec_("".join(inspect.getsourcelines(expand)[0]))811    conn.exec_("import glob")812    if remotedir:813        remotedir = remotedir.replace("\\", "/")814        if not remotedir.endswith("/"):815            remotedir = remotedir + "/"816    rv = []817    for filename in conn.eval_('expand(%s, accept_pipe=False)' %818                               repr(remotedir + pattern)).splitlines():819        try:820            data = conn.eval_("file(%r, 'rb').read()" % (filename,))821        except:822            rv.append("! error reading %r" % (filename,))823            continue824        file(os.path.basename(filename), "wb").write(data)825        rv.append(filename)826    return "\n".join(rv)827def pwd():828    """pwd829    print current working directory"""830    return os.getcwd().replace("\\", "/")831def pye(*code):832    """pye CODE833    evaluate Python CODE"""834    code = " ".join(code)835    if _g_pipe_has_data:836        _g_pyenv["pipe_in"] = file(expand(accept_pipe=True), "rb")837    try:838        return str(eval(code, globals(), _g_pyenv))839    finally:840        if "pipe_in" in _g_pyenv:841            del _g_pyenv["pipe_in"]842def pyx(*code):843    """pyx CODE844    execute Python CODE"""845    code = " ".join(code)846    if _g_pipe_has_data:847        _g_pyenv["pipe_in"] = file(expand(accept_pipe=True), "rb")848    try:849        try:850            exec code in globals(), _g_pyenv851        except Exception, e:852            return str(e)853    finally:854        if "pipe_in" in _g_pyenv:855            del _g_pyenv["pipe_in"]856    return ""857def sed(cmd, *filenames):858    """sed s/P/R[/N] [FILE]859    replace P with R in FILE"""860    rv = []861    try:862        pattern, repl, count = re.findall("s/([^/]*)/([^/]*)/(.*)", cmd)[0]863        pattern = re.compile(pattern)864    except:865        raise ValueError('invalid command "%s"' % (cmd,))866    all_files = expand(*filenames).splitlines()867    for filename in all_files:868        for line in file(filename).readlines():869            try:870                count_arg = (int(count),)871            except:872                if count == "g":873                    count_arg = ()874                elif count == "":875                    count_arg = (1,)876                else:877                    raise ValueError('invalid count: "%s"' % (count,))878            rv.append(re.subn(* ((pattern, repl, line) + count_arg))[0])879    return "".join(rv)880def sh(*cmd):881    """sh COMMAND882    run COMMAND in shell"""883    s, o, e = _shell_soe(" ".join(cmd))884    return "[exit status: %s]\n%s" % (s, o+e)885def sleep(seconds):886    """sleep SECONDS887    sleep for SECONDS (float)"""888    time.sleep(float(seconds))889    return ""890def sort(*args):891    """sort [-n] [-k N] [FILE]892    sort lines [numerically] according to column N"""893    opts, filenames = _getopts(args, "k:n")894    filenames = expand(*filenames, accept_pipe=True).splitlines()895    rv = []896    for filename in filenames:897        lines = [[l.split(), l] for l in file(filename).readlines()]898        if "-k" in opts:899            k = int(opts["-k"]) - 1900            for line in lines:901                line[0][0], line[0][k] = line[0][k], line[0][0]902        if "-n" in opts:903            for line in lines:904                try:905                    line[0][0] = int(line[0][0])906                except:907                    pass908        lines.sort()909        rv.extend([line[1] for line in lines])910    return "".join(rv)911def sync():912    """sync913    flush system write back caches"""914    if os.name == "nt":915        retval = str(ctypes.windll.kernel32.SetSystemFileCacheSize(-1, -1, 0))916    else:917        _, _, retval = _shell_soe("sync")918    return retval919def tail(*args):920    """tail [-n NUM] [FILE...]921    show last NUM lines in file(s)"""922    opts, filenames = _getopts(args, "n:")923    all_files = expand(*filenames).splitlines()924    if "-n" in opts:925        lines = int(opts["-n"])926    else:927        lines = 10928    rv = []929    if lines > 0:930        for filename in all_files:931            rv.extend(file(filename).readlines()[-lines:])932    return "".join(rv)933def tar(*args):934    """tar [-ctxfC] PKG [FILE...]935    create/list/extract a tar package"""936    opts, filenames = _getopts(args, "ctxf:C:")937    pkg = opts.get("-f", expand(accept_pipe=True))938    if not pkg:939        raise ValueError("package filename missing (-f)")940    rv = []941    if "-t" in opts:942        tf = tarfile.TarFile.open(pkg)943        rv.extend(tf.getnames())944    elif "-x" in opts:945        tf = tarfile.TarFile.open(pkg)946        filenames = expand(*filenames, accept_pipe=False).splitlines()947        if filenames:948            for filename in filenames:949                tf.extract(filename, path=opts.get('-C', os.getcwd()))950                rv.append(filename)951        else:952            tf.extractall(path=opts.get('-C', os.getcwd()))953    elif "-c" in opts:954        if pkg.endswith(".bz2"):955            mode = "w:bz2"956        elif pkg.endswith(".gz"):957            mode = "w:gz"958        else:959            mode = "w"960        tf = tarfile.TarFile.open(pkg, mode)961        filenames = expand(*filenames, accept_pipe=False).splitlines()962        for filename in filenames:963            tf.add(filename)964        tf.close()965    return "\n".join(rv)966def unzip(*args):967    """unzip [-l] [-d DEST] PKG [FILE...]968    extract all or FILEs from PKG to DEST"""969    opts, filenames = _getopts(args, "ld:")970    filenames = expand(*filenames, min=1).splitlines()971    rv = []972    if "-d" in opts:973        dest_dir = opts["-d"]974    else:975        dest_dir = os.getcwd()976    if "-l" in opts:977        # only list files in archive978        for filename in filenames:979            for zi in zipfile.ZipFile(filename).infolist():980                rv.append("%8s  %s  %s" % (981                    zi.file_size,982                    datetime.datetime(*zi.date_time).strftime("%Y-%m-%d %H:%M"),983                    zi.filename))984    else:985        pkg, extract_files = filenames[0], filenames[1:]986        zf = zipfile.ZipFile(pkg)987        if extract_files:988            for extract_file in extract_files:989                zf.extract(extract_file,path=dest_dir)990                rv.append(extract_file)991        else:992            zf.extractall(path=dest_dir)993            rv.extend(zf.namelist())994    return "\n".join(rv)995def whoami():996    """whoami997    print user name"""998    try:999        return getpass.getuser()1000    except Exception:1001        return ""1002def xargs(*args):1003    """xargs CMD1004    run CMD with args from stdin"""1005    if not args:1006        raise ValueError("xargs: CMD missing")1007    if not _g_pipe_has_data:1008        raise ValueError("xargs: no get arguments in pipe")1009    retval = []1010    for arg in open(_g_pipe_filename):1011        arg = arg.strip()1012        funccall = args[0] + repr(tuple(args[1:]) + (arg,))1013        try:1014            func_rv = eval(funccall)1015            if func_rv and not func_rv.endswith("\n"):1016                func_rv += "\n"1017            retval.append(func_rv)1018        except Exception, e:1019            retval.append(str(e).splitlines()[-1] + "\n")1020    return "".join(retval)1021def xxd(*args):1022    """xxd [FILE...]1023    make a hexdump"""1024    all_files = expand(*args).splitlines()1025    rv = []1026    for filename in all_files:1027        addr = 01028        f = file(filename, "rb")1029        while True:1030            data16 = f.read(16)1031            if len(data16) == 0:1032                break1033            hex_line = []1034            hex_line.append(("%x" % (addr,)).zfill(8) + ": ")1035            for bindex, byte in enumerate(data16):1036                hex_line.append(("%x" % (ord(byte),)).zfill(2))1037                if bindex & 1 == 1:1038                    hex_line.append(" ")1039            s_hex_line = "".join(hex_line)1040            s_hex_line += " " * (51 - len(s_hex_line))1041            ascii_line = []1042            for byte in data16:1043                if 32 <= ord(byte) <= 126:1044                    ascii_line.append(byte)1045                else:1046                    ascii_line.append(".")1047            rv.append(s_hex_line + "".join(ascii_line))1048            addr += 161049    return "\n".join(rv)1050def zip(zipfilename, *filenames):1051    """zip ZIPFILE [FILE...]1052    add FILEs to ZIPFILE"""1053    filenames = expand(*filenames, accept_pipe=False, min=1).splitlines()1054    zf = zipfile.ZipFile(zipfilename, "a")1055    for filename in filenames:1056        zf.write(filename)1057    zf.close()1058    return ""1059def exit():1060    if os.name == "nt":1061        raise Exception("Close connection with Ctrl-Z + Return")1062    else:1063        raise Exception("Close connection with Ctrl-D")1064def pycosh_eval(cmdline):1065    if _g_pspycosh_conn:1066        return _g_pspycosh_conn.eval_("pycosh_eval(%s)" % (repr(cmdline,)))1067    funccall = cmd2py(cmdline)1068    try:1069        retval = eval(funccall)1070    except Exception, e:1071        retval = str(e).splitlines()[-1]1072    return retval1073def _parse_pycoshrc(contents):1074    """returns rc settings in a dictionary"""1075    for line in contents.splitlines():1076        retval = str(pycosh_eval(line))1077        if retval:1078            _output(retval)1079def _main():1080    histfile = os.path.join(os.path.expanduser("~"), ".pycosh_history")1081    rcfile = os.path.join(os.path.expanduser("~"), ".pycoshrc")...

nag_external_commands.py

Source:nag_external_commands.py

...71            line = line.rstrip() + "\n" + ind + word72    if txt[-1] == '\n':73        line += "\n"74    return line75def cmd2py(cmd, descr):76    cmd = cmd.strip().replace('<', '').replace('>', '')77    descr = descr.strip()78    c = cmd.split(';')79    name = c[0].lower()80    args = ', '.join(c[1:])81    method = "    def %s(self, %s):\n" % (name, args) + \82        '        """\n' + \83        '        %s\n' % wrap(descr, ind='        ') + \84        '        """\n' + \85        "        self.run('%s', %s)\n" % (c[0], args)86    return method87if __name__ == '__main__':88    root_url = 'http://old.nagios.org/developerinfo/externalcommands/'89    f = urllib2.urlopen(root_url + 'commandlist.php')90    soup = BeautifulSoup(f.read())91    f.close()92    content_table = soup.find('table', { 'class' : 'Content' })93    hrefs = content_table.findAll('a')[4:]94    for a in hrefs:95        #print a['href']96        #print a.string97        f = urllib2.urlopen(root_url + a['href'])98        s = BeautifulSoup(f.read())99        t = s.find('table', { 'class': 'Content' })100        tds = t.findAll('td')101        p = False102        cmd = ''103        descr = ''104        set_cmd = set_descr = False105        for td in tds:106            if set_cmd:107                cmd = unescape(td.string.decode())108                set_cmd = False109            elif set_descr:110                descr = unescape(td.string.decode())111                set_descr = False112            if td.string == 'Command Format:':113                set_cmd = True114            elif td.string == 'Description:':115                set_descr = True116            if cmd and descr:117                break118        if cmd and descr:119            print cmd2py(cmd, descr)...

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.