import importlib
import sys
from unittest.mock import MagicMock, patch
@patch("motor.common.utils.process_utils.set_process_title")
def test_module_sets_engine_server_title_on_import(mock_set_title):
old_argv = sys.argv
try:
sys.argv = ["engine_server", "--dp-rank", "2"]
import motor.engine_server.cli.main as es_main
importlib.reload(es_main)
mock_set_title.assert_called_with("EngineServer-DP2")
finally:
sys.argv = old_argv
@patch("motor.engine_server.cli.main.setup_multiprocess_prometheus")
@patch("motor.engine_server.cli.main.EndpointFactory")
@patch("motor.engine_server.cli.main.ConfigFactory")
@patch("motor.config.endpoint.EndpointConfig.init_endpoint_config")
def test_main_runs_without_resetting_process_title(
mock_init_endpoint_config,
mock_config_factory_cls,
mock_endpoint_factory_cls,
mock_setup_prometheus,
):
mock_endpoint_config = MagicMock()
mock_endpoint_config.dp_rank = 2
mock_endpoint_config.engine_type = "vllm"
mock_init_endpoint_config.return_value = mock_endpoint_config
mock_config_factory_cls.return_value.parse.return_value = MagicMock()
mock_infer_instance = MagicMock()
mock_mgmt_instance = MagicMock()
mock_endpoint_factory_cls.return_value.get_infer_endpoint.return_value = mock_infer_instance
fake_infer_mod = MagicMock()
fake_infer_mod.InferEndpoint = MagicMock(return_value=mock_infer_instance)
fake_mgmt_mod = MagicMock()
fake_mgmt_mod.MgmtEndpoint = MagicMock(return_value=mock_mgmt_instance)
with patch.dict(
sys.modules,
{
"motor.engine_server.core.infer_endpoint": fake_infer_mod,
"motor.engine_server.core.mgmt_endpoint": fake_mgmt_mod,
},
):
import motor.engine_server.cli.main as es_main
es_main.main()
mock_init_endpoint_config.assert_called_once()