Coverage for mlos_bench/mlos_bench/tests/optimizers/opt_bulk_register_test.py: 100%

1# 

2# Copyright (c) Microsoft Corporation. 

3# Licensed under the MIT License. 

4# 

5"""Unit tests for mock mlos_bench optimizer.""" 

6 

7 

8import pytest 

9 

10from mlos_bench.environments.status import Status 

11from mlos_bench.optimizers.base_optimizer import Optimizer 

12from mlos_bench.optimizers.mlos_core_optimizer import MlosCoreOptimizer 

13from mlos_bench.optimizers.mock_optimizer import MockOptimizer 

14from mlos_bench.tunables.tunable import TunableValue 

15 

16# pylint: disable=redefined-outer-name 

17 

18 

19@pytest.fixture 

20def mock_configs_str(mock_configs: list[dict]) -> list[dict]: 

21 """ 

22 Same as `mock_config` above, but with all values converted to strings. 

23 

24 (This can happen when we retrieve the data from storage). 

25 """ 

26 return [{key: str(val) for (key, val) in config.items()} for config in mock_configs] 

27 

28 

29@pytest.fixture 

30def mock_scores() -> list[dict[str, TunableValue] | None]: 

31 """Mock benchmark results from earlier experiments.""" 

32 return [ 

33 None, 

34 {"score": 88.88}, 

35 {"score": 66.66}, 

36 {"score": 99.99}, 

37 ] 

38 

39 

40@pytest.fixture 

41def mock_status() -> list[Status]: 

42 """Mock status values for earlier experiments.""" 

43 return [Status.FAILED, Status.SUCCEEDED, Status.SUCCEEDED, Status.SUCCEEDED] 

44 

45 

46def _test_opt_update_min( 

47 opt: Optimizer, 

48 configs: list[dict], 

49 scores: list[dict[str, TunableValue] | None], 

50 status: list[Status] | None = None, 

51) -> None: 

52 """Test the bulk update of the optimizer on the minimization problem.""" 

53 opt.bulk_register(configs, scores, status) 

54 (score, tunables) = opt.get_best_observation() 

55 assert score is not None 

56 assert score["score"] == pytest.approx(66.66, 0.01) 

57 assert tunables is not None 

58 assert tunables.get_param_values() == { 

59 "vmSize": "Standard_B4ms", 

60 "idle": "mwait", 

61 "kernel_sched_migration_cost_ns": -1, 

62 "kernel_sched_latency_ns": 3000000, 

63 } 

64 

65 

66def _test_opt_update_max( 

67 opt: Optimizer, 

68 configs: list[dict], 

69 scores: list[dict[str, TunableValue] | None], 

70 status: list[Status] | None = None, 

71) -> None: 

72 """Test the bulk update of the optimizer on the maximization problem.""" 

73 opt.bulk_register(configs, scores, status) 

74 (score, tunables) = opt.get_best_observation() 

75 assert score is not None 

76 assert score["score"] == pytest.approx(99.99, 0.01) 

77 assert tunables is not None 

78 assert tunables.get_param_values() == { 

79 "vmSize": "Standard_B2s", 

80 "idle": "mwait", 

81 "kernel_sched_migration_cost_ns": 200000, 

82 "kernel_sched_latency_ns": 4000000, 

83 } 

84 

85 

86def test_update_mock_min( 

87 mock_opt: MockOptimizer, 

88 mock_configs: list[dict], 

89 mock_scores: list[dict[str, TunableValue] | None], 

90 mock_status: list[Status], 

91) -> None: 

92 """Test the bulk update of the mock optimizer on the minimization problem.""" 

93 _test_opt_update_min(mock_opt, mock_configs, mock_scores, mock_status) 

94 # make sure the first suggestion after bulk load is *NOT* the default config: 

95 assert mock_opt.suggest().get_param_values() == { 

96 "vmSize": "Standard_B4ms", 

97 "idle": "halt", 

98 "kernel_sched_migration_cost_ns": 13112, 

99 "kernel_sched_latency_ns": 796233790, 

100 } 

101 

102 

103def test_update_mock_min_str( 

104 mock_opt: MockOptimizer, 

105 mock_configs_str: list[dict], 

106 mock_scores: list[dict[str, TunableValue] | None], 

107 mock_status: list[Status], 

108) -> None: 

109 """Test the bulk update of the mock optimizer with all-strings data.""" 

110 _test_opt_update_min(mock_opt, mock_configs_str, mock_scores, mock_status) 

111 

112 

113def test_update_mock_max( 

114 mock_opt_max: MockOptimizer, 

115 mock_configs: list[dict], 

116 mock_scores: list[dict[str, TunableValue] | None], 

117 mock_status: list[Status], 

118) -> None: 

119 """Test the bulk update of the mock optimizer on the maximization problem.""" 

120 _test_opt_update_max(mock_opt_max, mock_configs, mock_scores, mock_status) 

121 

122 

123def test_update_flaml( 

124 flaml_opt: MlosCoreOptimizer, 

125 mock_configs: list[dict], 

126 mock_scores: list[dict[str, TunableValue] | None], 

127 mock_status: list[Status], 

128) -> None: 

129 """Test the bulk update of the FLAML optimizer.""" 

130 _test_opt_update_min(flaml_opt, mock_configs, mock_scores, mock_status) 

131 

132 

133def test_update_flaml_max( 

134 flaml_opt_max: MlosCoreOptimizer, 

135 mock_configs: list[dict], 

136 mock_scores: list[dict[str, TunableValue] | None], 

137 mock_status: list[Status], 

138) -> None: 

139 """Test the bulk update of the FLAML optimizer.""" 

140 _test_opt_update_max(flaml_opt_max, mock_configs, mock_scores, mock_status) 

141 

142 

143def test_update_smac( 

144 smac_opt: MlosCoreOptimizer, 

145 mock_configs: list[dict], 

146 mock_scores: list[dict[str, TunableValue] | None], 

147 mock_status: list[Status], 

148) -> None: 

149 """Test the bulk update of the SMAC optimizer.""" 

150 _test_opt_update_min(smac_opt, mock_configs, mock_scores, mock_status) 

151 

152 

153def test_update_smac_max( 

154 smac_opt_max: MlosCoreOptimizer, 

155 mock_configs: list[dict], 

156 mock_scores: list[dict[str, TunableValue] | None], 

157 mock_status: list[Status], 

158) -> None: 

159 """Test the bulk update of the SMAC optimizer.""" 

160 _test_opt_update_max(smac_opt_max, mock_configs, mock_scores, mock_status)