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""" 

6Unit tests for LocalExecService to run Python scripts locally. 

7""" 

8 

9from typing import Any, Dict 

10 

11import json 

12 

13import pytest 

14 

15from mlos_bench.tunables.tunable import TunableValue 

16from mlos_bench.services.local.local_exec import LocalExecService 

17from mlos_bench.services.config_persistence import ConfigPersistenceService 

18from mlos_bench.util import path_join 

19 

20# pylint: disable=redefined-outer-name 

21 

22 

23@pytest.fixture 

24def local_exec_service() -> LocalExecService: 

25 """ 

26 Test fixture for LocalExecService. 

27 """ 

28 return LocalExecService(parent=ConfigPersistenceService()) 

29 

30 

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

32 """ 

33 Run a Python script using a local_exec service. 

34 """ 

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

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

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

38 

39 # Tunable parameters to save in JSON 

40 params: Dict[str, TunableValue] = { 

41 "sched_migration_cost_ns": 40000, 

42 "sched_granularity_ns": 800000, 

43 } 

44 

45 # Tunable parameters metadata 

46 params_meta: Dict[str, Any] = { 

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

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

49 } 

50 

51 with local_exec_service.temp_dir_context() as temp_dir: 

52 

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

54 json.dump(params, fh_input) 

55 

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

57 json.dump(params_meta, fh_meta) 

58 

59 script_path = local_exec_service.config_loader_service.resolve_path( 

60 "environments/os/linux/runtime/scripts/local/generate_kernel_config_script.py") 

61 

62 (return_code, _stdout, stderr) = local_exec_service.local_exec([ 

63 f"{script_path} {input_file} {meta_file} {output_file}" 

64 ], cwd=temp_dir, env=params) 

65 

66 assert stderr.strip() == "" 

67 assert return_code == 0 

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

69 

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

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

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

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

74 ]