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 

7from typing import Dict, List, Optional 

8 

9import pytest 

10 

11from mlos_bench.environments.status import Status 

12from mlos_bench.optimizers.base_optimizer import Optimizer 

13from mlos_bench.optimizers.mlos_core_optimizer import MlosCoreOptimizer 

14from mlos_bench.optimizers.mock_optimizer import MockOptimizer 

15from mlos_bench.tunables.tunable import TunableValue 

16 

17# pylint: disable=redefined-outer-name 

18 

19 

20@pytest.fixture 

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

22 """ 

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

24 

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

26 """ 

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

28 

29 

30@pytest.fixture 

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

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

33 return [ 

34 None, 

35 {"score": 88.88}, 

36 {"score": 66.66}, 

37 {"score": 99.99}, 

38 ] 

39 

40 

41@pytest.fixture 

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

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

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

45 

46 

47def _test_opt_update_min( 

48 opt: Optimizer, 

49 configs: List[dict], 

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

51 status: Optional[List[Status]] = None, 

52) -> None: 

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

54 opt.bulk_register(configs, scores, status) 

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

56 assert score is not None 

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

58 assert tunables is not None 

59 assert tunables.get_param_values() == { 

60 "vmSize": "Standard_B4ms", 

61 "idle": "mwait", 

62 "kernel_sched_migration_cost_ns": -1, 

63 "kernel_sched_latency_ns": 3000000, 

64 } 

65 

66 

67def _test_opt_update_max( 

68 opt: Optimizer, 

69 configs: List[dict], 

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

71 status: Optional[List[Status]] = None, 

72) -> None: 

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

74 opt.bulk_register(configs, scores, status) 

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

76 assert score is not None 

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

78 assert tunables is not None 

79 assert tunables.get_param_values() == { 

80 "vmSize": "Standard_B2s", 

81 "idle": "mwait", 

82 "kernel_sched_migration_cost_ns": 200000, 

83 "kernel_sched_latency_ns": 4000000, 

84 } 

85 

86 

87def test_update_mock_min( 

88 mock_opt: MockOptimizer, 

89 mock_configs: List[dict], 

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

91 mock_status: List[Status], 

92) -> None: 

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

94 _test_opt_update_min(mock_opt, mock_configs, mock_scores, mock_status) 

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

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

97 "vmSize": "Standard_B4ms", 

98 "idle": "halt", 

99 "kernel_sched_migration_cost_ns": 13112, 

100 "kernel_sched_latency_ns": 796233790, 

101 } 

102 

103 

104def test_update_mock_min_str( 

105 mock_opt: MockOptimizer, 

106 mock_configs_str: List[dict], 

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

108 mock_status: List[Status], 

109) -> None: 

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

111 _test_opt_update_min(mock_opt, mock_configs_str, mock_scores, mock_status) 

112 

113 

114def test_update_mock_max( 

115 mock_opt_max: MockOptimizer, 

116 mock_configs: List[dict], 

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

118 mock_status: List[Status], 

119) -> None: 

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

121 _test_opt_update_max(mock_opt_max, mock_configs, mock_scores, mock_status) 

122 

123 

124def test_update_flaml( 

125 flaml_opt: MlosCoreOptimizer, 

126 mock_configs: List[dict], 

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

128 mock_status: List[Status], 

129) -> None: 

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

131 _test_opt_update_min(flaml_opt, mock_configs, mock_scores, mock_status) 

132 

133 

134def test_update_flaml_max( 

135 flaml_opt_max: MlosCoreOptimizer, 

136 mock_configs: List[dict], 

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

138 mock_status: List[Status], 

139) -> None: 

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

141 _test_opt_update_max(flaml_opt_max, mock_configs, mock_scores, mock_status) 

142 

143 

144def test_update_smac( 

145 smac_opt: MlosCoreOptimizer, 

146 mock_configs: List[dict], 

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

148 mock_status: List[Status], 

149) -> None: 

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

151 _test_opt_update_min(smac_opt, mock_configs, mock_scores, mock_status) 

152 

153 

154def test_update_smac_max( 

155 smac_opt_max: MlosCoreOptimizer, 

156 mock_configs: List[dict], 

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

158 mock_status: List[Status], 

159) -> None: 

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

161 _test_opt_update_max(smac_opt_max, mock_configs, mock_scores, mock_status)