From a99dbeba4b54b6d709742b722fbcd7b6952631c4 Mon Sep 17 00:00:00 2001 From: codeofwxz Date: Sat, 12 Sep 2026 23:49:55 +0800 Subject: [PATCH] Preserve inplace input until processing and encoding succeed --- srt_tools/srt-deduplicate | 5 +- srt_tools/srt-fixed-timeshift | 5 +- srt_tools/srt-linear-timeshift | 5 +- srt_tools/srt-lines-matching | 5 +- srt_tools/srt-mux | 5 +- srt_tools/srt-normalise | 5 +- srt_tools/srt-process | 5 +- srt_tools/utils.py | 20 +++- tests/test_tools.py | 178 +++++++++++++++++++++++++++++++++ 9 files changed, 204 insertions(+), 29 deletions(-) create mode 100644 tests/test_tools.py diff --git a/srt_tools/srt-deduplicate b/srt_tools/srt-deduplicate index 29ad29d..9cd0963 100755 --- a/srt_tools/srt-deduplicate +++ b/srt_tools/srt-deduplicate @@ -86,10 +86,7 @@ def main(): output = srt_tools.utils.compose_suggest_on_fail(subs, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-fixed-timeshift b/srt_tools/srt-fixed-timeshift index bb9417d..d9f9f5e 100755 --- a/srt_tools/srt-fixed-timeshift +++ b/srt_tools/srt-fixed-timeshift @@ -37,10 +37,7 @@ def main(): corrected_subs = scalar_correct_subs(args.input, args.seconds) output = srt_tools.utils.compose_suggest_on_fail(corrected_subs, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-linear-timeshift b/srt_tools/srt-linear-timeshift index 9db33ea..326371c 100755 --- a/srt_tools/srt-linear-timeshift +++ b/srt_tools/srt-linear-timeshift @@ -95,10 +95,7 @@ def main(): corrected_subs = linear_correct_subs(args.input, angular, linear) output = srt_tools.utils.compose_suggest_on_fail(corrected_subs, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-lines-matching b/srt_tools/srt-lines-matching index 514594a..4e8cf7f 100755 --- a/srt_tools/srt-lines-matching +++ b/srt_tools/srt-lines-matching @@ -75,10 +75,7 @@ def main(): matching_subtitles_only, strict=args.strict ) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-mux b/srt_tools/srt-mux index 62edf6e..bf7699e 100755 --- a/srt_tools/srt-mux +++ b/srt_tools/srt-mux @@ -102,10 +102,7 @@ def main(): output = srt_tools.utils.compose_suggest_on_fail(muxed_subs, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-normalise b/srt_tools/srt-normalise index 7d36e95..f644534 100755 --- a/srt_tools/srt-normalise +++ b/srt_tools/srt-normalise @@ -18,10 +18,7 @@ def main(): srt_tools.utils.set_basic_args(args) output = srt_tools.utils.compose_suggest_on_fail(args.input, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/srt-process b/srt_tools/srt-process index 09cacbe..24899a2 100755 --- a/srt_tools/srt-process +++ b/srt_tools/srt-process @@ -47,10 +47,7 @@ def main(): processed_subs = strip_to_matching_lines_only(args.input, args.module, args.func) output = srt_tools.utils.compose_suggest_on_fail(processed_subs, strict=args.strict) - try: - args.output.write(output) - except (UnicodeEncodeError, TypeError): # Python 2 fallback - args.output.write(output.encode(args.encoding)) + srt_tools.utils.write_output(args, output) if __name__ == "__main__": # pragma: no cover diff --git a/srt_tools/utils.py b/srt_tools/utils.py index e02c5eb..b130093 100755 --- a/srt_tools/utils.py +++ b/srt_tools/utils.py @@ -199,10 +199,28 @@ def set_basic_args(args): args.input = srt.parse( f.read(), ignore_errors=args.ignore_parsing_errors ) - else: + elif not getattr(args, "inplace", False): args.output = w_enc(open(args.output, "wb")) +def write_output(args, output): + """Write composed output, delaying in-place truncation until encoding succeeds. + + This protects against parsing, processing, and encoding errors, but not + errors during the file write itself. Opening the existing file preserves + its inode and follows links, just as the original in-place write did. + """ + if getattr(args, "inplace", False): + encoded = output.encode(args.encoding or "utf-8") + with open(args.output, "wb") as output_file: + output_file.write(encoded) + else: + try: + args.output.write(output) + except (UnicodeEncodeError, TypeError): # Python 2 fallback + args.output.write(output.encode(args.encoding)) + + def compose_suggest_on_fail(subs, strict=True): try: return srt.compose(subs, strict=strict, eol=os.linesep, in_place=True) diff --git a/tests/test_tools.py b/tests/test_tools.py new file mode 100644 index 0000000..f28f0d5 --- /dev/null +++ b/tests/test_tools.py @@ -0,0 +1,178 @@ +"""Exercise the actual command-line scripts using synthetic subtitle files.""" + +from __future__ import unicode_literals + +import os +import subprocess +import sys + +import pytest + + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +VALID = b"1\n00:00:01,000 --> 00:00:02,000\nSynthetic subtitle\n\n" +COMMANDS = [ + ("normalise", []), + ("deduplicate", []), + ("fixed-timeshift", ["--seconds", "1"]), + ( + "linear-timeshift", + [ + "--f1", + "00:00:01,000", + "--t1", + "00:00:01,000", + "--f2", + "00:00:02,000", + "--t2", + "00:00:03,000", + ], + ), + ("process", ["-f", "lambda text: text.upper()"]), + ("lines-matching", ["-f", "lambda text: True"]), +] + + +def run_tool(name, args): + launcher = ( + "import sys,runpy; sys.path.insert(0,sys.argv.pop(1)); " + "runpy.run_path(sys.argv.pop(1),run_name='__main__')" + ) + command = [ + sys.executable, + "-c", + launcher, + ROOT, + os.path.join(ROOT, "srt_tools", "srt-" + name), + ] + args + process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + stdout, stderr = process.communicate() + return process.returncode, stdout, stderr + + +@pytest.mark.parametrize("name,extra", COMMANDS) +def test_inplace_parse_failure_preserves_source(tmpdir, name, extra): + source = tmpdir.join("invalid.srt") + original = b"NOT AN SRT FILE\n" + source.write_binary(original) + status, _, stderr = run_tool(name, ["-i", str(source), "--inplace"] + extra) + assert status != 0 + assert b"SRTParseError" in stderr + assert source.read_binary() == original + + +@pytest.mark.parametrize("name,extra", COMMANDS) +def test_inplace_matches_stdout_and_file_output(tmpdir, name, extra): + source = tmpdir.join("valid.srt") + destination = tmpdir.join("output.srt") + source.write_binary(VALID) + status, expected, stderr = run_tool(name, ["-i", str(source)] + extra) + assert status == 0, stderr + assert source.read_binary() == VALID + status, _, stderr = run_tool( + name, ["-i", str(source), "-o", str(destination)] + extra + ) + assert status == 0, stderr + assert destination.read_binary() == expected + before = os.stat(str(source)) + status, _, stderr = run_tool(name, ["-i", str(source), "--inplace"] + extra) + assert status == 0, stderr + assert source.read_binary() == expected + after = os.stat(str(source)) + assert (before.st_ino, before.st_mode) == (after.st_ino, after.st_mode) + + +@pytest.mark.parametrize( + "name,extra,error", + [ + ("process", ["-f", "lambda text: 1 / 0"], b"ZeroDivisionError"), + ("lines-matching", ["-f", "lambda text: 1 / 0"], b"ZeroDivisionError"), + ("fixed-timeshift", ["--seconds", "1e100"], b"OverflowError"), + ( + "process", + ["--encoding", "ascii", "-f", "lambda text: text + u'\\u00e9'"], + b"UnicodeEncodeError", + ), + ], +) +def test_inplace_conversion_failure_preserves_source(tmpdir, name, extra, error): + source = tmpdir.join("valid.srt") + source.write_binary(VALID) + status, _, stderr = run_tool(name, ["-i", str(source), "--inplace"] + extra) + assert status != 0 + assert error in stderr + assert source.read_binary() == VALID + + +def test_mux_file_output_matches_stdout(tmpdir): + source = tmpdir.join("valid.srt") + destination = tmpdir.join("output.srt") + source.write_binary(VALID) + status, expected, stderr = run_tool("mux", ["-i", str(source)]) + assert status == 0, stderr + status, _, stderr = run_tool("mux", ["-i", str(source), "-o", str(destination)]) + assert status == 0, stderr + assert destination.read_binary() == expected + + +@pytest.mark.parametrize("kind", ["link", "symlink"]) +def test_inplace_preserves_links(tmpdir, kind): + target = tmpdir.join("target.srt") + source = tmpdir.join("source.srt") + target.write_binary(VALID) + if not hasattr(os, kind): + pytest.skip("This Python does not provide os." + kind) + try: + getattr(os, kind)(str(target), str(source)) + except OSError as error: + pytest.skip("Cannot create test link: " + str(error)) + before = os.stat(str(target)) + status, _, stderr = run_tool("normalise", ["-i", str(source), "--inplace"]) + assert status == 0, stderr + assert target.read_binary() == source.read_binary() + assert os.stat(str(source)).st_ino == before.st_ino + assert os.stat(str(target)).st_ino == before.st_ino + if kind == "symlink": + assert os.path.islink(str(source)) + + +@pytest.mark.parametrize("encoding", ["utf-8", "utf-16"]) +def test_inplace_encoding_matches_stdout(tmpdir, encoding): + source = tmpdir.join("unicode.srt") + text = VALID.decode("ascii").replace("Synthetic subtitle", "caf\u00e9") + source.write_binary(text.encode(encoding)) + args = ["-i", str(source), "--encoding", encoding] + status, expected, stderr = run_tool("normalise", args) + assert status == 0, stderr + status, _, stderr = run_tool("normalise", args + ["--inplace"]) + assert status == 0, stderr + assert source.read_binary() == expected + + +def test_inplace_ignore_parsing_errors(tmpdir): + source = tmpdir.join("recoverable.srt") + source.write_binary(b"Junk\n" + VALID) + args = ["-i", str(source), "--ignore-parsing-errors"] + status, expected, stderr = run_tool("normalise", args) + assert status == 0, stderr + status, _, stderr = run_tool("normalise", args + ["--inplace"]) + assert status == 0, stderr + assert source.read_binary() == expected + + +def test_inplace_rejects_stdin(): + status, _, stderr = run_tool("normalise", ["--inplace"]) + assert status != 0 + assert b"Cannot use --inplace on stdin" in stderr + + +def test_inplace_rejects_explicit_output(tmpdir): + source = tmpdir.join("valid.srt") + destination = tmpdir.join("output.srt") + source.write_binary(VALID) + args = ["-i", str(source), "-o", str(destination), "--inplace"] + status, _, stderr = run_tool("normalise", args) + assert status != 0 + assert b"Cannot use -o and -p together" in stderr + assert source.read_binary() == VALID + assert not destination.exists()