Coverage for mlos_bench/mlos_bench/tests/services/local/local_exec_python_test.py: 100%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5"""Unit tests for LocalExecService to run Python scripts locally.""" 

6 

7import json 

8from typing import Any, Dict 

9 

10import pytest 

11 

12from mlos_bench.services.config_persistence import ConfigPersistenceService 

13from mlos_bench.services.local.local_exec import LocalExecService 

14from mlos_bench.tunables.tunable import TunableValue 

15from mlos_bench.util import path_join 

16 

17# pylint: disable=redefined-outer-name 

18 

19 

20@pytest.fixture 

21def local_exec_service() -> LocalExecService: 

22 """Test fixture for LocalExecService.""" 

23 return LocalExecService(parent=ConfigPersistenceService()) 

24 

25 

26def test_run_python_script(local_exec_service: LocalExecService) -> None: 

27 """Run a Python script using a local_exec service.""" 

28 input_file = "./input-params.json" 

29 meta_file = "./input-params-meta.json" 

30 output_file = "./config-kernel.sh" 

31 

32 # Tunable parameters to save in JSON 

33 params: Dict[str, TunableValue] = { 

34 "sched_migration_cost_ns": 40000, 

35 "sched_granularity_ns": 800000, 

36 } 

37 

38 # Tunable parameters metadata 

39 params_meta: Dict[str, Any] = { 

40 "sched_migration_cost_ns": {"name_prefix": "/proc/sys/kernel/"}, 

41 "sched_granularity_ns": {"name_prefix": "/proc/sys/kernel/"}, 

42 } 

43 

44 with local_exec_service.temp_dir_context() as temp_dir: 

45 

46 with open(path_join(temp_dir, input_file), "wt", encoding="utf-8") as fh_input: 

47 json.dump(params, fh_input) 

48 

49 with open(path_join(temp_dir, meta_file), "wt", encoding="utf-8") as fh_meta: 

50 json.dump(params_meta, fh_meta) 

51 

52 script_path = local_exec_service.config_loader_service.resolve_path( 

53 "environments/os/linux/runtime/scripts/local/generate_kernel_config_script.py" 

54 ) 

55 

56 (return_code, _stdout, stderr) = local_exec_service.local_exec( 

57 [f"{script_path} {input_file} {meta_file} {output_file}"], 

58 cwd=temp_dir, 

59 env=params, 

60 ) 

61 

62 assert stderr.strip() == "" 

63 assert return_code == 0 

64 # assert stdout.strip() == "" 

65 

66 with open(path_join(temp_dir, output_file), "rt", encoding="utf-8") as fh_output: 

67 assert [ln.strip() for ln in fh_output.readlines()] == [ 

68 'echo "40000" > /proc/sys/kernel/sched_migration_cost_ns', 

69 'echo "800000" > /proc/sys/kernel/sched_granularity_ns', 

70 ]