355 lines
11 KiB
Python
355 lines
11 KiB
Python
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_root_debug_flag_parses_without_loading_runtime_modules(self):
|
|
result = self.run_python(
|
|
textwrap.dedent(
|
|
f"""
|
|
import json
|
|
import sys
|
|
|
|
sys.path.insert(0, {str(SRC_ROOT)!r})
|
|
|
|
import ffx.cli
|
|
|
|
context = ffx.cli.ffx.make_context(
|
|
"ffx",
|
|
["--debug", "help"],
|
|
resilient_parsing=True,
|
|
)
|
|
|
|
print(json.dumps({{
|
|
"debug": context.params["debug"],
|
|
"modules": {{
|
|
module_name: module_name in sys.modules
|
|
for module_name in {HEAVY_MODULES!r}
|
|
}},
|
|
}}))
|
|
"""
|
|
)
|
|
)
|
|
|
|
self.assertTrue(result["debug"])
|
|
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_convert_copy_flags_parse_without_loading_runtime_modules(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
|
|
|
|
context = ffx.cli.convert.make_context(
|
|
"convert",
|
|
["--copy-video", "--copy-audio"],
|
|
resilient_parsing=True,
|
|
)
|
|
help_output = ffx.cli.convert.get_help(click.Context(ffx.cli.convert))
|
|
|
|
print(json.dumps({{
|
|
"copy_video": context.params["copy_video"],
|
|
"copy_audio": context.params["copy_audio"],
|
|
"output": help_output,
|
|
"modules": {{
|
|
module_name: module_name in sys.modules
|
|
for module_name in {HEAVY_MODULES!r}
|
|
}},
|
|
}}))
|
|
"""
|
|
)
|
|
)
|
|
|
|
self.assertTrue(result["copy_video"])
|
|
self.assertTrue(result["copy_audio"])
|
|
self.assertIn("--copy-video", result["output"])
|
|
self.assertIn("--copy-audio", 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()
|