From 9a2a0c93cc30af979a26d957fef603e2378676be Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Fri, 16 Oct 2026 16:38:35 +0200 Subject: feat(assistant): fileops, the Do row's local operations One CLI for converting, extracting, merging and compressing staged files, answering with one JSON line. Results are written under a temporary name and renamed to the first free one, so a failure or a stop leaves nothing half written, and a stop kills the tool's whole process group. This ffmpeg has no libx264 and no AAC encoder, so MP4 is H.264 through VAAPI with MP3 audio, and audio extraction copies AAC, MP3 and Opus tracks as they are. The test builds its inputs and checks every result's type. Co-Authored-By: Claude Opus 5.5 --- desktop/assistant/test_fileops.py | 141 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100755 desktop/assistant/test_fileops.py (limited to 'desktop/assistant/test_fileops.py') 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. +# +# 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") -- cgit v1.2.3