diff --git a/rere.py b/rere.py index 667fbbb..9f7bda3 100755 --- a/rere.py +++ b/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);