Coverage for manila/tests/share/drivers/dell_emc/common/enas/utils.py: 85%

1# Copyright (c) 2016 Dell Inc. or its subsidiaries. 

2# All Rights Reserved. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); you may 

5# not use this file except in compliance with the License. You may obtain 

6# a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

13# License for the specific language governing permissions and limitations 

14# under the License. 

15 

16import doctest 

17from unittest import mock 

18 

19from lxml import doctestcompare 

20 

21 

22CHECKER = doctestcompare.LXMLOutputChecker() 

23PARSE_XML = doctest.register_optionflag('PARSE_XML') 

24 

25 

26class RequestSideEffect(object): 

27 def __init__(self): 

28 self.actions = [] 

29 self.started = False 

30 

31 def append(self, resp=None, ex=None): 

32 if not self.started: 32 ↛ exitline 32 didn't return from function 'append' because the condition on line 32 was always true

33 self.actions.append((resp, ex)) 

34 

35 def __call__(self, *args, **kwargs): 

36 if not self.started: 

37 self.started = True 

38 self.actions.reverse() 

39 item = self.actions.pop() 

40 if item[1]: 

41 raise item[1] 

42 else: 

43 return item[0] 

44 

45 

46class SSHSideEffect(object): 

47 def __init__(self): 

48 self.actions = [] 

49 self.started = False 

50 

51 def append(self, resp=None, err=None, ex=None): 

52 if not self.started: 52 ↛ exitline 52 didn't return from function 'append' because the condition on line 52 was always true

53 self.actions.append((resp, err, ex)) 

54 

55 def __call__(self, rel_url, req_data=None, method=None, 

56 return_rest_err=True, *args, **kwargs): 

57 if not self.started: 

58 self.started = True 

59 self.actions.reverse() 

60 item = self.actions.pop() 

61 if item[2]: 

62 raise item[2] 

63 else: 

64 if return_rest_err: 64 ↛ 67line 64 didn't jump to line 67 because the condition on line 64 was always true

65 return item[0:2] 

66 else: 

67 return item[1] 

68 

69 

70class EMCMock(mock.Mock): 

71 def _get_req_from_call(self, call): 

72 if len(call) == 3: 72 ↛ 74line 72 didn't jump to line 74 because the condition on line 72 was always true

73 return call[1][0] 

74 elif len(call) == 2: 

75 return call[0][0] 

76 

77 def assert_has_calls(self, calls): 

78 if len(calls) != len(self.mock_calls): 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 raise AssertionError( 

80 'Mismatch error.\nExpected: %r\n' 

81 'Actual: %r' % (calls, self.mock_calls) 

82 ) 

83 

84 iter_expect = iter(calls) 

85 iter_actual = iter(self.mock_calls) 

86 

87 while True: 

88 try: 

89 expect = self._get_req_from_call(next(iter_expect)) 

90 actual = self._get_req_from_call(next(iter_actual)) 

91 except StopIteration: 

92 return True 

93 

94 if not isinstance(expect, bytes): 94 ↛ 96line 94 didn't jump to line 96 because the condition on line 94 was always true

95 expect = expect.encode("latin-1") 

96 if not isinstance(actual, bytes): 96 ↛ 98line 96 didn't jump to line 98 because the condition on line 96 was always true

97 actual = actual.encode("latin-1") 

98 if not CHECKER.check_output(expect, actual, PARSE_XML): 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 raise AssertionError( 

100 'Mismatch error.\nExpected: %r\n' 

101 'Actual: %r' % (calls, self.mock_calls) 

102 ) 

103 

104 

105class EMCNFSShareMock(mock.Mock): 

106 def assert_has_calls(self, calls): 

107 if len(calls) != len(self.mock_calls): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 raise AssertionError( 

109 'Mismatch error.\nExpected: %r\n' 

110 'Actual: %r' % (calls, self.mock_calls) 

111 ) 

112 

113 iter_expect = iter(calls) 

114 iter_actual = iter(self.mock_calls) 

115 

116 while True: 

117 try: 

118 expect = next(iter_expect)[1][0] 

119 actual = next(iter_actual)[1][0] 

120 except StopIteration: 

121 return True 

122 

123 if not self._option_check(expect, actual): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 raise AssertionError( 

125 'Mismatch error.\nExpected: %r\n' 

126 'Actual: %r' % (calls, self.mock_calls) 

127 ) 

128 

129 def _option_parser(self, option): 

130 option_map = {} 

131 for item in option.split(','): 

132 key, value = item.split('=') 

133 option_map[key] = value 

134 

135 return option_map 

136 

137 @staticmethod 

138 def _opt_value_from_map(opt_map, key): 

139 value = opt_map.get(key) 

140 if value: 

141 ret = set(value.split(':')) 

142 else: 

143 ret = set() 

144 return ret 

145 

146 def _option_check(self, expect, actual): 

147 if '-option' in actual and '-option' in expect: 

148 exp_option = expect[expect.index('-option') + 1] 

149 act_option = actual[actual.index('-option') + 1] 

150 

151 exp_opt_map = self._option_parser(exp_option) 

152 act_opt_map = self._option_parser(act_option) 

153 

154 for key in exp_opt_map: 

155 exp_set = self._opt_value_from_map(exp_opt_map, key) 

156 act_set = self._opt_value_from_map(act_opt_map, key) 

157 if exp_set != act_set: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 return False 

159 

160 return True 

161 

162 

163def patch_get_managed_ports_vnx(*arg, **kwargs): 

164 return mock.patch('manila.share.drivers.dell_emc.plugins.vnx.connection.' 

165 'VNXStorageConnection.get_managed_ports', 

166 mock.Mock(*arg, **kwargs)) 

167 

168 

169def patch_get_managed_ports_powermax(*arg, **kwargs): 

170 return mock.patch( 

171 'manila.share.drivers.dell_emc.plugins.powermax.connection.' 

172 'PowerMaxStorageConnection.get_managed_ports', 

173 mock.Mock(*arg, **kwargs))