Implement peek command

src/coredumpy/coredumpy.py

@@ -100,6 +100,22 @@ def load(cls, path):
         pdb_instance.interaction(frame, None)
         PyObjectProxy.clear()  # pragma: no cover
 
+    @classmethod
+    def peek(cls, path):
+        with gzip.open(path, "rt") as f:
+            data = json.load(f)
+
+        from coredumpy import __version__
+        if data["metadata"]["version"] != __version__:  # pragma: no cover
+            print(f"Warning! the dump file is created by {data['metadata']['version']}\n"
+                  f"but the current coredumpy version is {__version__}")
+        patch_all()
+        metadata = data["metadata"]
+        system = metadata["system"]
+        print(f"{os.path.abspath(path)}")
+        print(f"    Python v{metadata['python_version']} on {system['system']} {system['node']} {system['release']}")
+        print(f"    {metadata['dump_time']}")
+
     @classmethod
     def get_metadata(cls):
         from coredumpy import __version__
@@ -118,3 +134,4 @@ def get_metadata(cls):
 
 dump = Coredumpy.dump
 load = Coredumpy.load
+peek = Coredumpy.peek

src/coredumpy/main.py

@@ -5,7 +5,7 @@
 import argparse
 import os
 
-from .coredumpy import load
+from .coredumpy import load, peek
 
 
 def main():
@@ -15,10 +15,30 @@ def main():
     subparsers_load = subparsers.add_parser("load", help="Load a dump file.")
     subparsers_load.add_argument("file", type=str, help="The dump file to load.")
 
+    subparsers_load = subparsers.add_parser("peek", help="Peek a dump file.")
+    subparsers_load.add_argument("files", help="The dump file to load.", nargs="+")
+
     args = parser.parse_args()
 
     if args.command == "load":
         if os.path.exists(args.file):
             load(args.file)
         else:
             print(f"File {args.file} not found.")
+    elif args.command == "peek":
+        for file in args.files:
+            if os.path.exists(file):
+                if os.path.isdir(file):
+                    for f in os.listdir(file):
+                        try:
+                            path = os.path.join(file, f)
+                            peek(path)
+                        except Exception:
+                            pass
+                else:
+                    try:
+                        peek(file)
+                    except Exception:
+                        pass
+            else:
+                print(f"File {file} not found.")

tests/base.py

@@ -46,3 +46,11 @@ def run_script(self, script, expected_returncode=0):
             self.assertEqual(process.returncode, expected_returncode,
                              f"script failed with return code {process.returncode}\n{stderr}")
         return stdout, stderr
+
+    def run_peek(self, paths):
+        process = subprocess.Popen(normalize_commands(["coredumpy", "peek"] + paths),
+                                   stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        stdout, stderr = process.communicate()
+        stdout = stdout.decode(errors='backslashreplace')
+        stderr = stderr.decode(errors='backslashreplace')
+        return stdout, stderr

tests/test_basic.py

@@ -67,6 +67,26 @@ def g(arg):
         self.assertIn("return 1 / arg", stdout)
         self.assertIn("0", stdout)
 
+    def test_peek(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script = f"""
+                import coredumpy
+                coredumpy.dump(directory={repr(tmpdir)})
+            """
+            self.run_script(script)
+            self.run_script(script)
+            with open(os.path.join(tmpdir, "invalid"), "w") as f:
+                f.write("{invalid}")
+
+            self.assertEqual(len(os.listdir(tmpdir)), 3)
+            stdout, _ = self.run_peek([tmpdir])
+            stdout2, _ = self.run_peek([os.path.join(tmpdir, file) for file in os.listdir(tmpdir)])
+
+            self.assertEqual(stdout, stdout2)
+
+            stdout, _ = self.run_peek([os.path.join(tmpdir, "nosuchfile")])
+            self.assertIn("not found", stdout)
+
     def test_nonexist_file(self):
         stdout, stderr = self.run_test("", "nonexist_dump", [])
         self.assertIn("File nonexist_dump not found", stdout)