From 8a375ccce161b7f470df2515a8873aeecec1f1fd Mon Sep 17 00:00:00 2001 From: Javanaut Date: Fri, 19 Jun 2026 08:19:43 +0200 Subject: [PATCH] Unmux output dir generation --- src/ffx/cli.py | 39 +++++++++- tests/integration/test_cli_unmux.py | 18 ++++- tests/unit/test_cli_unmux_output_directory.py | 77 ++++++++++++++++++- 3 files changed, 126 insertions(+), 8 deletions(-) diff --git a/src/ffx/cli.py b/src/ffx/cli.py index 2327462..8317403 100755 --- a/src/ffx/cli.py +++ b/src/ffx/cli.py @@ -285,7 +285,10 @@ def resolveUnmuxOutputDirectory(context, outputDirectory, subtitlesOnly, label): ) resolvedLabel = str(label).strip() - if resolvedOutputDirectory or not subtitlesOnly or not resolvedLabel: + if resolvedOutputDirectory: + return resolvedOutputDirectory, True + + if not subtitlesOnly or not resolvedLabel: return resolvedOutputDirectory, False configuredSubtitlesBaseDirectory = context['config'].getSubtitlesDirectoryPath() @@ -298,6 +301,34 @@ def resolveUnmuxOutputDirectory(context, outputDirectory, subtitlesOnly, label): return os.path.join(configuredSubtitlesBaseDirectory, resolvedLabel), True +def ensureUnmuxOutputDirectory(context, outputDirectory): + resolvedOutputDirectory = os.path.expanduser(str(outputDirectory).strip()) + if not resolvedOutputDirectory: + return False + + if os.path.isdir(resolvedOutputDirectory): + return False + + if os.path.exists(resolvedOutputDirectory): + raise click.ClickException( + "Unmux output path exists but is not a directory: " + + resolvedOutputDirectory + ) + + if context.get('dry_run', False): + return False + + if not click.confirm( + "Create unmux output directory and missing parents: " + + resolvedOutputDirectory, + default=True, + ): + raise click.ClickException("Unmux output directory creation aborted by user.") + + os.makedirs(resolvedOutputDirectory, exist_ok=True) + return True + + def resolveIndicatorDigitLengths(context=None, showDescriptor=None): from ffx.show_descriptor import ShowDescriptor @@ -857,14 +888,14 @@ def unmux(ctx, ctx.obj['resource_limits']['cpu_limit'] = cpu ctx.obj['resource_limits']['cpu_percent'] = cpu - output_directory, create_output_directory = resolveUnmuxOutputDirectory( + output_directory, requires_output_directory = resolveUnmuxOutputDirectory( ctx.obj, output_directory, subtitles_only, label, ) - if create_output_directory and existingSourcePaths and not ctx.obj.get('dry_run', False): - os.makedirs(output_directory, exist_ok=True) + if requires_output_directory and existingSourcePaths: + ensureUnmuxOutputDirectory(ctx.obj, output_directory) shiftedSeasonController = ShiftedSeasonController(ctx.obj) diff --git a/tests/integration/test_cli_unmux.py b/tests/integration/test_cli_unmux.py index a09728f..5717ee3 100644 --- a/tests/integration/test_cli_unmux.py +++ b/tests/integration/test_cli_unmux.py @@ -35,7 +35,13 @@ if pytest is not None: SRC_ROOT = Path(__file__).resolve().parents[2] / "src" -def run_ffx_unmux(workdir: Path, home_dir: Path, database_path: Path, *args: str) -> subprocess.CompletedProcess[str]: +def run_ffx_unmux( + workdir: Path, + home_dir: Path, + database_path: Path, + *args: str, + input_text: str | None = None, +) -> subprocess.CompletedProcess[str]: env = os.environ.copy() env["HOME"] = str(home_dir) existing_pythonpath = env.get("PYTHONPATH", "") @@ -50,7 +56,14 @@ def run_ffx_unmux(workdir: Path, home_dir: Path, database_path: Path, *args: str "unmux", *args, ] - return subprocess.run(command, cwd=workdir, env=env, capture_output=True, text=True) + return subprocess.run( + command, + cwd=workdir, + env=env, + capture_output=True, + input=input_text, + text=True, + ) class UnmuxCliTests(unittest.TestCase): @@ -164,6 +177,7 @@ class UnmuxCliTests(unittest.TestCase): "--label", "dball", str(source_path), + input_text="y\n", ) self.assertCompleted(completed) diff --git a/tests/unit/test_cli_unmux_output_directory.py b/tests/unit/test_cli_unmux_output_directory.py index f417fc6..525714f 100644 --- a/tests/unit/test_cli_unmux_output_directory.py +++ b/tests/unit/test_cli_unmux_output_directory.py @@ -4,6 +4,7 @@ from pathlib import Path import sys import tempfile import unittest +from unittest.mock import patch import click @@ -42,7 +43,7 @@ class UnmuxOutputDirectoryTests(unittest.TestCase): self.assertEqual(str(Path(tempdir) / "subtitles" / "dball"), resolved_output_directory) self.assertTrue(should_create) - def test_explicit_output_directory_keeps_existing_behavior(self): + def test_explicit_output_directory_requires_directory(self): with tempfile.TemporaryDirectory() as tempdir: context = { "config": StaticConfig(str(Path(tempdir) / "subtitles")), @@ -57,7 +58,7 @@ class UnmuxOutputDirectoryTests(unittest.TestCase): ) self.assertEqual(explicit_output_directory, resolved_output_directory) - self.assertFalse(should_create) + self.assertTrue(should_create) def test_subtitles_only_without_label_keeps_existing_behavior(self): context = { @@ -89,6 +90,78 @@ class UnmuxOutputDirectoryTests(unittest.TestCase): self.assertIn("subtitlesDirectory default", str(caught.exception)) + def test_missing_output_directory_can_be_confirmed_and_created_with_parents(self): + with tempfile.TemporaryDirectory() as tempdir: + output_directory = Path(tempdir) / "missing" / "parents" / "manual" + + with patch("ffx.cli.click.confirm", return_value=True) as mocked_confirm: + created = cli.ensureUnmuxOutputDirectory( + {"dry_run": False}, + str(output_directory), + ) + + self.assertTrue(created) + self.assertTrue(output_directory.is_dir()) + mocked_confirm.assert_called_once() + + def test_missing_output_directory_can_be_rejected(self): + with tempfile.TemporaryDirectory() as tempdir: + output_directory = Path(tempdir) / "missing" / "manual" + + with patch("ffx.cli.click.confirm", return_value=False) as mocked_confirm: + with self.assertRaises(click.ClickException) as caught: + cli.ensureUnmuxOutputDirectory( + {"dry_run": False}, + str(output_directory), + ) + + self.assertFalse(output_directory.exists()) + self.assertIn("aborted by user", str(caught.exception)) + mocked_confirm.assert_called_once() + + def test_existing_output_directory_does_not_prompt(self): + with tempfile.TemporaryDirectory() as tempdir: + output_directory = Path(tempdir) / "manual" + output_directory.mkdir() + + with patch("ffx.cli.click.confirm") as mocked_confirm: + created = cli.ensureUnmuxOutputDirectory( + {"dry_run": False}, + str(output_directory), + ) + + self.assertFalse(created) + mocked_confirm.assert_not_called() + + def test_existing_non_directory_output_path_fails_without_prompt(self): + with tempfile.TemporaryDirectory() as tempdir: + output_path = Path(tempdir) / "manual" + output_path.write_text("not a directory", encoding="utf-8") + + with patch("ffx.cli.click.confirm") as mocked_confirm: + with self.assertRaises(click.ClickException) as caught: + cli.ensureUnmuxOutputDirectory( + {"dry_run": False}, + str(output_path), + ) + + self.assertIn("not a directory", str(caught.exception)) + mocked_confirm.assert_not_called() + + def test_dry_run_does_not_prompt_or_create_missing_output_directory(self): + with tempfile.TemporaryDirectory() as tempdir: + output_directory = Path(tempdir) / "missing" / "manual" + + with patch("ffx.cli.click.confirm") as mocked_confirm: + created = cli.ensureUnmuxOutputDirectory( + {"dry_run": True}, + str(output_directory), + ) + + self.assertFalse(created) + self.assertFalse(output_directory.exists()) + mocked_confirm.assert_not_called() + if __name__ == "__main__": unittest.main()