ffx/tests/unit/test_cli_lazy_imports.py

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"],
        )


if __name__ == "__main__":
    unittest.main()