Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/netaddr/core.py: 45%

1#----------------------------------------------------------------------------- 

2# Copyright (c) 2008 by David P. D. Moss. All rights reserved. 

3# 

4# Released under the BSD license. See the LICENSE file for details. 

5#----------------------------------------------------------------------------- 

6"""Common code shared between various netaddr sub modules""" 

7 

8import sys as _sys 

9import pprint as _pprint 

10 

11from netaddr.compat import _callable, _iter_dict_keys 

12 

13#: True if platform is natively big endian, False otherwise. 

14BIG_ENDIAN_PLATFORM = _sys.byteorder == 'big' 

15 

16#: Use inet_pton() semantics instead of inet_aton() when parsing IPv4. 

17P = INET_PTON = 1 

18 

19#: Remove any preceding zeros from IPv4 address octets before parsing. 

20Z = ZEROFILL = 2 

21 

22#: Remove any host bits found to the right of an applied CIDR prefix. 

23N = NOHOST = 4 

24 

25#----------------------------------------------------------------------------- 

26# Custom exceptions. 

27#----------------------------------------------------------------------------- 

28class AddrFormatError(Exception): 

29 """ 

30 An Exception indicating a network address is not correctly formatted. 

31 """ 

32 pass 

33 

34 

35class AddrConversionError(Exception): 

36 """ 

37 An Exception indicating a failure to convert between address types or 

38 notations. 

39 """ 

40 pass 

41 

42 

43class NotRegisteredError(Exception): 

44 """ 

45 An Exception indicating that an OUI or IAB was not found in the IEEE 

46 Registry. 

47 """ 

48 pass 

49 

50 

51try: 

52 a = 42 

53 a.bit_length() 

54 # No exception, must be Python 2.7 or 3.1+ -> can use bit_length() 

55 def num_bits(int_val): 

56 """ 

57 :param int_val: an unsigned integer. 

58 

59 :return: the minimum number of bits needed to represent value provided. 

60 """ 

61 return int_val.bit_length() 

62except AttributeError: 

63 # a.bit_length() excepted, must be an older Python version. 

64 def num_bits(int_val): 

65 """ 

66 :param int_val: an unsigned integer. 

67 

68 :return: the minimum number of bits needed to represent value provided. 

69 """ 

70 numbits = 0 

71 while int_val: 

72 numbits += 1 

73 int_val >>= 1 

74 return numbits 

75 

76 

77class Subscriber(object): 

78 """ 

79 An abstract class defining the interface expected by a Publisher. 

80 """ 

81 

82 def update(self, data): 

83 """ 

84 A callback method used by a Publisher to notify this Subscriber about 

85 updates. 

86 

87 :param data: a Python object containing data provided by Publisher. 

88 """ 

89 raise NotImplementedError('cannot invoke virtual method!') 

90 

91 

92class PrettyPrinter(Subscriber): 

93 """ 

94 A concrete Subscriber that employs the pprint in the standard library to 

95 format all data from updates received, writing them to a file-like 

96 object. 

97 

98 Useful as a debugging aid. 

99 """ 

100 

101 def __init__(self, fh=_sys.stdout, write_eol=True): 

102 """ 

103 Constructor. 

104 

105 :param fh: a file-like object to write updates to. 

106 Default: sys.stdout. 

107 

108 

109 :param write_eol: if ``True`` this object will write newlines to 

110 output, if ``False`` it will not. 

111 """ 

112 self.fh = fh 

113 self.write_eol = write_eol 

114 

115 def update(self, data): 

116 """ 

117 A callback method used by a Publisher to notify this Subscriber about 

118 updates. 

119 

120 :param data: a Python object containing data provided by Publisher. 

121 """ 

122 self.fh.write(_pprint.pformat(data)) 

123 if self.write_eol: 

124 self.fh.write("\n") 

125 

126 

127class Publisher(object): 

128 """ 

129 A 'push' Publisher that maintains a list of Subscriber objects notifying 

130 them of state changes by passing them update data when it encounter events 

131 of interest. 

132 """ 

133 

134 def __init__(self): 

135 """Constructor""" 

136 self.subscribers = [] 

137 

138 def attach(self, subscriber): 

139 """ 

140 Add a new subscriber. 

141 

142 :param subscriber: a new object that implements the Subscriber object 

143 interface. 

144 """ 

145 if hasattr(subscriber, 'update') and _callable(subscriber.update): 

146 if subscriber not in self.subscribers: 

147 self.subscribers.append(subscriber) 

148 else: 

149 raise TypeError('%r does not support required interface!' % subscriber) 

150 

151 def detach(self, subscriber): 

152 """ 

153 Remove an existing subscriber. 

154 

155 :param subscriber: a new object that implements the Subscriber object 

156 interface. 

157 """ 

158 try: 

159 self.subscribers.remove(subscriber) 

160 except ValueError: 

161 pass 

162 

163 def notify(self, data): 

164 """ 

165 Send update data to to all registered Subscribers. 

166 

167 :param data: the data to be passed to each registered Subscriber. 

168 """ 

169 for subscriber in self.subscribers: 

170 subscriber.update(data) 

171 

172 

173class DictDotLookup(object): 

174 """ 

175 Creates objects that behave much like a dictionaries, but allow nested 

176 key access using object '.' (dot) lookups. 

177 

178 Recipe 576586: Dot-style nested lookups over dictionary based data 

179 structures - http://code.activestate.com/recipes/576586/ 

180 

181 """ 

182 

183 def __init__(self, d): 

184 for k in d: 

185 if isinstance(d[k], dict): 

186 self.__dict__[k] = DictDotLookup(d[k]) 

187 elif isinstance(d[k], (list, tuple)): 

188 l = [] 

189 for v in d[k]: 

190 if isinstance(v, dict): 

191 l.append(DictDotLookup(v)) 

192 else: 

193 l.append(v) 

194 self.__dict__[k] = l 

195 else: 

196 self.__dict__[k] = d[k] 

197 

198 def __getitem__(self, name): 

199 if name in self.__dict__: 

200 return self.__dict__[name] 

201 

202 def __iter__(self): 

203 return _iter_dict_keys(self.__dict__) 

204 

205 def __repr__(self): 

206 return _pprint.pformat(self.__dict__)