mirror of
https://github.com/affaan-m/ECC.git
synced 2026-09-29 13:05:18 +02:00
928c1dea72
* feat: bundle standalone taste distillation and application workflows * docs: fix imported taste skill markdown lint * docs: align Turkish agent catalog with taste skills * refactor: make ECC the canonical reusable video engine * fix: preserve video duration when applying image overlays * fix: preserve background colors in image compositing * fix: report best-effort duration targets and shortfalls * feat: ship verified Fusion presets with compatibility provenance * feat(tasteforge): preserve native edits in application bundles * feat(tasteforge): compile local preservation without hosted input * fix: update js-yaml to patched 4.3.2 * test: report bounded Stop wrapper failure diagnostics * fix(tasteforge): fail closed on unsafe output names, missing overlays and cadence - cli: default report and spec paths are derived from pack name and profile genre; require the manifest's name pattern before using either as a filename part so a traversal string cannot write outside cwd/out. - apply_local: a pack without cadence.json, or with no measured shots and no explicit mean_shot, raises instead of silently planning 1.0s shots and reporting a measured cadence. - legacy apply: a missing overlay aborts before any paid upload; forge() would have rejected it after every take was generated. - requirements-live: pin fal-client>=0.13.0, the first release whose subscribe() accepts client_timeout. Addresses the five P1 findings from the independent review of #3033. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015fxHRsydPqEcYngGbqkgt1 --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
419 lines
16 KiB
Python
419 lines
16 KiB
Python
import io
|
|
import json
|
|
import pathlib
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, call, patch
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).parents[1] / "scripts"))
|
|
from tasteforge.media import capcut, stills
|
|
|
|
|
|
class MediaUtilitiesTests(unittest.TestCase):
|
|
def test_reject_bad_geometry_before_execution(self):
|
|
runner = Mock()
|
|
for kwargs in ({"width": 0}, {"fps": float("nan")}, {"duration": -1}):
|
|
with self.assertRaises(ValueError):
|
|
stills.make_clip("missing.png", "out.mp4", runner=runner, **kwargs)
|
|
runner.assert_not_called()
|
|
|
|
def test_still_failure_propagates_and_preserves_destination(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "in.png"
|
|
source.touch()
|
|
target = pathlib.Path(directory) / "out.mp4"
|
|
target.write_bytes(b"old")
|
|
runner = Mock(side_effect=subprocess.CalledProcessError(1, "ffmpeg"))
|
|
with self.assertRaises(FileExistsError):
|
|
stills.make_clip(source, target, runner=runner)
|
|
runner.assert_not_called()
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
stills.make_clip(source, target, overwrite=True, runner=runner)
|
|
self.assertEqual(target.read_bytes(), b"old")
|
|
|
|
def test_crop_changes_filter(self):
|
|
first = stills.filter_graph(
|
|
(0, 0, 1, 1), (0.1, 0.1, 0.8, 0.8), 60, 320, 180, 30
|
|
)
|
|
second = stills.filter_graph(
|
|
(0.2, 0.1, 0.5, 0.6), (0.1, 0.1, 0.8, 0.8), 60, 320, 180, 30
|
|
)
|
|
self.assertNotEqual(first, second)
|
|
|
|
def test_capcut_preflight_before_draft_creation(self):
|
|
cc = Mock()
|
|
with self.assertRaises(ValueError):
|
|
capcut.export_draft([], "/tmp/drafts", "../escape", cc=cc)
|
|
cc.DraftFolder.assert_not_called()
|
|
|
|
def test_capcut_overwrite_refused_before_app_mutation(self):
|
|
cc = Mock()
|
|
with self.assertRaises(ValueError):
|
|
capcut.export_draft(
|
|
["clip.mp4"], "/tmp/drafts", "draft", overwrite=True, cc=cc
|
|
)
|
|
cc.DraftFolder.assert_not_called()
|
|
|
|
def test_crop_height_changes_aspect_fit(self):
|
|
first = stills.filter_graph(
|
|
(0.25, 0.25, 0.5, 0.5), (0, 0, 1, 1), 60, 320, 180, 30
|
|
)
|
|
second = stills.filter_graph(
|
|
(0.25, 0.125, 0.5, 0.75), (0, 0, 1, 1), 60, 320, 180, 30
|
|
)
|
|
self.assertNotEqual(first, second)
|
|
|
|
def test_capcut_duration_failure_prevents_creation(self):
|
|
cc = Mock()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "clip.mp4"
|
|
source.touch()
|
|
with self.assertRaises(ValueError):
|
|
capcut.export_draft(
|
|
[source], directory, "draft", cc=cc, probe=lambda _: float("nan")
|
|
)
|
|
cc.DraftFolder.assert_not_called()
|
|
|
|
@unittest.skipUnless(
|
|
shutil.which("ffmpeg") and shutil.which("ffprobe"), "FFmpeg required"
|
|
)
|
|
def test_real_ffmpeg_still_and_feedback(self):
|
|
try:
|
|
import PIL # noqa: F401
|
|
from tasteforge.media.glitch import render
|
|
except ImportError:
|
|
self.skipTest("numpy and Pillow required for effect smoke")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = pathlib.Path(directory)
|
|
image = root / "source.ppm"
|
|
image.write_bytes(b"P6\n32 18\n255\n" + bytes([100, 20, 200]) * 32 * 18)
|
|
clip = root / "still.mp4"
|
|
stills.make_clip(image, clip, duration=0.5, width=32, height=18, fps=10)
|
|
final = root / "feedback.mp4"
|
|
self.assertEqual(render(clip, 0, 0.5, final, 42, "feedback", 32, 18, 10), 5)
|
|
result = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v",
|
|
"error",
|
|
"-count_frames",
|
|
"-show_entries",
|
|
"stream=width,height,nb_read_frames,r_frame_rate",
|
|
"-of",
|
|
"json",
|
|
str(final),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
stream = json.loads(result.stdout)["streams"][0]
|
|
self.assertEqual(
|
|
(stream["width"], stream["height"], stream["nb_read_frames"]),
|
|
(32, 18, "5"),
|
|
)
|
|
self.assertEqual(stream["r_frame_rate"], "10/1")
|
|
|
|
def test_empty_decoder_fails(self):
|
|
try:
|
|
from tasteforge.media.glitch import transform_stream
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
with self.assertRaises(RuntimeError):
|
|
transform_stream(
|
|
Mock(stdout=io.BytesIO()),
|
|
Mock(stdin=io.BytesIO()),
|
|
1,
|
|
32,
|
|
18,
|
|
30,
|
|
42,
|
|
"drift",
|
|
)
|
|
|
|
def test_seeded_effect_is_repeatable(self):
|
|
try:
|
|
from tasteforge.media.glitch import transform_stream
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
frame = bytes(range(256)) * (32 * 18 * 3 // 256) + bytes(range(192))
|
|
results = []
|
|
for _ in range(2):
|
|
destination = io.BytesIO()
|
|
transform_stream(
|
|
Mock(stdout=io.BytesIO(frame)),
|
|
Mock(stdin=destination),
|
|
1,
|
|
32,
|
|
18,
|
|
30,
|
|
42,
|
|
"drift",
|
|
)
|
|
results.append(destination.getvalue())
|
|
self.assertEqual(results[0], results[1])
|
|
|
|
def test_capcut_success_preserves_order_and_frame_rate(self):
|
|
cc = Mock()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
sources = [
|
|
pathlib.Path(directory) / name for name in ("first.mp4", "second.mp4")
|
|
]
|
|
for source in sources:
|
|
source.touch()
|
|
receipt = capcut.export_draft(
|
|
sources,
|
|
directory,
|
|
"Review 2",
|
|
width=320,
|
|
height=180,
|
|
fps=24,
|
|
cc=cc,
|
|
probe=Mock(side_effect=[1.25, 2.5]),
|
|
)
|
|
cc.DraftFolder.return_value.create_draft.assert_called_once_with(
|
|
"Review 2", 320, 180, fps=24, allow_replace=False
|
|
)
|
|
self.assertEqual(
|
|
cc.trange.call_args_list,
|
|
[call("0.000000s", "1.250000s"), call("1.250000s", "2.500000s")],
|
|
)
|
|
self.assertEqual(
|
|
[entry.args[0] for entry in cc.VideoSegment.call_args_list],
|
|
[str(source.resolve()) for source in sources],
|
|
)
|
|
self.assertEqual(receipt["duration"], 3.75)
|
|
self.assertEqual(receipt["segments"], 2)
|
|
cc.DraftFolder.return_value.create_draft.return_value.save.assert_called_once()
|
|
|
|
def test_capcut_save_failure_is_not_reported_as_success(self):
|
|
cc = Mock()
|
|
cc.DraftFolder.return_value.create_draft.return_value.save.side_effect = (
|
|
OSError("disk full")
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "clip.mp4"
|
|
source.touch()
|
|
with self.assertRaisesRegex(OSError, "disk full"):
|
|
capcut.export_draft(
|
|
[source], directory, "new-draft", cc=cc, probe=lambda _: 1
|
|
)
|
|
|
|
def test_capcut_empty_list_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
capcut.export_draft([], "/tmp/drafts", "valid-name", cc=Mock())
|
|
|
|
def test_concat_paths_resolve_against_list_not_current_directory(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
concat = pathlib.Path(directory) / "concat.txt"
|
|
concat.write_text("# heading\nfile 'with space.mp4'\n\nfile second.mp4\n")
|
|
self.assertEqual(
|
|
capcut.read_concat(concat),
|
|
[
|
|
(pathlib.Path(directory) / "with space.mp4").resolve(),
|
|
(pathlib.Path(directory) / "second.mp4").resolve(),
|
|
],
|
|
)
|
|
concat.write_text("file first.mp4 unexpected\n")
|
|
with self.assertRaises(ValueError):
|
|
capcut.read_concat(concat)
|
|
|
|
def test_probe_failure_and_valid_duration(self):
|
|
with patch.object(
|
|
capcut.subprocess, "run", return_value=Mock(stdout="1.125\n")
|
|
) as runner:
|
|
self.assertEqual(capcut.duration_of("source.mp4"), 1.125)
|
|
self.assertTrue(runner.call_args.kwargs["check"])
|
|
with patch.object(
|
|
capcut.subprocess,
|
|
"run",
|
|
side_effect=subprocess.CalledProcessError(1, "ffprobe"),
|
|
):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
capcut.duration_of("source.mp4")
|
|
|
|
def test_media_cli_parameter_forwarding(self):
|
|
with patch.object(stills, "make_clip") as renderer:
|
|
stills.main(
|
|
[
|
|
"source.png",
|
|
"final.mp4",
|
|
"--duration",
|
|
".5",
|
|
"--width",
|
|
"320",
|
|
"--height",
|
|
"180",
|
|
"--fps",
|
|
"24",
|
|
"--overwrite",
|
|
]
|
|
)
|
|
self.assertEqual(renderer.call_args.kwargs["fps"], 24)
|
|
self.assertTrue(renderer.call_args.kwargs["overwrite"])
|
|
with (
|
|
patch.object(capcut, "read_concat", return_value=["source.mp4"]),
|
|
patch.object(capcut, "export_draft") as exporter,
|
|
):
|
|
capcut.main(
|
|
[
|
|
"list.txt",
|
|
"--drafts",
|
|
"/tmp/drafts",
|
|
"--name",
|
|
"review",
|
|
"--fps",
|
|
"24",
|
|
]
|
|
)
|
|
self.assertEqual(exporter.call_args.kwargs["files"], ["source.mp4"])
|
|
self.assertEqual(exporter.call_args.kwargs["name"], "review")
|
|
self.assertEqual(exporter.call_args.kwargs["fps"], 24)
|
|
|
|
def test_crop_bounds_missing_source_and_odd_geometry(self):
|
|
for crop in ((0, 0, 1), (0, 0, float("nan"), 1), (0.5, 0, 1, 1)):
|
|
with self.assertRaises(ValueError):
|
|
stills.filter_graph(crop, (0, 0, 1, 1), 10, 320, 180, 30)
|
|
with self.assertRaises(ValueError):
|
|
stills.make_clip("missing.png", "out.mp4", width=319)
|
|
with self.assertRaises(FileNotFoundError):
|
|
stills.make_clip("missing.png", "out.mp4")
|
|
|
|
def test_output_empty_rejected_and_successful_replace_atomic(self):
|
|
from tasteforge.media.common import output_file
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
target = pathlib.Path(directory) / "out.mp4"
|
|
target.write_bytes(b"previous")
|
|
with self.assertRaises(RuntimeError):
|
|
with output_file(target, overwrite=True):
|
|
pass
|
|
self.assertEqual(target.read_bytes(), b"previous")
|
|
with output_file(target, overwrite=True) as temporary:
|
|
temporary.write_bytes(b"complete")
|
|
self.assertEqual(target.read_bytes(), b"previous")
|
|
self.assertEqual(target.read_bytes(), b"complete")
|
|
self.assertEqual(list(pathlib.Path(directory).glob(".media-*")), [])
|
|
|
|
def test_glitch_validation_and_cli(self):
|
|
try:
|
|
from tasteforge.media import glitch
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "source.mp4"
|
|
source.touch()
|
|
for options in ({"start": -1}, {"mode": "unknown"}, {"seed": -1}):
|
|
params = dict(
|
|
source=source, start=0, duration=1, output="out.mp4", seed=42
|
|
)
|
|
params.update(options)
|
|
with self.assertRaises(ValueError):
|
|
glitch.render(**params)
|
|
with self.assertRaises(FileNotFoundError):
|
|
glitch.render(source / "missing", 0, 1, "out.mp4", 42)
|
|
with (
|
|
patch.object(glitch, "render", return_value=12) as renderer,
|
|
patch("builtins.print"),
|
|
):
|
|
glitch.main(
|
|
["source.mp4", "1", ".5", "out.mp4", "42", "mosh", "--fps", "24"]
|
|
)
|
|
self.assertEqual(renderer.call_args.kwargs["mode"], "mosh")
|
|
self.assertEqual(renderer.call_args.kwargs["fps"], 24)
|
|
|
|
def test_glitch_truncated_frame_and_mosh_are_explicit(self):
|
|
try:
|
|
from tasteforge.media.glitch import transform_stream
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
with self.assertRaisesRegex(RuntimeError, "truncated"):
|
|
transform_stream(
|
|
Mock(stdout=io.BytesIO(b"truncated")),
|
|
Mock(stdin=io.BytesIO()),
|
|
1,
|
|
32,
|
|
18,
|
|
30,
|
|
42,
|
|
"drift",
|
|
)
|
|
frame = bytes(range(256)) * 6 + bytes(range(192))
|
|
destination = io.BytesIO()
|
|
self.assertEqual(
|
|
transform_stream(
|
|
Mock(stdout=io.BytesIO(frame * 3)),
|
|
Mock(stdin=destination),
|
|
0.1,
|
|
32,
|
|
18,
|
|
30,
|
|
42,
|
|
"mosh",
|
|
),
|
|
3,
|
|
)
|
|
self.assertEqual(len(destination.getvalue()), len(frame) * 3)
|
|
self.assertNotEqual(destination.getvalue(), frame * 3)
|
|
|
|
def test_encoder_launch_failure_reaps_decoder(self):
|
|
try:
|
|
from tasteforge.media import glitch
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
decoder = Mock(stdin=None, stdout=io.BytesIO(), poll=Mock(return_value=None))
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "source.mp4"
|
|
source.touch()
|
|
target = pathlib.Path(directory) / "out.mp4"
|
|
with patch.object(
|
|
glitch.subprocess,
|
|
"Popen",
|
|
side_effect=[decoder, OSError("encoder unavailable")],
|
|
):
|
|
with self.assertRaisesRegex(OSError, "encoder unavailable"):
|
|
glitch.render(source, 0, 1, target, 42)
|
|
decoder.kill.assert_called_once()
|
|
decoder.wait.assert_called_once()
|
|
self.assertTrue(decoder.stdout.closed)
|
|
self.assertFalse(target.exists())
|
|
|
|
def test_still_cannot_replace_source_through_same_path_or_symlink(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "source.png"
|
|
source.write_bytes(b"original")
|
|
alias = pathlib.Path(directory) / "alias.png"
|
|
alias.symlink_to(source)
|
|
renderer = Mock()
|
|
for output in (source, alias):
|
|
with self.assertRaises(ValueError):
|
|
stills.make_clip(source, output, overwrite=True, runner=renderer)
|
|
renderer.assert_not_called()
|
|
self.assertEqual(source.read_bytes(), b"original")
|
|
self.assertTrue(alias.is_symlink())
|
|
|
|
def test_glitch_cannot_replace_source_through_same_path_or_symlink(self):
|
|
try:
|
|
from tasteforge.media import glitch
|
|
except ImportError:
|
|
self.skipTest("numpy required")
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = pathlib.Path(directory) / "source.mp4"
|
|
source.write_bytes(b"original")
|
|
alias = pathlib.Path(directory) / "alias.mp4"
|
|
alias.symlink_to(source)
|
|
with patch.object(glitch.subprocess, "Popen") as process:
|
|
for output in (source, alias):
|
|
with self.assertRaises(ValueError):
|
|
glitch.render(source, 0, 1, output, 42, overwrite=True)
|
|
process.assert_not_called()
|
|
self.assertEqual(source.read_bytes(), b"original")
|
|
self.assertTrue(alias.is_symlink())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|