add ut for envs.py

Signed-off-by: chengyuan <[email protected]>

Author: chengyuan

tests/ut/test_envs.py

@@ -13,95 +13,47 @@
 # This file is a part of the vllm-ascend project.
 
 import os
-from unittest.mock import patch
+import inspect
 
 from tests.ut.base import TestBase
-from vllm_ascend.envs import env_variables
-
+from vllm_ascend import envs
 
 class TestEnvVariables(TestBase):
-    _default_values = {
-        "MAX_JOBS": None,
-        "CMAKE_BUILD_TYPE": None,
-        "COMPILE_CUSTOM_KERNELS": True,
-        "CXX_COMPILER": None,
-        "C_COMPILER": None,
-        "SOC_VERSION": "ASCEND910B1",
-        "VERBOSE": False,
-        "ASCEND_HOME_PATH": None,
-        "HCCN_PATH": "/usr/local/Ascend/driver/tools/hccn_tool",
-        "HCCL_SO_PATH": None,
-        "PROMPT_DEVICE_ID": None,
-        "DECODE_DEVICE_ID": None,
-        "LLMDATADIST_COMM_PORT": "26000",
-        "LLMDATADIST_SYNC_CACHE_WAIT_TIME": "5000",
-        "VLLM_VERSION": None,
-        "VLLM_ASCEND_TRACE_RECOMPILES": False,
-        "VLLM_ENABLE_FUSED_EXPERTS_ALLGATHER_EP": False,
-        "VLLM_ASCEND_ENABLE_DBO": False,
-        "VLLM_ASCEND_MODEL_EXECUTE_TIME_OBSERVE": False,
-        "MOE_ALL2ALL_BUFFER": False,
-        "USE_OPTIMIZED_MODEL": True,
-        "SELECT_GATING_TOPK_SOTFMAX_EXPERTS": False,
-        "VLLM_ASCEND_KV_CACHE_MEGABYTES_FLOATING_TOLERANCE": 64,
-        "VLLM_ASCEND_ENABLE_TOPK_TOPP_OPTIMIZATION": True,
-        "DISAGGREGATED_PREFILL_RANK_TABLE_PATH": None,
-        "VLLM_ASCEND_LLMDD_RPC_IP": "0.0.0.0",
-        "VLLM_LLMDD_RPC_PORT": 5557,
-        "VLLM_ASCEND_MLA_PA": False,
-        "VLLM_ASCEND_ENABLE_MATMUL_ALLREDUCE": False
-    }
-
-    _test_values = {
-        "MAX_JOBS": ("8", "8"),
-        "CMAKE_BUILD_TYPE": ("Debug", "Debug"),
-        "COMPILE_CUSTOM_KERNELS": ("0", False),
-        "CXX_COMPILER": ("/usr/bin/g++", "/usr/bin/g++"),
-        "C_COMPILER": ("/usr/bin/gcc", "/usr/bin/gcc"),
-        "SOC_VERSION": ("ASCEND910A", "ASCEND910A"),
-        "VERBOSE": ("1", True),
-        "ASCEND_HOME_PATH": ("/opt/ascend", "/opt/ascend"),
-        "HCCN_PATH": ("/custom/hccn_tool", "/custom/hccn_tool"),
-        "HCCL_SO_PATH": ("libhccl_custom.so", "libhccl_custom.so"),
-        "PROMPT_DEVICE_ID": ("1", "1"),
-        "DECODE_DEVICE_ID": ("2", "2"),
-        "LLMDATADIST_COMM_PORT": ("27000", "27000"),
-        "LLMDATADIST_SYNC_CACHE_WAIT_TIME": ("6000", "6000"),
-        "VLLM_VERSION": ("0.9.1", "0.9.1"),
-        "VLLM_ASCEND_TRACE_RECOMPILES": ("1", True),
-        "VLLM_ENABLE_FUSED_EXPERTS_ALLGATHER_EP": ("1", True),
-        "VLLM_ASCEND_ENABLE_DBO": ("1", True),
-        "VLLM_ASCEND_MODEL_EXECUTE_TIME_OBSERVE": ("1", True),
-        "MOE_ALL2ALL_BUFFER": ("1", True),
-        "USE_OPTIMIZED_MODEL": ("0", False),
-        "SELECT_GATING_TOPK_SOTFMAX_EXPERTS": ("1", True),
-        "VLLM_ASCEND_KV_CACHE_MEGABYTES_FLOATING_TOLERANCE": ("128", 128),
-        "VLLM_ASCEND_ENABLE_TOPK_TOPP_OPTIMIZATION": ("0", False),
-        "DISAGGREGATED_PREFILL_RANK_TABLE_PATH":
-        ("/path/to/rank_table.json", "/path/to/rank_table.json"),
-        "VLLM_ASCEND_LLMDD_RPC_IP": ("192.168.1.1", "192.168.1.1"),
-        "VLLM_LLMDD_RPC_PORT": ("5558", 5558),
-        "VLLM_ASCEND_MLA_PA": ("1", True),
-        "VLLM_ASCEND_ENABLE_MATMUL_ALLREDUCE": ("1", True)
-    }
-
-    def setUp(self):
-        env_keys = set(env_variables.keys())
-        self.assertEqual(env_keys, set(self._default_values.keys()))
-        self.assertEqual(env_keys, set(self._test_values.keys()))
-
-    def test_default_values(self):
-        for var_name, getter in env_variables.items():
-            with self.subTest(var_name=var_name):
-                with patch.dict(os.environ, {}, clear=True):
-                    actual = getter()
-                    expected = self._default_values[var_name]
-                    self.assertEqual(actual, expected)
-
-    def test_set_values(self):
-        for var_name, getter in env_variables.items():
-            with self.subTest(var_name=var_name):
-                input_val, expected = self._test_values[var_name]
-                with patch.dict(os.environ, {var_name: input_val}):
-                    actual = getter()
-                    self.assertEqual(actual, expected)
+    @classmethod
+    def setUpClass(cls):
+        cls.env_vars = list(envs.env_variables.keys())
+
+    def test_env_vars_behavior(self):
+        for var_name in self.env_vars:
+            with self.subTest(var=var_name):
+                original_val = os.environ.get(var_name)
+                var_handler = envs.env_variables[var_name]
+
+                try:
+                    if var_name in os.environ:
+                        del os.environ[var_name]
+                    self.assertEqual(getattr(envs, var_name), var_handler())
+
+                    handler_source = inspect.getsource(var_handler)
+                    if 'int(' in handler_source:
+                        test_vals = ["123", "456"]
+                    elif 'bool(int(' in handler_source:
+                        test_vals = ["0", "1"]
+                    else:
+                        test_vals = [f"test_{var_name}", f"custom_{var_name}"]
+
+                    for test_val in test_vals:
+                        os.environ[var_name] = test_val
+                        self.assertEqual(getattr(envs, var_name), var_handler())
+
+                finally:
+                    if original_val is None:
+                        os.environ.pop(var_name, None)
+                    else:
+                        os.environ[var_name] = original_val
+
+    def test_dir_and_getattr(self):
+        self.assertEqual(sorted(envs.__dir__()), sorted(self.env_vars))
+        for var_name in self.env_vars:
+            with self.subTest(var=var_name):
+                getattr(envs, var_name)