Verify expected shell commands

rere.py

@@ -8,7 +8,7 @@
 def read_blob_field(f: BinaryIO, name: bytes) -> bytes:
     line = f.readline()
     field = b':b ' + name + b' '
-    assert line.startswith(field), "%s" % field
+    assert line.startswith(field), field
     assert line.endswith(b'\n')
     size = int(line[len(field):-1])
     blob = f.read(size)
@@ -31,7 +31,7 @@ def write_blob_field(f: BinaryIO, name: bytes, blob: bytes):
     f.write(b'\n')
 
 def capture(shell: str) -> dict:
-    print(f"Capturing `{shell}`...")
+    print(f"CAPTURING: {shell}")
     process = subprocess.run(['sh', '-c', shell], capture_output = True)
     return {
         'shell': shell,
@@ -40,7 +40,11 @@ def capture(shell: str) -> dict:
         'stderr': process.stderr,
     }
 
-def dump_snapshots(file_path: str, snapshots: [dict]):
+def load_list(file_path: str) -> list[str]:
+    with open(file_path) as f:
+        return [line.strip() for line in f]
+
+def dump_snapshots(file_path: str, snapshots: list[dict]):
     with open(file_path, "wb") as f:
         write_int_field(f, b"count", len(snapshots))
         for snapshot in snapshots:
@@ -49,7 +53,7 @@ def dump_snapshots(file_path: str, snapshots: [dict]):
             write_blob_field(f, b"stdout", snapshot['stdout'])
             write_blob_field(f, b"stderr", snapshot['stderr'])
 
-def load_snapshots(file_path: str) -> [dict]:
+def load_snapshots(file_path: str) -> list[dict]:
     snapshots = []
     with open(file_path, "rb") as f:
         count = read_int_field(f, b"count")
@@ -83,10 +87,7 @@ def load_snapshots(file_path: str) -> [dict]:
             exit(1)
         test_list_path, *argv = argv
 
-        snapshots = []
-        with open(test_list_path) as f:
-            snapshots = [capture(shell.strip()) for shell in f]
-
+        snapshots = [capture(shell.strip()) for shell in load_list(test_list_path)]
         dump_snapshots(f'{test_list_path}.bi', snapshots)
     elif subcommand == 'replay':
         if len(argv) == 0:
@@ -95,29 +96,45 @@ def load_snapshots(file_path: str) -> [dict]:
             exit(1)
         test_list_path, *argv = argv
 
+        shells = load_list(test_list_path)
         snapshots = load_snapshots(f'{test_list_path}.bi')
-        for snapshot in snapshots:
-            print(f"Replaying `{snapshot['shell']}`...")
-            process = subprocess.run(['sh', '-c', snapshot['shell']], capture_output = True);
+
+        if len(shells) != len(snapshots):
+            print(f"UNEXPECTED: Amount of shell commands in f{test_list_path}")
+            print(f"    EXPECTED: {len(snapshots)}")
+            print(f"    ACTUAL:   {len(shells)}")
+            print(f"NOTE: You may want to do `{program_name} record {test_list_path}` to update {test_list_path}.bi")
+            exit(1)
+
+        for (shell, snapshot) in zip(shells, snapshots):
+            print(f"REPLAYING: {shell}")
+            if shell != snapshot['shell'].decode('utf-8'):
+                print(f"UNEXPECTED: shell command")
+                print(f"    EXPECTED: {snapshot['shell']}")
+                print(f"    ACTUAL:   {shell}")
+                print(f"NOTE: You may want to do `{program_name} record {test_list_path}` to update {test_list_path}.bi")
+                exit(1)
+            process = subprocess.run(['sh', '-c', shell], capture_output = True);
             if process.returncode != snapshot['returncode']:
-                print(f"UNEXPECTED RETURN CODE:")
+                print(f"UNEXPECTED: return code")
                 print(f"    EXPECTED: {snapshot['returncode']}")
                 print(f"    ACTUAL:   {process.returncode}")
                 exit(1)
             if process.stdout != snapshot['stdout']:
                 a = snapshot['stdout'].decode('utf-8').splitlines(keepends=True)
                 b = process.stdout.decode('utf-8').splitlines(keepends=True)
-                print(f"UNEXPECTED STDOUT:")
+                print(f"UNEXPECTED: stdout")
                 for line in unified_diff(a, b, fromfile="expected", tofile="actual"):
                     print(line, end='')
                 exit(1)
             if process.stderr != snapshot['stderr']:
                 a = snapshot['stderr'].decode('utf-8').splitlines(keepends=True)
                 b = process.stderr.decode('utf-8').splitlines(keepends=True)
-                print(f"UNEXPECTED STDERR:")
+                print(f"UNEXPECTED: stderr")
                 for line in unified_diff(a, b, fromfile="expected", tofile="actual"):
                     print(line, end='')
                 exit(1)
+        print('OK')
     else:
         print(f'ERROR: unknown subcommand {subcommand}');
         exit(1);