# SPDX-License-Identifier: Apache-2.0 """ CLI tests for oMLX. Tests CLI argument parsing, command setup, and help text. Note: Configuration validation tests are in test_config.py. """ import argparse import socket import subprocess import sys from types import ModuleType, SimpleNamespace from unittest.mock import MagicMock, patch import pytest from omlx._version import __version__ class TestCLIModule: """Tests for CLI module existence and basic functionality.""" def test_cli_module_importable(self): """Test that CLI module can be imported.""" from omlx import cli assert hasattr(cli, "main") def test_cli_has_serve_command(self): """Test that CLI has serve command setup.""" from omlx import cli # The module should have the main entry point assert callable(cli.main) class TestCLIHelp: """Tests for CLI help functionality.""" def test_main_help(self): """Test main CLI help output.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "--help"], capture_output=True, text=True, timeout=10, ) # Should succeed with help assert result.returncode == 0 # Should show available commands assert "serve" in result.stdout.lower() def test_main_version(self): """Test main CLI version output.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "--version"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 assert result.stdout.strip() == __version__ assert result.stderr == "" def test_serve_help(self): """Test serve command help output.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) # Should succeed with help assert result.returncode == 0 # Should show serve options stdout_lower = result.stdout.lower() assert "host" in stdout_lower assert "port" in stdout_lower assert "model-dir" in stdout_lower def test_lifecycle_commands_in_main_help(self): """Test start/stop/restart lifecycle commands are exposed.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 stdout_lower = result.stdout.lower() assert "start" in stdout_lower assert "stop" in stdout_lower assert "restart" in stdout_lower class TestLifecycleCommand: """Tests for managed background server lifecycle commands.""" @staticmethod def _args(command, **overrides): values = {"command": command, "timeout": 1.0, "no_wait": False} values.update(overrides) return SimpleNamespace(**values) def test_app_bundle_stop_without_running_app_is_success(self, monkeypatch, capsys): """`omlx stop` must not launch the macOS app just to stop it.""" from omlx import cli import omlx.utils.install as install monkeypatch.setattr(install, "is_app_bundle", lambda: True) monkeypatch.setattr(install, "is_homebrew", lambda: False) monkeypatch.setattr(cli, "_send_app_control", MagicMock(side_effect=OSError)) monkeypatch.setattr(cli, "_open_macos_app", MagicMock()) assert cli.lifecycle_command(self._args("stop")) == 0 assert capsys.readouterr().out.strip() == "oMLX stopped" cli._open_macos_app.assert_not_called() def test_app_bundle_start_sends_command_and_waits(self, monkeypatch): """`omlx start` asks the app to start and waits for a running state.""" from omlx import cli import omlx.utils.install as install monkeypatch.setattr(install, "is_app_bundle", lambda: True) monkeypatch.setattr(install, "is_homebrew", lambda: False) send = MagicMock(return_value={"ok": True, "state": "starting", "port": 8000}) wait = MagicMock(return_value={"ok": True, "state": "running", "port": 8000}) monkeypatch.setattr(cli, "_send_app_control_with_launch", send) monkeypatch.setattr(cli, "_wait_app_control_state", wait) assert cli.lifecycle_command(self._args("start")) == 0 send.assert_called_once_with("start", timeout=1.0) wait.assert_called_once_with({"running", "unresponsive"}, 1.0) def test_homebrew_start_delegates_to_brew_services(self, monkeypatch): """Homebrew installs use the Homebrew service supervisor.""" from omlx import cli import omlx.utils.install as install monkeypatch.setattr(install, "is_app_bundle", lambda: False) monkeypatch.setattr(install, "is_homebrew", lambda: True) run_brew = MagicMock(return_value=0) monkeypatch.setattr(cli, "_run_brew_services", run_brew) assert cli.lifecycle_command(self._args("start")) == 0 run_brew.assert_called_once_with("start") class TestCLIEntryPoint: """Tests for CLI entry point functionality.""" def test_module_runnable(self): """Test that CLI module is runnable.""" # Should not crash when running with --help result = subprocess.run( [sys.executable, "-m", "omlx.cli", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 def test_invalid_command_error(self): """Test error handling for invalid command.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "invalid_command"], capture_output=True, text=True, timeout=10, ) # Should fail with non-zero exit code assert result.returncode != 0 def test_no_command_shows_help(self): """Test that no command shows help.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli"], capture_output=True, text=True, timeout=10, ) # Should exit with non-zero (no command provided) assert result.returncode != 0 class TestServeCommandOptions: """Tests for serve command options via help output.""" def test_serve_has_model_dir_option(self): """Test that serve command has --model-dir option.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--model-dir" in result.stdout def test_serve_has_no_max_memory_options(self): """The --max-model-memory and --max-process-memory CLI flags are removed.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--max-model-memory" not in result.stdout assert "--max-process-memory" not in result.stdout def test_serve_has_memory_guard_options(self): """Test that serve command exposes memory guard controls.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--memory-guard" in result.stdout assert "--memory-guard-gb" in result.stdout assert "safe" in result.stdout assert "balanced" in result.stdout assert "aggressive" in result.stdout def test_serve_no_model_specific_options(self): """Test that serve command does not have model-specific options (managed via admin page).""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) # These options are now managed via admin page, not CLI assert "--pin" not in result.stdout assert "--default-model" not in result.stdout assert "--max-tokens" not in result.stdout assert "--temperature" not in result.stdout assert "--top-p" not in result.stdout assert "--top-k" not in result.stdout assert "--force-sampling" not in result.stdout def test_serve_has_host_port_options(self): """Test that serve command has --host and --port options.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--host" in result.stdout assert "--port" in result.stdout def test_serve_has_scheduler_options(self): """Test that serve command has scheduler options.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--max-concurrent-requests" in result.stdout assert "--embedding-batch-size" in result.stdout def test_serve_has_cache_options(self): """Test that serve command has cache options.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--paged-ssd-cache-dir" in result.stdout assert "--paged-ssd-cache-max-size" in result.stdout assert "--no-cache" in result.stdout def test_serve_has_mcp_option(self): """Test that serve command has --mcp-config option.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--mcp-config" in result.stdout def test_serve_has_base_path_option(self): """Test that serve command has --base-path option.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--base-path" in result.stdout def test_serve_has_api_key_option(self): """Test that serve command has --api-key option.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) assert "--api-key" in result.stdout class TestLaunchCommandOptions: """Tests for launch command options via help output.""" def test_launch_has_host_port_options(self): """Test that launch command has --host and --port options.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "launch", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 assert "--host" in result.stdout assert "--port" in result.stdout def test_launch_has_model_option(self): """Test that launch command has --model option.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "launch", "--help"], capture_output=True, text=True, timeout=10, ) assert "--model" in result.stdout def test_launch_has_claude_tier_options(self): """Claude tier options should remain accepted for copied app commands.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "launch", "--help"], capture_output=True, text=True, timeout=10, ) assert "--opus" in result.stdout assert "--sonnet" in result.stdout assert "--haiku" in result.stdout def test_launch_lists_hermes(self): """Test that launch help lists Hermes as an available integration.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "launch", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 assert "hermes" in result.stdout assert "Hermes Agent" in result.stdout def test_launch_lists_codex_app(self): """Test that launch help lists the Codex Desktop App target.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "launch", "--help"], capture_output=True, text=True, timeout=10, ) assert result.returncode == 0 assert "codex_app" in result.stdout assert "Codex App" in " ".join(result.stdout.split()) class TestLaunchCommandFunction: """Tests for launch command runtime behavior.""" def test_launch_command_passes_model_type_to_integration(self): """VLM model metadata should be forwarded to integrations.""" from omlx.cli import launch_command integration = MagicMock() integration.display_name = "OpenCode" integration.is_installed.return_value = True health_response = MagicMock() health_response.raise_for_status.return_value = None status_response = MagicMock() status_response.ok = True status_response.json.return_value = { "models": [ { "id": "qwen2.5-vl", "model_type": "vlm", "max_context_window": 32768, "max_tokens": 8192, } ] } settings = MagicMock() settings.server.host = "127.0.0.1" settings.server.port = 8000 args = argparse.Namespace( tool="opencode", host=None, port=None, api_key="test-key", model="qwen2.5-vl", tools_profile="coding", ) with ( patch("requests.get", side_effect=[health_response, status_response]), patch("omlx.integrations.get_integration", return_value=integration), patch("omlx.settings.GlobalSettings.load", return_value=settings), ): launch_command(args) integration.launch.assert_called_once() ctx = integration.launch.call_args.args[0] assert ctx.host == "127.0.0.1" assert ctx.port == 8000 assert ctx.api_key == "test-key" assert ctx.model == "qwen2.5-vl" assert ctx.tools_profile == "coding" assert ctx.context_window == 32768 assert ctx.max_tokens == 8192 assert ctx.model_type == "vlm" assert ctx.extra_args == () def test_launch_command_resolves_alias_status_metadata(self): """Alias model IDs should keep status metadata from the real model.""" from omlx.cli import launch_command integration = MagicMock() integration.display_name = "OpenCode" integration.is_installed.return_value = True health_response = MagicMock() health_response.raise_for_status.return_value = None status_response = MagicMock() status_response.ok = True status_response.json.return_value = { "models": [ { "id": "qwen2.5-vl-raw", "model_alias": "gpt-4o", "model_type": "vlm", "max_context_window": 32768, "max_tokens": 8192, "enable_thinking": False, } ] } settings = MagicMock() settings.server.host = "127.0.0.1" settings.server.port = 8000 args = argparse.Namespace( tool="opencode", host=None, port=None, api_key="test-key", model="gpt-4o", tools_profile="coding", ) with ( patch("requests.get", side_effect=[health_response, status_response]), patch("omlx.integrations.get_integration", return_value=integration), patch("omlx.settings.GlobalSettings.load", return_value=settings), ): launch_command(args) ctx = integration.launch.call_args.args[0] assert ctx.model == "gpt-4o" assert ctx.context_window == 32768 assert ctx.max_tokens == 8192 assert ctx.model_type == "vlm" assert ctx.reasoning is False def test_launch_command_forwards_extra_args(self): """Unknown CLI tokens (e.g. --resume ) should reach integration.launch.""" from omlx.cli import launch_command integration = MagicMock() integration.display_name = "Claude Code" integration.is_installed.return_value = True health_response = MagicMock() health_response.raise_for_status.return_value = None status_response = MagicMock() status_response.ok = True status_response.json.return_value = { "models": [ { "id": "qwen2.5-vl", "model_type": "llm", "max_context_window": 32768, "max_tokens": 8192, } ] } settings = MagicMock() settings.server.host = "127.0.0.1" settings.server.port = 8000 args = argparse.Namespace( tool="claude", host=None, port=None, api_key="test-key", model="qwen2.5-vl", tools_profile="coding", ) with ( patch("requests.get", side_effect=[health_response, status_response]), patch("omlx.integrations.get_integration", return_value=integration), patch("omlx.settings.GlobalSettings.load", return_value=settings), ): launch_command(args, extra_args=["--resume", "abc123"]) ctx = integration.launch.call_args.args[0] assert ctx.extra_args == ("--resume", "abc123") def test_launch_command_shows_picker_and_clears_saved_tiers(self): """Bare `omlx launch claude` shows the picker and ignores saved tier models.""" from omlx.cli import launch_command integration = MagicMock() integration.display_name = "Claude Code" integration.is_installed.return_value = True integration.select_model.return_value = "sonnet-local" health_response = MagicMock() health_response.raise_for_status.return_value = None status_map_response = MagicMock() status_map_response.ok = True status_map_response.json.return_value = {"models": []} models_response = MagicMock() models_response.raise_for_status.return_value = None models_response.json.return_value = { "data": [ {"id": "sonnet-local", "model_type": "llm"}, {"id": "opus-local", "model_type": "llm"}, ] } settings = SimpleNamespace( server=SimpleNamespace(host="127.0.0.1", port=8000), auth=SimpleNamespace(api_key="saved-key"), claude_code=SimpleNamespace( opus_model="opus-local", sonnet_model="sonnet-local", haiku_model="haiku-local", ), ) args = argparse.Namespace( tool="claude", host=None, port=None, api_key=None, model=None, tools_profile="coding", opus_model=None, sonnet_model=None, haiku_model=None, ) with ( patch( "requests.get", side_effect=[health_response, status_map_response, models_response], ), patch("omlx.integrations.get_integration", return_value=integration), patch("omlx.settings.GlobalSettings.load", return_value=settings), ): launch_command(args) integration.select_model.assert_called_once() ctx = integration.launch.call_args.args[0] assert ctx.model == "sonnet-local" assert ctx.opus_model is None assert ctx.sonnet_model is None assert ctx.haiku_model is None assert ctx.api_key == "saved-key" def test_launch_command_claude_cli_tiers_override_saved_settings(self): """Explicit --opus/--sonnet/--haiku should win over saved settings.""" from omlx.cli import launch_command integration = MagicMock() integration.display_name = "Claude Code" integration.is_installed.return_value = True health_response = MagicMock() health_response.raise_for_status.return_value = None status_response = MagicMock() status_response.ok = True status_response.json.return_value = {"models": []} settings = SimpleNamespace( server=SimpleNamespace(host="127.0.0.1", port=8000), auth=SimpleNamespace(api_key="saved-key"), claude_code=SimpleNamespace( opus_model="saved-opus", sonnet_model="saved-sonnet", haiku_model="saved-haiku", ), ) args = argparse.Namespace( tool="claude", host=None, port=None, api_key=None, model=None, tools_profile="coding", opus_model="cli-opus", sonnet_model="cli-sonnet", haiku_model="cli-haiku", ) with ( patch("requests.get", side_effect=[health_response, status_response]), patch("omlx.integrations.get_integration", return_value=integration), patch("omlx.settings.GlobalSettings.load", return_value=settings), ): launch_command(args) ctx = integration.launch.call_args.args[0] assert ctx.model == "cli-sonnet" assert ctx.opus_model == "cli-opus" assert ctx.sonnet_model == "cli-sonnet" assert ctx.haiku_model == "cli-haiku" class TestLaunchArgvParsing: """Tests for top-level argv parsing of `omlx launch ...`.""" def test_serve_still_rejects_unknown_args(self): """Non-launch commands must keep strict argparse rejection.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--bogus-flag"], capture_output=True, text=True, timeout=10, ) assert result.returncode != 0 assert ( "unrecognized arguments" in result.stderr or "--bogus-flag" in result.stderr ) class TestServeCommandFunctions: """Tests for serve command function.""" @staticmethod def _make_serve_args(tmp_path, host="127.0.0.1", port=8000, **overrides): defaults = { "model_dir": None, "host": host, "port": port, "log_level": None, "sse_keepalive_mode": None, "max_concurrent_requests": None, "embedding_batch_size": None, "memory_guard": None, "memory_guard_gb": None, "paged_ssd_cache_dir": None, "paged_ssd_cache_max_size": None, "hot_cache_max_size": None, "no_cache": True, "initial_cache_blocks": None, "mcp_config": None, "hf_endpoint": None, "hf_cache_enabled": None, "ms_endpoint": None, "http_proxy": None, "https_proxy": None, "no_proxy": None, "ca_bundle": None, "base_path": str(tmp_path), "api_key": None, } defaults.update(overrides) return argparse.Namespace(**defaults) @staticmethod def _make_settings(tmp_path, host="127.0.0.1", port=8000): log_dir = tmp_path / "logs" settings = SimpleNamespace() settings.base_path = tmp_path settings.server = SimpleNamespace( host=host, port=port, log_level="info", burst_decode_mode="balanced" ) settings.huggingface = SimpleNamespace(endpoint=None, hf_cache_enabled=True) settings.modelscope = SimpleNamespace(endpoint=None) settings.network = SimpleNamespace( http_proxy=None, https_proxy=None, no_proxy=None, ca_bundle=None, ) settings.logging = SimpleNamespace( retention_days=7, get_log_dir=lambda base_path: log_dir, ) settings.model = SimpleNamespace( get_model_dirs=lambda base_path: [tmp_path / "models"], ) settings.get_effective_model_dirs = lambda: [tmp_path / "models"] settings.memory = SimpleNamespace(memory_guard_tier="balanced") settings.mcp = SimpleNamespace(config_path=None) settings.cache = SimpleNamespace( enabled=False, get_ssd_cache_dir=lambda base_path: tmp_path / "cache", get_ssd_cache_max_size_bytes=lambda base_path: 0, get_hot_cache_max_size_bytes=lambda: 0, ) settings.auth = SimpleNamespace(api_key=None) settings.ensure_directories = lambda: log_dir.mkdir(parents=True, exist_ok=True) settings.validate = lambda: [] settings.save = MagicMock() settings.to_scheduler_config = lambda: SimpleNamespace( paged_ssd_cache_dir=None, paged_ssd_cache_max_size=0, hot_cache_max_size=0, ) return settings @staticmethod def _reserve_port(host="127.0.0.1"): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((host, 0)) sock.listen(1) return sock def test_serve_command_exists(self): """Test that serve_command function exists.""" from omlx.cli import serve_command assert callable(serve_command) def test_serve_model_dir_optional_with_default(self): """Test that serve --model-dir is optional with default ~/.omlx/models.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) # Should show that model-dir has a default assert "default" in result.stdout.lower() # Help text should mention ~/.omlx/models or similar assert ".omlx" in result.stdout or "model" in result.stdout.lower() def test_invalid_embedding_batch_size_is_not_persisted(self, tmp_path): """Invalid CLI scheduler values should fail before saving settings.json.""" result = subprocess.run( [ sys.executable, "-m", "omlx.cli", "serve", "--base-path", str(tmp_path), "--embedding-batch-size", "0", ], capture_output=True, text=True, timeout=10, ) assert result.returncode != 0 assert "embedding_batch_size" in result.stdout assert not (tmp_path / "settings.json").exists() def test_invalid_memory_guard_gb_is_not_persisted(self, tmp_path): """Invalid custom memory guard values should fail before saving settings.json.""" result = subprocess.run( [ sys.executable, "-m", "omlx.cli", "serve", "--base-path", str(tmp_path), "--memory-guard-gb", "0", ], capture_output=True, text=True, timeout=10, ) assert result.returncode != 0 assert "--memory-guard-gb" in result.stderr assert not (tmp_path / "settings.json").exists() def test_non_finite_memory_guard_gb_is_rejected(self, tmp_path): """NaN and infinity must not be accepted as custom memory ceilings.""" result = subprocess.run( [ sys.executable, "-m", "omlx.cli", "serve", "--base-path", str(tmp_path), "--memory-guard-gb", "nan", ], capture_output=True, text=True, timeout=10, ) assert result.returncode != 0 assert "finite number" in result.stderr assert not (tmp_path / "settings.json").exists() def test_serve_exits_on_port_conflict_before_importing_server( self, tmp_path, monkeypatch ): """Port conflicts should fail before server import can preload pinned models.""" import uvicorn from omlx.cli import serve_command listener = self._reserve_port() host, port = listener.getsockname() settings = self._make_settings(tmp_path, host=host, port=port) args = self._make_serve_args(tmp_path, host=host, port=port) previous_server = sys.modules.pop("omlx.server", None) events = [] original_bind_socket = uvicorn.Config.bind_socket def tracking_bind_socket(config): events.append("bind") return original_bind_socket(config) monkeypatch.setattr("omlx.settings.init_settings", lambda **kwargs: settings) monkeypatch.setattr( "omlx.logging_config.configure_file_logging", lambda **kwargs: None, ) monkeypatch.setattr("faulthandler.enable", lambda *args, **kwargs: None) monkeypatch.setattr("uvicorn.Config.bind_socket", tracking_bind_socket) try: with pytest.raises(SystemExit) as exc: serve_command(args) assert exc.value.code != 0 assert events == ["bind"] assert "omlx.server" not in sys.modules finally: listener.close() if previous_server is not None: sys.modules["omlx.server"] = previous_server def test_serve_hands_prebound_socket_to_uvicorn(self, tmp_path, monkeypatch): """Successful serve startup should pass the pre-bound socket into uvicorn.""" import omlx import uvicorn from omlx.cli import serve_command host, port = "127.0.0.1", 0 settings = self._make_settings(tmp_path, host=host, port=port) args = self._make_serve_args(tmp_path, host=host, port=port) events = [] fake_server = ModuleType("omlx.server") async def app(scope, receive, send): return None def fake_init_server(**kwargs): events.append("init") fake_server.app = app fake_server.init_server = MagicMock(side_effect=fake_init_server) monkeypatch.setitem(sys.modules, "omlx.server", fake_server) monkeypatch.setattr(omlx, "server", fake_server, raising=False) fake_mlx = ModuleType("mlx") fake_mlx_core = ModuleType("mlx.core") fake_mlx_core.device_info = lambda: {"memory_size": 0} fake_mlx_core.set_cache_limit = MagicMock() fake_mlx.core = fake_mlx_core monkeypatch.setitem(sys.modules, "mlx", fake_mlx) monkeypatch.setitem(sys.modules, "mlx.core", fake_mlx_core) monkeypatch.setattr("omlx.settings.init_settings", lambda **kwargs: settings) monkeypatch.setattr( "omlx.logging_config.configure_file_logging", lambda **kwargs: None, ) monkeypatch.setattr("faulthandler.enable", lambda *args, **kwargs: None) captured = {} original_bind_socket = uvicorn.Config.bind_socket def tracking_bind_socket(config): sock = original_bind_socket(config) events.append("bind") return sock def fake_run(self, sockets=None): self.config.load() events.append("run") captured["socket_name"] = sockets[0].getsockname() captured["socket_count"] = len(sockets) monkeypatch.setattr("uvicorn.Config.bind_socket", tracking_bind_socket) monkeypatch.setattr("uvicorn.Server.run", fake_run) serve_command(args) fake_server.init_server.assert_called_once() assert events == ["bind", "init", "run"] assert captured["socket_count"] == 1 assert captured["socket_name"][0] == host assert captured["socket_name"][1] > 0 class TestHasCliOverrides: """Tests for _has_cli_overrides() — detects explicitly passed CLI args.""" @staticmethod def _make_args(**kwargs): """Namespace with all serve defaults (None), then apply overrides.""" defaults = { "model_dir": None, "port": None, "host": None, "log_level": None, "embedding_batch_size": None, "memory_guard": None, "memory_guard_gb": None, } defaults.update(kwargs) return argparse.Namespace(**defaults) def test_no_overrides_returns_false(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args()) is False def test_host_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(host="0.0.0.0")) is True # Even the default value, when explicitly passed, counts as override assert _has_cli_overrides(self._make_args(host="127.0.0.1")) is True def test_port_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(port=9000)) is True assert _has_cli_overrides(self._make_args(port=8000)) is True def test_model_dir_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(model_dir="/tmp/models")) is True def test_log_level_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(log_level="info")) is True assert _has_cli_overrides(self._make_args(log_level="debug")) is True def test_embedding_batch_size_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(embedding_batch_size=4)) is True def test_memory_guard_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(memory_guard="safe")) is True def test_memory_guard_gb_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(memory_guard_gb=48.0)) is True def test_hf_cache_explicit(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(hf_cache_enabled=False)) is True assert _has_cli_overrides(self._make_args(hf_cache_enabled=True)) is True def test_multiple_overrides(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(self._make_args(host="0.0.0.0", port=9000)) is True def test_empty_namespace(self): from omlx.cli import _has_cli_overrides assert _has_cli_overrides(argparse.Namespace()) is False class TestCLIDocstrings: """Tests for CLI module docstrings and descriptions.""" def test_main_has_description(self): """Test that main help has description.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "--help"], capture_output=True, text=True, timeout=10, ) # Should have some description assert "omlx" in result.stdout.lower() or "llm" in result.stdout.lower() def test_serve_has_description(self): """Test that serve command has description.""" result = subprocess.run( [sys.executable, "-m", "omlx.cli", "serve", "--help"], capture_output=True, text=True, timeout=10, ) # Should describe multi-model serving assert ( "multi-model" in result.stdout.lower() or "server" in result.stdout.lower() )