1 files changed, 141 insertions, 0 deletions
diff --git a/desktop/assistant/test_fileops.py b/desktop/assistant/test_fileops.py
new file mode 100755
index 0000000..850d1d7
--- /dev/null
+++ b/desktop/assistant/test_fileops.py
@@ -0,0 +1,141 @@
+#!/usr/bin/python3
+# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 2 as published
+# by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
+# Public License for more details.
+"""Run: /usr/bin/python3 desktop/assistant/test_fileops.py
+
+Builds throwaway inputs in a temp folder and runs every operation but
+trackcrop (it opens a window) into a temp output folder.
+"""
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+FILEOPS = os.path.join(HERE, "fileops")
+
+
+def fileops(op, outdir, *files):
+    p = subprocess.run([sys.executable, FILEOPS, op, outdir, *files],
+                       capture_output=True, text=True, timeout=120)
+    return json.loads(p.stdout.strip().splitlines()[-1]), p.returncode
+
+
+def mime(path):
+    return subprocess.run(["file", "-b", "--mime-type", path], capture_output=True,
+                          text=True).stdout.strip()
+
+
+def sh(*argv):
+    subprocess.run(argv, check=True, capture_output=True)
+
+
+with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as out:
+    pdf_a, pdf_b = os.path.join(src, "a.pdf"), os.path.join(src, "b.pdf")
+    png = os.path.join(src, "pic.png")
+    # This ffmpeg has no libx264 and no AAC encoder: inputs are VP9 + Opus
+    # in WebM and MPEG-4 + PCM in Matroska.
+    clip = os.path.join(src, "clip.webm")
+    raw = os.path.join(src, "raw.mkv")
+    tarball = os.path.join(src, "bundle.tar")
+    sh("magick", "-size", "200x100", "xc:white", pdf_a)
+    sh("magick", "-size", "200x100", "xc:gray", pdf_b)
+    sh("magick", "-size", "64x32", "xc:red", png)
+    sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=1:size=160x120:rate=10",
+       "-f", "lavfi", "-i", "sine=duration=1", "-c:v", "libvpx-vp9", "-c:a", "libopus", "-shortest", clip)
+    sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=1:size=160x120:rate=10",
+       "-f", "lavfi", "-i", "sine=duration=1", "-c:v", "mpeg4", "-c:a", "pcm_s16le", "-shortest", raw)
+    with open(os.path.join(src, "note.txt"), "w") as f:
+        f.write("hello\n")
+    with tarfile.open(tarball, "w") as t:
+        t.add(os.path.join(src, "note.txt"), arcname="note.txt")
+
+    def one(op, *files):
+        r, rc = fileops(op, out, *files)
+        assert r["ok"] and rc == 0, (op, r)
+        return r["out"]
+
+    # Each result, by the type file(1) sees in it.
+    [m] = one("pdf-merge", pdf_a, pdf_b)
+    assert mime(m) == "application/pdf" and os.path.basename(m) == "merged.pdf"
+    assert subprocess.run(["qpdf", "--show-npages", m], capture_output=True,
+                          text=True).stdout.strip() == "2"
+    [d] = one("pdf-images", pdf_a)
+    assert os.path.basename(d) == "a (pages)" and [mime(os.path.join(d, x)) for x in os.listdir(d)] == ["image/png"]
+    [a] = one("audio", clip)                       # Opus: copied as it is
+    assert a.endswith("clip.opus") and mime(a) == "audio/ogg", mime(a)
+    [a2] = one("audio", raw)                       # PCM: encoded to MP3
+    assert a2.endswith("raw.mp3") and mime(a2) == "audio/mpeg", mime(a2)
+    [w] = one("webm", raw)
+    assert mime(w) == "video/webm" and w.endswith("raw.webm")
+    [v] = one("mp4", raw)                          # h264_vaapi + MP3
+    assert mime(v) == "video/mp4" and v.endswith("raw.mp4")
+    [g] = one("gif", clip)
+    assert mime(g) == "image/gif"
+    [j] = one("jpeg", png)
+    assert mime(j) == "image/jpeg"
+    [p] = one("png", j)
+    assert mime(p) == "image/png" and os.path.basename(p) == "pic.png"
+    [wp] = one("webp", png)
+    assert mime(wp) == "image/webp"
+    [h] = one("half", png)
+    assert os.path.basename(h) == "pic (50%).png"
+    assert subprocess.run(["magick", "identify", "-format", "%wx%h", h], capture_output=True,
+                          text=True).stdout == "32x16"
+    [x] = one("extract", tarball)
+    assert os.path.basename(x) == "bundle" and os.listdir(x) == ["note.txt"]
+    for op, want, ext in (("zip", "application/zip", ".zip"), ("tgz", "application/gzip", ".tar.gz"),
+                          ("txz", "application/x-xz", ".tar.xz"), ("tbz2", "application/x-bzip2", ".tar.bz2")):
+        [z] = one(op, png, pdf_a)
+        assert mime(z) == want and os.path.basename(z) == "archive" + ext, (op, z)
+    [z1] = one("zip", png)                         # one file: named after it
+    assert os.path.basename(z1) == "pic.zip"
+
+    # One result per file, and taken names get a number.
+    rs = one("png", png, png)
+    assert [os.path.basename(r) for r in rs] == ["pic (1).png", "pic (2).png"], rs
+
+    # A failure leaves nothing behind.
+    before = set(os.listdir(out))
+    fake = os.path.join(src, "fake.mp4")
+    with open(fake, "w") as f:
+        f.write("not a video\n")
+    r, rc = fileops("gif", out, fake)
+    assert not r["ok"] and rc == 1 and r["error"], r
+    assert set(os.listdir(out)) == before, set(os.listdir(out)) - before
+    for bad in (("nope", out, png), ("png", "relative/dir", png), ("png", out, src + "/missing.png")):
+        r, rc = fileops(*bad)
+        assert not r["ok"] and rc == 1, bad
+    r, _ = fileops("audio", out, g)
+    assert not r["ok"] and "audio" in r["error"], r   # a GIF has no audio track
+
+    # SIGTERM mid-encode (the canvas's Stop): no ffmpeg left, no partial file.
+    long_clip = os.path.join(src, "long.mp4")
+    sh("ffmpeg", "-v", "error", "-f", "lavfi", "-i", "testsrc=duration=20:size=1280x720:rate=30",
+       "-c:v", "mpeg4", long_clip)
+    with tempfile.TemporaryDirectory() as slow:
+        proc = subprocess.Popen([sys.executable, FILEOPS, "webm", slow, long_clip],
+                                stdout=subprocess.PIPE, text=True)
+        time.sleep(1.5)
+        proc.send_signal(signal.SIGTERM)
+        r = json.loads(proc.communicate(timeout=10)[0].strip().splitlines()[-1])
+        assert not r["ok"] and r["error"] == "stopped", r
+        time.sleep(0.5)
+        left = subprocess.run(["pgrep", "-f", slow], capture_output=True, text=True).stdout
+        assert left == "", left
+        assert os.listdir(slow) == [], os.listdir(slow)
+
+print("ok")