from __future__ import annotations import json from pathlib import Path import subprocess import sys import textwrap import unittest REPO_ROOT = Path(__file__).resolve().parents[2] SRC_ROOT = REPO_ROOT / "src" HEAVY_MODULES = [ "ffx.configuration_controller", "ffx.database", "ffx.ffx_app", "ffx.ffx_controller", "ffx.file_properties", "ffx.tmdb_controller", ] class CliLazyImportTests(unittest.TestCase): def run_python(self, code: str) -> dict: completed = subprocess.run( [sys.executable, "-c", code], capture_output=True, cwd=REPO_ROOT, text=True, ) if completed.returncode != 0: self.fail( "Python helper failed\n" f"STDOUT:\n{completed.stdout}\n" f"STDERR:\n{completed.stderr}" ) return json.loads(completed.stdout) def test_importing_cli_keeps_runtime_modules_unloaded(self): result = self.run_python( textwrap.dedent( f""" import json import sys sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli print(json.dumps({{ "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertTrue( all(not is_loaded for is_loaded in result["modules"].values()), result["modules"], ) def test_lightweight_configure_workstation_command_stays_light(self): result = self.run_python( textwrap.dedent( f""" import json import sys from click.testing import CliRunner sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli runner = CliRunner() invoke_result = runner.invoke( ffx.cli.ffx, ["--dry-run", "configure_workstation", "--check"], ) if invoke_result.exit_code != 0: raise SystemExit(invoke_result.output) print(json.dumps({{ "output": invoke_result.output, "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertIn("configure_workstation.sh --check", result["output"]) self.assertTrue( all(not is_loaded for is_loaded in result["modules"].values()), result["modules"], ) def test_lightweight_setup_command_stays_light(self): result = self.run_python( textwrap.dedent( f""" import json import sys from click.testing import CliRunner sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli runner = CliRunner() invoke_result = runner.invoke( ffx.cli.ffx, ["--dry-run", "setup", "--check", "--with-tests"], ) if invoke_result.exit_code != 0: raise SystemExit(invoke_result.output) print(json.dumps({{ "output": invoke_result.output, "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertIn("tools/setup.sh --check --with-tests", result["output"]) self.assertTrue( all(not is_loaded for is_loaded in result["modules"].values()), result["modules"], ) def test_convert_help_describes_absolute_and_percent_cpu_limits(self): result = self.run_python( textwrap.dedent( f""" import click import json import sys sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli help_output = ffx.cli.convert.get_help(click.Context(ffx.cli.convert)) print(json.dumps({{ "output": help_output, "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertIn("200", result["output"]) self.assertIn("25%", result["output"]) self.assertTrue( all(not is_loaded for is_loaded in result["modules"].values()), result["modules"], ) def test_convert_cut_option_supports_flag_duration_and_start_duration_forms(self): result = self.run_python( textwrap.dedent( f""" import click import json import sys sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli flag_context = ffx.cli.convert.make_context( "convert", ["--cut"], resilient_parsing=True, ) duration_context = ffx.cli.convert.make_context( "convert", ["--cut", "12"], resilient_parsing=True, ) explicit_context = ffx.cli.convert.make_context( "convert", ["--cut=12,34"], resilient_parsing=True, ) disabled_context = ffx.cli.convert.make_context( "convert", [], resilient_parsing=True, ) help_output = ffx.cli.convert.get_help(click.Context(ffx.cli.convert)) print(json.dumps({{ "flag_cut": flag_context.params["cut"], "duration_cut": duration_context.params["cut"], "explicit_cut": explicit_context.params["cut"], "disabled_cut": disabled_context.params["cut"], "output": help_output, "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertEqual([60, 180], result["flag_cut"]) self.assertEqual([0, 12], result["duration_cut"]) self.assertEqual([12, 34], result["explicit_cut"]) self.assertIsNone(result["disabled_cut"]) self.assertIn("--cut DURATION|START,DURATION", result["output"]) self.assertIn("60,180", result["output"]) self.assertIn("START,DURATION", result["output"]) self.assertTrue( all(not is_loaded for is_loaded in result["modules"].values()), result["modules"], ) def test_edit_command_avoids_database_bootstrap(self): result = self.run_python( textwrap.dedent( f""" import json import os import sys import tempfile from click.testing import CliRunner sys.path.insert(0, {str(SRC_ROOT)!r}) import ffx.cli import ffx.ffx_app import ffx.logging_utils ffx.ffx_app.FfxApp.run = lambda self: None ffx.logging_utils.configure_ffx_logger = lambda *args, **kwargs: None runner = CliRunner() with tempfile.TemporaryDirectory() as tmpdir: sample_path = os.path.join(tmpdir, "sample.mkv") with open(sample_path, "w", encoding="utf-8"): pass invoke_result = runner.invoke( ffx.cli.ffx, ["--dry-run", "edit", sample_path], ) print(json.dumps({{ "exit_code": invoke_result.exit_code, "output": invoke_result.output, "modules": {{ module_name: module_name in sys.modules for module_name in {HEAVY_MODULES!r} }}, }})) """ ) ) self.assertEqual(0, result["exit_code"], result["output"]) self.assertFalse(result["modules"]["ffx.database"], result["modules"]) if __name__ == "__main__": unittest.main()