Grep

bash_builtins.py

@@ -65,20 +65,18 @@ def wc_function(args, stdin, stdout):
     parsed_args = parser.parse_args(args)
     filepath = vars(parsed_args).get("FILE")
     fout = open(stdout, "w", closefd=False)  # we should not close stdout
+    fin = None
     if filepath:
         # READ FILEPATH
         fin = open(filepath, "r")
-        lines = fin.readlines()
-        words_count = sum([len(line.split()) for line in lines])
-        file_size = os.path.getsize(filepath)
-        fout.write(str(len(lines)) + " " + str(words_count) + " " + str(file_size))
-        fin.close()
     else:
         # READ STDIN
         fin = open(stdin, "r", closefd=False)
+
+    with fin:
         lines = fin.readlines()
         words_count = sum([len(line.split()) for line in lines])
-        file_size = sum([len(line) for line in lines])
+        file_size = sum([len(line) for line in lines])  # TODO maybe replace with os.path.getsize() for files
         fout.write(str(len(lines)) + " " + str(words_count) + " " + str(file_size))
 
 
@@ -107,12 +105,60 @@ def exit_function(args, stdin, stdout):
     os.kill(os.getpid(), signal.SIGTERM)
 
 
+def grep_function(args, stdin, stdout):
+    """
+    Match lines in FILE or stdin with PATTERN
+    :param args: arguments
+    :param stdin: input file descriptor
+    :param stdout: output file descriptor
+    :return: nothing
+    """
+    parser = argparse.ArgumentParser(prog="grep", description='print lines matching a pattern')
+    parser.add_argument("-i", action='store_true',
+                        help='Ignore case distinctions, so that characters that differ only in case match each other.')
+    parser.add_argument("-w", action='store_true',
+                        help='Select  only  those  lines  containing  matches  that form whole words.')
+    parser.add_argument("-A", nargs='?', action='store', default=0,
+                        help='Print A lines after matched lines')
+    parser.add_argument("PATTERN", help='Regular expression to be matched in line')
+    parser.add_argument("FILE", nargs='?', help='Path to file to scan for')
+    parsed_args = vars(parser.parse_args(args))
+
+    pattern = parsed_args.get("PATTERN")
+    if parsed_args.get("w"):
+        pattern = r'\b' + pattern + r'\b'
+    flags = 0
+    if parsed_args.get('i'):
+        flags |= re.IGNORECASE
+
+    regexp = re.compile(pattern, flags)
+
+    fin = None
+    if parsed_args.get('FILE'):
+        fin = open(parsed_args.get('FILE'), 'r', closefd=True)
+    else:
+        fin = open(stdin, 'r', closefd=False)
+
+    fout = open(stdout, 'w', closefd=False)
+
+    with fin, fout:
+        after_parameter = int(parsed_args.get('A'))
+        remaining_after = 0
+        for line in fin:
+            if regexp.search(line):
+                remaining_after = after_parameter + 1
+            if remaining_after > 0:
+                remaining_after -= 1
+                fout.write(line)
+
+
 command_to_function = {
     "cat": cat_function,
     "echo": echo_function,
     "wc": wc_function,
     "pwd": pwd_function,
-    "exit": exit_function
+    "exit": exit_function,
+    "grep": grep_function
 }
 
 

unittests.py

@@ -29,8 +29,8 @@ def test_cat(self):
         file = tempfile.NamedTemporaryFile("w", delete=False)
         filename = file.name
         file.write(self.TEST_STRING)
-        read_pipe, write_pipe = os.pipe()
         file.close()
+        read_pipe, write_pipe = os.pipe()
 
         thread = bash_builtins.simple_interprete_single_builtin_command(["cat", file.name],
                                                                         stdin=sys.stdin.fileno(),
@@ -55,8 +55,8 @@ def test_wc(self):
         file = tempfile.NamedTemporaryFile("w", delete=False)
         filename = file.name
         file.write(self.TEST_STRING)
-        read_pipe, write_pipe = os.pipe()
         file.close()
+        read_pipe, write_pipe = os.pipe()
 
         thread = bash_builtins.simple_interprete_single_builtin_command(["wc", file.name],
                                                                         stdin=sys.stdin.fileno(),
@@ -65,7 +65,7 @@ def test_wc(self):
         os.close(write_pipe)
         with open(read_pipe, "r") as fin:
             file_content = fin.read()
-            self.assertEqual(file_content, self.TEST_STRING_WC + " " + str(os.path.getsize(filename)))
+            self.assertEqual(file_content, self.TEST_STRING_WC + " " + str(len(self.TEST_STRING)))
         os.remove(filename)
 
     def test_interactive_wc(self):
@@ -93,6 +93,99 @@ def test_pwd(self):
         with open(read_pipe, "r") as fin:
             self.assertEqual(fin.read(), os.getcwd())
 
+    def test_grep(self):
+        read_stdin_pipe, write_stdin_pipe = os.pipe()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "abc"],
+                                                                        stdin=read_stdin_pipe,
+                                                                        stdout=write_pipe)
+        with open(write_stdin_pipe, "w") as finout:
+            finout.writelines(
+                ["a\n", "ab\n", "abc\n", "abcd\n", "dcba\n", "aab\n", "aabcd\n", "aabc\n", "abacbca\n", "abc"])
+        thread.wait()
+        os.close(read_stdin_pipe)
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(), ['abc\n', 'abcd\n', 'aabcd\n', 'aabc\n', 'abc'])
+
+    def test_grep_with_file(self):
+        file = tempfile.NamedTemporaryFile("w", delete=False)
+        filename = file.name
+        file.writelines(["a\n", "ab\n", "abc\n", "abcd\n", "dcba\n", "aab\n", "aabcd\n", "aabc\n", "abacbca\n", "abc"])
+        file.close()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "abc", filename],
+                                                                        stdin=sys.stdin.fileno(),
+                                                                        stdout=write_pipe)
+        thread.wait()
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(), ['abc\n', 'abcd\n', 'aabcd\n', 'aabc\n', 'abc'])
+        os.remove(filename)
+
+    def test_grep_i(self):
+        read_stdin_pipe, write_stdin_pipe = os.pipe()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "-i", "aBc"],
+                                                                        stdin=read_stdin_pipe,
+                                                                        stdout=write_pipe)
+        with open(write_stdin_pipe, "w") as finout:
+            finout.writelines(
+                ["a\n", "ab\n", "abc\n", "ABcd\n", "dCbA\n", "aaB\n", "aabCd\n", "aABC\n", "aBacBcA\n", "AbC"])
+        thread.wait()
+        os.close(read_stdin_pipe)
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(), ['abc\n', 'ABcd\n', 'aabCd\n', 'aABC\n', 'AbC'])
+
+    def test_grep_w(self):
+        read_stdin_pipe, write_stdin_pipe = os.pipe()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "-w", "abc def ghi"],
+                                                                        stdin=read_stdin_pipe,
+                                                                        stdout=write_pipe)
+        with open(write_stdin_pipe, "w") as finout:
+            finout.writelines(
+                ["a\n", "abc def ghi\n", "abc def ghij\n", "abc abc def ghi ghi\n", "aabc def ghi\n",
+                 " abc def ghi \n"])
+        thread.wait()
+        os.close(read_stdin_pipe)
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(), ['abc def ghi\n', 'abc abc def ghi ghi\n', ' abc def ghi \n'])
+
+    def test_grep_A1(self):
+        read_stdin_pipe, write_stdin_pipe = os.pipe()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "-A", "1", "abc"],
+                                                                        stdin=read_stdin_pipe,
+                                                                        stdout=write_pipe)
+        with open(write_stdin_pipe, "w") as finout:
+            finout.writelines(
+                ["a\n", "ab\n", "abc\n", "abcd\n", "dcba\n", "aab\n", "aabcd\n", "aabc\n", "abacbca\n", "abc"])
+        thread.wait()
+        os.close(read_stdin_pipe)
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(),
+                                 ['abc\n', 'abcd\n', 'dcba\n', 'aabcd\n', 'aabc\n', 'abacbca\n', 'abc'])
+
+    def test_grep_A2(self):
+        read_stdin_pipe, write_stdin_pipe = os.pipe()
+        read_pipe, write_pipe = os.pipe()
+        thread = bash_builtins.simple_interprete_single_builtin_command(["grep", "-A", "2", "abc"],
+                                                                        stdin=read_stdin_pipe,
+                                                                        stdout=write_pipe)
+        with open(write_stdin_pipe, "w") as finout:
+            finout.writelines(
+                ["a\n", "ab\n", "abc\n", "abcd\n", "dcba\n", "aab\n", "aabcd\n", "aabc\n", "abacbca\n", "abc"])
+        thread.wait()
+        os.close(read_stdin_pipe)
+        os.close(write_pipe)
+        with open(read_pipe, "r") as fin:
+            self.assertListEqual(fin.readlines(),
+                                 ['abc\n', 'abcd\n', 'dcba\n', 'aab\n', 'aabcd\n', 'aabc\n', 'abacbca\n', 'abc'])
+
 
 class TestTokenize(unittest.TestCase):
     def test_shlex_tokenize(self):