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

1# Copyright (c) 2014 EMC Corporation. 

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 fnmatch 

17import types 

18 

19from oslo_config import cfg 

20from oslo_log import log 

21from oslo_utils import netutils 

22from oslo_utils import timeutils 

23from oslo_utils import units 

24import ssl 

25 

26CONF = cfg.CONF 

27LOG = log.getLogger(__name__) 

28 

29 

30def decorate_all_methods(decorator, debug_only=False): 

31 if debug_only and not CONF.debug: 31 ↛ 34line 31 didn't jump to line 34 because the condition on line 31 was always true

32 return lambda cls: cls 

33 

34 def _decorate_all_methods(cls): 

35 for attr_name, attr_val in cls.__dict__.items(): 

36 if (isinstance(attr_val, types.FunctionType) and 

37 not attr_name.startswith("_")): 

38 setattr(cls, attr_name, decorator(attr_val)) 

39 return cls 

40 

41 return _decorate_all_methods 

42 

43 

44def log_enter_exit(func): 

45 if not CONF.debug: 45 ↛ 48line 45 didn't jump to line 48 because the condition on line 45 was always true

46 return func 

47 

48 def inner(self, *args, **kwargs): 

49 LOG.debug("Entering %(cls)s.%(method)s.", 

50 {'cls': self.__class__.__name__, 

51 'method': func.__name__}) 

52 start = timeutils.utcnow() 

53 ret = func(self, *args, **kwargs) 

54 end = timeutils.utcnow() 

55 LOG.debug("Exiting %(cls)s.%(method)s. " 

56 "Spent %(duration)s sec. " 

57 "Return %(return)s.", 

58 {'cls': self.__class__.__name__, 

59 'duration': timeutils.delta_seconds(start, end), 

60 'method': func.__name__, 

61 'return': ret}) 

62 return ret 

63 

64 return inner 

65 

66 

67def do_match_any(full, matcher_list): 

68 """Finds items that match any of the matchers. 

69 

70 :param full: Full item list 

71 :param matcher_list: The list of matchers. Each matcher supports 

72 Unix shell-style wildcards 

73 :return: The matched items set and the unmatched items set 

74 """ 

75 matched = set() 

76 not_matched = set() 

77 

78 full = set([item.strip() for item in full]) 

79 matcher_list = set([item.strip() for item in matcher_list]) 

80 

81 for matcher in matcher_list: 

82 for item in full: 

83 if fnmatch.fnmatchcase(item, matcher): 

84 matched.add(item) 

85 not_matched = full - matched 

86 return matched, not_matched 

87 

88 

89def create_ssl_context(configuration): 

90 """Create context for ssl verification. 

91 

92 .. note:: starting from python 2.7.9 ssl adds create_default_context. 

93 We need to keep compatibility with previous python as well. 

94 """ 

95 try: 

96 if configuration.emc_ssl_cert_verify: 

97 context = ssl.create_default_context( 

98 capath=configuration.emc_ssl_cert_path) 

99 else: 

100 context = ssl.create_default_context() 

101 context.check_hostname = False 

102 context.verify_mode = ssl.CERT_NONE 

103 except AttributeError: 

104 LOG.warning('Creating ssl context is not supported on this ' 

105 'version of Python, ssl verification is disabled.') 

106 context = None 

107 return context 

108 

109 

110def parse_ipaddr(text): 

111 """Parse the output of VNX server_export command, get IPv4/IPv6 addresses. 

112 

113 Example: 

114 input: 192.168.100.102:[fdf8:f53b:82e4::57]:[fdf8:f53b:82e4::54] 

115 output: ['192.168.100.102', '[fdf8:f53b:82e4::57]', '[fdf8:f53b:82e4::54]'] 

116 

117 :param text: The output of VNX server_export command. 

118 :return: The list of IPv4/IPv6 addresses. The IPv6 address enclosed by []. 

119 """ 

120 rst = [] 

121 stk = [] 

122 

123 ipaddr = '' 

124 it = iter(text) 

125 

126 try: 

127 while True: 

128 i = next(it) 

129 if i == ':' and not stk and ipaddr: 

130 rst.append(ipaddr) 

131 ipaddr = '' 

132 elif i == ':' and not ipaddr: 

133 continue 

134 elif i == '[': 

135 stk.append(i) 

136 elif i == ']': 

137 rst.append('[%s]' % ipaddr) 

138 stk.pop() 

139 ipaddr = '' 

140 else: 

141 ipaddr += i 

142 except StopIteration: 

143 if ipaddr: 

144 rst.append(ipaddr) 

145 

146 return rst 

147 

148 

149def convert_ipv6_format_if_needed(ip_addr): 

150 """Convert IPv6 address format if needed. The IPv6 address enclosed by []. 

151 

152 For the invalid IPv6 cidr, its format will not be changed. 

153 

154 :param ip_addr: IPv6 address. 

155 :return: Converted IPv6 address. 

156 """ 

157 if netutils.is_valid_ipv6_cidr(ip_addr): 

158 ip_addr = '[%s]' % ip_addr 

159 return ip_addr 

160 

161 

162def export_unc_path(ip_addr): 

163 """Convert IPv6 address to valid UNC path. 

164 

165 In Microsoft Windows OS, UNC (Uniform Naming Convention) specifies a 

166 common syntax to describe the location of a network resource. 

167 

168 The colon which used by IPv6 is an illegal character in a UNC path name. 

169 So the IPv6 address need to be converted to valid UNC path. 

170 

171 References: 

172 - https://en.wikipedia.org/wiki/IPv6_address 

173 #Literal_IPv6_addresses_in_UNC_path_names 

174 - https://en.wikipedia.org/wiki/Path_(computing)#Uniform_Naming_Convention 

175 

176 :param ip_addr: IPv6 address. 

177 :return: UNC path. 

178 """ 

179 unc_suffix = '.ipv6-literal.net' 

180 if netutils.is_valid_ipv6(ip_addr): 

181 ip_addr = ip_addr.replace(':', '-') + unc_suffix 

182 return ip_addr 

183 

184 

185def bytes_to_gb(size): 

186 return round(float(size) / units.Gi, 2) 

187 

188 

189def mb_to_gb(size): 

190 return bytes_to_gb(size * units.Mi)