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

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 

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

12BIG_ENDIAN_PLATFORM = _sys.byteorder == 'big' 

13 

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

15INET_PTON = 1 

16 

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

18ZEROFILL = 2 

19 

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

21NOHOST = 4 

22 

23#: Use legacy ``inet_aton()`` semantics when parsing IPv4. 

24INET_ATON = 8 

25 

26 

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

28# Custom exceptions. 

29# ----------------------------------------------------------------------------- 

30class AddrFormatError(Exception): 

31 """ 

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

33 """ 

34 

35 pass 

36 

37 

38class AddrConversionError(Exception): 

39 """ 

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

41 notations. 

42 """ 

43 

44 pass 

45 

46 

47class NotRegisteredError(Exception): 

48 """ 

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

50 Registry. 

51 """ 

52 

53 pass 

54 

55 

56class Subscriber(object): 

57 """ 

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

59 """ 

60 

61 def update(self, data): 

62 """ 

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

64 updates. 

65 

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

67 """ 

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

69 

70 

71class PrettyPrinter(Subscriber): 

72 """ 

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

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

75 object. 

76 

77 Useful as a debugging aid. 

78 """ 

79 

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

81 """ 

82 Constructor. 

83 

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

85 Default: sys.stdout. 

86 

87 

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

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

90 """ 

91 self.fh = fh 

92 self.write_eol = write_eol 

93 

94 def update(self, data): 

95 """ 

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

97 updates. 

98 

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

100 """ 

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

102 if self.write_eol: 

103 self.fh.write('\n') 

104 

105 

106class Publisher(object): 

107 """ 

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

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

110 of interest. 

111 """ 

112 

113 def __init__(self): 

114 """Constructor""" 

115 self.subscribers = [] 

116 

117 def attach(self, subscriber): 

118 """ 

119 Add a new subscriber. 

120 

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

122 interface. 

123 """ 

124 if hasattr(subscriber, 'update') and hasattr(subscriber.update, '__call__'): 

125 if subscriber not in self.subscribers: 

126 self.subscribers.append(subscriber) 

127 else: 

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

129 

130 def detach(self, subscriber): 

131 """ 

132 Remove an existing subscriber. 

133 

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

135 interface. 

136 """ 

137 try: 

138 self.subscribers.remove(subscriber) 

139 except ValueError: 

140 pass 

141 

142 def notify(self, data): 

143 """ 

144 Send update data to to all registered Subscribers. 

145 

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

147 """ 

148 for subscriber in self.subscribers: 

149 subscriber.update(data) 

150 

151 

152class DictDotLookup(object): 

153 """ 

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

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

156 

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

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

159 

160 """ 

161 

162 def __init__(self, d): 

163 for k in d: 

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

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

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

167 l = [] 

168 for v in d[k]: 

169 if isinstance(v, dict): 

170 l.append(DictDotLookup(v)) 

171 else: 

172 l.append(v) 

173 self.__dict__[k] = l 

174 else: 

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

176 

177 def __getitem__(self, name): 

178 if name in self.__dict__: 

179 return self.__dict__[name] 

180 

181 def __iter__(self): 

182 return self.__dict__.keys() 

183 

184 def __repr__(self): 

185 return _pprint.pformat(self.__dict__)