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

6Unit tests for mock mlos_bench optimizer. 

7""" 

8 

9from typing import Dict, List, Optional 

10 

11import pytest 

12 

13from mlos_bench.environments.status import Status 

14from mlos_bench.optimizers.base_optimizer import Optimizer 

15from mlos_bench.optimizers.mock_optimizer import MockOptimizer 

16from mlos_bench.optimizers.mlos_core_optimizer import MlosCoreOptimizer 

17from mlos_bench.tunables.tunable import TunableValue 

18 

19# pylint: disable=redefined-outer-name 

20 

21 

22@pytest.fixture 

23def mock_configs_str(mock_configs: List[dict]) -> List[dict]: 

24 """ 

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

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

27 """ 

28 return [ 

29 {key: str(val) for (key, val) in config.items()} 

30 for config in mock_configs 

31 ] 

32 

33 

34@pytest.fixture 

35def mock_scores() -> List[Optional[Dict[str, TunableValue]]]: 

36 """ 

37 Mock benchmark results from earlier experiments. 

38 """ 

39 return [ 

40 None, 

41 {"score": 88.88}, 

42 {"score": 66.66}, 

43 {"score": 99.99}, 

44 ] 

45 

46 

47@pytest.fixture 

48def mock_status() -> List[Status]: 

49 """ 

50 Mock status values for earlier experiments. 

51 """ 

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

53 

54 

55def _test_opt_update_min(opt: Optimizer, 

56 configs: List[dict], 

57 scores: List[Optional[Dict[str, TunableValue]]], 

58 status: Optional[List[Status]] = None) -> None: 

59 """ 

60 Test the bulk update of the optimizer on the minimization problem. 

61 """ 

62 opt.bulk_register(configs, scores, status) 

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

64 assert score == pytest.approx(66.66, 0.01) 

65 assert tunables is not None 

66 assert tunables.get_param_values() == { 

67 "vmSize": "Standard_B4ms", 

68 "idle": "mwait", 

69 "kernel_sched_migration_cost_ns": -1, 

70 'kernel_sched_latency_ns': 3000000, 

71 } 

72 

73 

74def _test_opt_update_max(opt: Optimizer, 

75 configs: List[dict], 

76 scores: List[Optional[Dict[str, TunableValue]]], 

77 status: Optional[List[Status]] = None) -> None: 

78 """ 

79 Test the bulk update of the optimizer on the maximization problem. 

80 """ 

81 opt.bulk_register(configs, scores, status) 

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

83 assert score == pytest.approx(99.99, 0.01) 

84 assert tunables is not None 

85 assert tunables.get_param_values() == { 

86 "vmSize": "Standard_B2s", 

87 "idle": "mwait", 

88 "kernel_sched_migration_cost_ns": 200000, 

89 'kernel_sched_latency_ns': 4000000, 

90 } 

91 

92 

93def test_update_mock_min(mock_opt: MockOptimizer, 

94 mock_configs: List[dict], 

95 mock_scores: List[Optional[Dict[str, TunableValue]]], 

96 mock_status: List[Status]) -> None: 

97 """ 

98 Test the bulk update of the mock optimizer on the minimization problem. 

99 """ 

100 _test_opt_update_min(mock_opt, mock_configs, mock_scores, mock_status) 

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

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

103 "vmSize": "Standard_B4ms", 

104 "idle": "halt", 

105 "kernel_sched_migration_cost_ns": 13112, 

106 'kernel_sched_latency_ns': 796233790, 

107 } 

108 

109 

110def test_update_mock_min_str(mock_opt: MockOptimizer, 

111 mock_configs_str: List[dict], 

112 mock_scores: List[Optional[Dict[str, TunableValue]]], 

113 mock_status: List[Status]) -> None: 

114 """ 

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

116 """ 

117 _test_opt_update_min(mock_opt, mock_configs_str, mock_scores, mock_status) 

118 

119 

120def test_update_mock_max(mock_opt_max: MockOptimizer, 

121 mock_configs: List[dict], 

122 mock_scores: List[Optional[Dict[str, TunableValue]]], 

123 mock_status: List[Status]) -> None: 

124 """ 

125 Test the bulk update of the mock optimizer on the maximization problem. 

126 """ 

127 _test_opt_update_max(mock_opt_max, mock_configs, mock_scores, mock_status) 

128 

129 

130def test_update_flaml(flaml_opt: MlosCoreOptimizer, 

131 mock_configs: List[dict], 

132 mock_scores: List[Optional[Dict[str, TunableValue]]], 

133 mock_status: List[Status]) -> None: 

134 """ 

135 Test the bulk update of the FLAML optimizer. 

136 """ 

137 _test_opt_update_min(flaml_opt, mock_configs, mock_scores, mock_status) 

138 

139 

140def test_update_flaml_max(flaml_opt_max: MlosCoreOptimizer, 

141 mock_configs: List[dict], 

142 mock_scores: List[Optional[Dict[str, TunableValue]]], 

143 mock_status: List[Status]) -> None: 

144 """ 

145 Test the bulk update of the FLAML optimizer. 

146 """ 

147 _test_opt_update_max(flaml_opt_max, mock_configs, mock_scores, mock_status) 

148 

149 

150def test_update_smac(smac_opt: MlosCoreOptimizer, 

151 mock_configs: List[dict], 

152 mock_scores: List[Optional[Dict[str, TunableValue]]], 

153 mock_status: List[Status]) -> None: 

154 """ 

155 Test the bulk update of the SMAC optimizer. 

156 """ 

157 _test_opt_update_min(smac_opt, mock_configs, mock_scores, mock_status) 

158 

159 

160def test_update_smac_max(smac_opt_max: MlosCoreOptimizer, 

161 mock_configs: List[dict], 

162 mock_scores: List[Optional[Dict[str, TunableValue]]], 

163 mock_status: List[Status]) -> None: 

164 """ 

165 Test the bulk update of the SMAC optimizer. 

166 """ 

167 _test_opt_update_max(smac_opt_max, mock_configs, mock_scores, mock_status)