How to use _grep method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

_grep2.py

Source:_grep2.py

1#!/usr/bin/env python32r"""3usage: bin/_grep.py [--help] [-h] [--rip [SHRED]] [PATTERN ...]4write and run code to search for patterns, or just write it5positional arguments:6  PATTERN        search key, in the syntax of a Python regular expression7options:8  --help         show this help message and exit9  -h             say less (same as grep -h)10  --rip [SHRED]  rip code to stdout (or to a filename with a '.' dot in it)11quirks in the syntax:12  writes code that works like "|zgrep -in ... |zgrep -in ..."13  welcomes only more patterns, leaves you to do the work of searching multiple files14examples:15  bin/_grep.py key1  # search stdin lines for key, ignoring case and wordseps and order16  bin/_grep.py key1 key2  # search stdin lines for both keys, ignoring case etc17  bin/_grep.py key1 key2 --r  # just print the code, don't run it18  bin/_grep.py key1 key2 --rip py  # rip out only python code, no other kind19  bin/_grep.py key1 key2 --r p.py  # just save the code in "p.py", don't run it20  grep -n . bin/*.py |bin/_grep.py key1 key2  # search multiple py files21  bin/_grep.py -- -key3 # search stdin for "key3" but spell it starting with a dash22"""23# FIXME: convert between regexes, such as from "grep -E" to "grep"24# FIXME: argdoc: accept, do not require, [--] separating optionals from positionals25# FIXME: see unmarked args as and-also when presented as patterns then >= 0 files26import os27import sys28import textwrap29import argdoc30def main(argv):31    """Run once"""32    # Parse the command line, per the top-of-file docstring33    args = argdoc.parse_args(argv[1:])34    if args.h:35        sys.stderr.write("bin/_grep.py: -h not implemented\n")36        sys.exit(2)  # exit 2 from rejecting usage37    # Pick out the ext (and notice when given a filename)38    p_py = None39    ext = ".py3"  # default: ".py3"40    if args.rip not in (False, None):41        ext = args.rip  # # ".py3" etc mean themselves42        if not args.rip.startswith(os.path.extsep):43            ext = os.path.extsep + args.rip  # "py3" etc means ".py3" etc44            if os.path.extsep in args.rip:45                p_py = args.rip46                ext = os.path.splitext(p_py)[-1]  # else {name}.{ext}47    # Agree to speak only some languages48    exts = ".py .py2 .py23 .py3".split()49    if ext not in exts:50        sys.stderr.write(51            "bin/_grep.py: choose ext from one of {}, not {!r}\n".format(52                "|".join(exts), ext53            )54        )55        sys.exit(2)  # exit 2 from rejecting usage56    # Write the code57    py = emit_searches_as_py(ext, p_py=p_py, patterns=args.patterns)58    # Run the code, print the code, or save the code59    if args.rip is False:60        exec(py, globals())  # pylint: disable=exec-used61        # call through Class ZCatLines via the main Globals62    elif p_py is not None:63        with open(p_py, "w") as writing:64            writing.write(py)65    else:66        sys.stdout.write(py)67        sys.stdout.flush()68def emit_searches_as_py(ext, p_py, patterns):69    py = r""70    py += emit_header(ext, p_py=p_py)71    py += "\n"72    py += emit_body(patterns)73    py += "\n"74    py += "\n"  # twice75    py += emit_trailer()76    return py77def emit_header(ext, p_py):78    py = (79        textwrap.dedent(80            r'''81            #!/usr/bin/env python382            """83            usage: p.py84            search "/dev/stdin" for keys, ignoring case and wordseps and order85            """86            import contextlib87            import gzip88            import io89            import re90            import sys91            def main(argv):92                lines = {}93                filename = "/dev/stdin"94                # with open(filename) as reading:95                with ZCatLines(filename) as reading:96                    for (index, ended_line) in enumerate(reading):97                        line_number = 1 + index98                        line = ended_line.splitlines()[0]99                        lines[index] = line100                        text = line.expandtabs(tabsize=8).strip()101                        text = text.replace("_", "-")102                        text = text.replace("-", " ")103                        text = text.strip()104            '''105        ).strip()106        + "\n"107    )108    if p_py is not None:109        py = py.replace(r"p.py", p_py)110    if ext == ".py2":111        py = py.replace(r"python3", r"python2")112    elif ext == ".py23":113        py = py.replace(r"python3", r"python")114    return py115def emit_trailer():116    py = r""117    py += (118        textwrap.dedent(119            r'''120            # deffed in many files  # missing from docs.python.org121            class ZCatLines(contextlib.ContextDecorator):122                """GUnzip and read each line, else read each line"""123                def __init__(self, filename):  # a la open(file), gzip.open(filename)124                    self.filename = filename125                    self.closing = None126                def __enter__(self):127                    filename = self.filename128                    # read from uncompressed tty129                    reading = open(filename)130                    if reading.isatty():131                        self.closing = reading  # read a tty132                        return self.closing133                    reading.close()134                    # read from uncompressed file135                    with open(filename, mode="rb") as peeking:136                        peeked_bytes = peeking.read()137                    unzipped = None138                    with io.BytesIO(peeked_bytes) as copying:139                        with gzip.open(copying) as unzipping:140                            try:141                                unzipped = unzipping.read()142                            except Exception:  # often gzip.BadGzipFile143                                chars = peeked_bytes.decode()144                                self.closing = io.StringIO(chars)  # read uncompressed145                                return self.closing146                    # read from compressed file147                    lines = unzipped.decode().splitlines(keepends=True)148                    return lines149                def __exit__(self, *_):150                    if self.closing is not None:151                        return self.closing152            '''153        ).strip()154        + "\n"155    )156    py += "\n"157    py += "\n"  # twice158    py += (159        textwrap.dedent(160            r"""161            if __name__ == "__main__":162                sys.exit(main(sys.argv))163            """164        ).strip()165        + "\n"166    )167    return py168def emit_body(patterns):169    py = r""170    py += (171        textwrap.dedent(172            r"""173                if False:174                    print("visiting {}:{}".format(line_number, text), file=sys.stderr)175            """176        ).strip()177        + "\n"178    )179    py += "\n"180    for pattern in patterns:181        regex = pattern.expandtabs(tabsize=8).strip()182        regex = regex.replace("_", "-")183        regex = regex.replace("-", " ")184        regex = regex.strip()185        assert not (set(regex) & set(r"\""))  # FIXME: accept more patterns186        pattern_py = (187            textwrap.dedent(188                r"""189                    if not re.search(r"{regex}", string=text, flags=re.IGNORECASE):190                        continue191                """192            )193            .format(regex=regex)194            .strip()195            + "\n"196        )197        py = py + pattern_py198    py += "\n"199    py += (200        textwrap.dedent(201            r"""202                print("{}:{}".format(line_number, line), file=sys.stderr)203                sys.stdout.flush()204            """205        ).strip()206        + "\n"207    )208    dents = 3 * "    "209    py = "\n".join((dents + _).rstrip() for _ in py.splitlines()) + "\n"210    return py211_ = """212non-interactive examples, for testing more modes of Class ZCatLines:213  cat bin/_grep.py |bin/_grep.py key1214  rm -fr b.bin b.bin.gz215  cp -ip bin/_grep.py b.bin216  gzip b.bin  # make b.bin.gz217  cat b.bin.gz |bin/_grep.py key1 key2218  bin/_grep.py key1 key2 --rip >o219  bin/_grep.py key1 key2 --rip py >o.py220  bin/_grep.py key1 key2 --rip .py2 >o.py2221  bin/_grep.py key1 key2 --rip p.py23222  bin/_grep.py key1 key2 --rip p.py3223  diff -burp p.py3 o224  diff -burp p.py3 o.py225  diff -burp p.py3 o.py2226  diff -burp p.py3 p.py23227"""228if __name__ == "__main__":229    sys.exit(main(sys.argv))...

vrc_task_2_scoring_test

Source:vrc_task_2_scoring_test

...43            pass44	self.pub.unregister()45	return self.result46class Tester(unittest.TestCase):47    def _grep(self, fname, string):48        fullpath = os.path.join(LOGDIR, fname)49        for line in open(fullpath, 'r'):50            if string in line:51                return True52        return False53    # This function isn't used right now; need to sort out the state.log54    # flushing problem before this can be reliably used.55    def _get_atlas_pose(self, fname):56        # Check the current robot pose, from the incomplete gazebo log57        # file. Not sure how robust this is.58        fullpath = os.path.join(LOGDIR, fname)59        while not os.path.exists(fullpath):60            print('Waiting for %s to exist'%(fullpath))61            time.sleep(0.5)62        tmp = tempfile.NamedTemporaryFile(delete=False)63        tmp.close()64        shutil.copyfile(fullpath, tmp.name)65        # Terminate our copy of the log file66        with open(tmp.name, 'a') as f:67            f.write('</gazebo_log>\n')68        # Use gzlog to extract the pose69        cmd = ['gzlog', 'echo', tmp.name, '-r', '--filter', 'atlas.pose']70        po = subprocess.Popen(cmd, stdout=subprocess.PIPE, 71                              stderr=subprocess.PIPE)72        out,err = po.communicate()73        # Take the last non-empty line74        pose = []75        for l in reversed(out.split('\n')):76            if len(l) > 0 and l[0] != ' ':77                pose = l.split()78                break79        self.assertTrue(len(pose) == 6)80        for i in range(len(pose)):81            pose[i] = float(pose[i])82        os.remove(tmp.name)83        return pose84    def score_callback(self, data):85        self.last_score = data86        self.total_score_msgs += 187        rospy.loginfo(self.last_score.completion_score)88    # wait and get new score data 89    def wait_new_score_msg(self):90        current = self.total_score_msgs91        while (self.total_score_msgs <= current):92	        time.sleep(1)93    def assertROSScore(self, expected_score):94	result = False95	for i in range(4): # 4 seconds to be sure message is not arriving96	    self.wait_new_score_msg()97	    ros_pkg = self.last_score98	    if (ros_pkg.completion_score == expected_score):99		result = True100		break101	self.assertTrue(result,102	  'Score in ros topic is not ' + str(expected_score))103	return ros_pkg104    105    def assertROSElapsedTimeZero(self, ros_pkg):106	self.assertEqual(ros_pkg.wall_time_elapsed, rospy.Time(0))107	self.assertEqual(ros_pkg.sim_time_elapsed, rospy.Time(0))108	109    def assertROSElapsedTimeNotZero(self, ros_pkg):110	self.assertNotEqual(ros_pkg.wall_time_elapsed, rospy.Time(0))111	self.assertNotEqual(ros_pkg.sim_time_elapsed, rospy.Time(0))112    def assertROSNoFalls(self, ros_pkg):113    	self.assertEqual(ros_pkg.falls, 0)114    def test_scoring(self):115        pose_pub = rospy.Publisher('atlas/set_pose', Pose)116        pose = Pose()117     	self.total_score_msgs = 0118        self.score_pub = rospy.Subscriber('/vrc_score', VRCScore, self.score_callback)119        # Wait for subscribers to hook up120        time.sleep(3.0)121	# Previous checks:122	# - Check that robot is standup123    	stand_up_checker = OrientationChecker()124	self.assertTrue(stand_up_checker.run(),125	      'Robot seems to be fallen')126	# - no point yet127	pkg = self.assertROSScore(0)128	self.assertFalse(self._grep('score.log', 129          'Successfully passed through gate 1'))130	# - no elapsed time131	self.assertROSElapsedTimeZero(pkg)132	# - no falls133	self.assertROSNoFalls(pkg)134        # Teleport through first gate135        self.assertFalse(self._grep('score.log', 136          'Successfully passed through gate 1'))137        x_str, y_str, z_str = rospy.get_param("~pose_gate1").split(" ")138        pose.position.x = float(x_str)139        pose.position.y = float(y_str)140        pose.position.z = float(z_str)141        pose_pub.publish(pose)142        time.sleep(3.0)143        self.assertTrue(self._grep('score.log', 144          'Successfully passed through gate 1'))145	pkg = self.assertROSScore(0)146	self.assertROSElapsedTimeNotZero(pkg)147        # Teleport through second gate148        self.assertFalse(self._grep('score.log', 149          'Successfully passed through gate 2'))150        pose.position.x = -0.03151        pose.position.y = -27.76152        pose.position.z = 1.00153        pose_pub.publish(pose)154        time.sleep(3.0)155        pose.position.x = 1.36156        pose.position.y = -19.71157        pose.position.z = 1.00158        pose_pub.publish(pose)159        time.sleep(3.0)160        self.assertTrue(self._grep('score.log', 161          'Successfully passed through gate 2'))162        self.assertROSScore(1)163        # Teleport through third gate164        self.assertFalse(self._grep('score.log', 165          'Successfully passed through gate 3'))166        pose.position.x = 0.644167        pose.position.y = -7.35168        pose.position.z = 1.00169        pose_pub.publish(pose)170        time.sleep(3.0)171        self.assertTrue(self._grep('score.log', 172          'Successfully passed through gate 3'))173        self.assertROSScore(2)174        # Teleport through fourth gate175        self.assertFalse(self._grep('score.log', 176          'Successfully passed through gate 4'))177        pose.position.x = 6.68178        pose.position.y = 4.44179        pose.position.z = 1.00180        pose_pub.publish(pose)181        time.sleep(3.0)182        pose.position.x = 7.00183        pose.position.y = 12.07184        pose.position.z = 1.00185        pose_pub.publish(pose)186        time.sleep(3.0)187        self.assertTrue(self._grep('score.log', 188          'Successfully passed through gate 4'))189        self.assertROSScore(3)190        # Teleport through fifth gate191        self.assertFalse(self._grep('score.log', 192          'Successfully passed through gate 5'))193        pose.position.x = 6.30194        pose.position.y = 20.31195        pose.position.z = 1.00196        pose_pub.publish(pose)197        time.sleep(3.0)198        self.assertTrue(self._grep('score.log', 199          'Successfully passed through gate 5'))200        self.assertROSScore(4)201if __name__ == '__main__':202    rospy.init_node('vrc_task_2_scoring_test', anonymous=True)203    try:204	LOGDIR = rospy.get_param('~logdir')205    except KeyError:206	rospy.loginfo("Use default value for logdir: " + LOGDIR)207    # Wait until /clock is being published; this can take an unpredictable208    # amount of time when we're downloading models.209    while rospy.Time.now().to_sec() == 0.0:210        print('Waiting for Gazebo to start...')211        time.sleep(1.0)212    # Take an extra nap, to allow plugins to be loaded...

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.