Coverage for watcher/objects/fields.py: 94%

1# Copyright 2013 IBM Corp. 

2# 

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

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

5# a copy of the License at 

6# 

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

8# 

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

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

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

12# License for the specific language governing permissions and limitations 

13# under the License. 

14 

15"""Utility methods for objects""" 

16 

17import ast 

18 

19from oslo_serialization import jsonutils 

20from oslo_versionedobjects import fields 

21 

22 

23BaseEnumField = fields.BaseEnumField 

24BooleanField = fields.BooleanField 

25DateTimeField = fields.DateTimeField 

26Enum = fields.Enum 

27FloatField = fields.FloatField 

28IntegerField = fields.IntegerField 

29ListOfStringsField = fields.ListOfStringsField 

30NonNegativeFloatField = fields.NonNegativeFloatField 

31NonNegativeIntegerField = fields.NonNegativeIntegerField 

32ObjectField = fields.ObjectField 

33StringField = fields.StringField 

34UnspecifiedDefault = fields.UnspecifiedDefault 

35 

36 

37class UUIDField(fields.UUIDField): 

38 def coerce(self, obj, attr, value): 

39 if value is None or value == "": 

40 return self._null(obj, attr) 

41 else: 

42 return self._type.coerce(obj, attr, value) 

43 

44 

45class Numeric(fields.FieldType): 

46 @staticmethod 

47 def coerce(obj, attr, value): 

48 if value is None: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 return value 

50 f_value = float(value) 

51 return f_value if not f_value.is_integer() else value 

52 

53 

54class NumericField(fields.AutoTypedField): 

55 AUTO_TYPE = Numeric() 

56 

57 

58class DictField(fields.AutoTypedField): 

59 AUTO_TYPE = fields.Dict(fields.FieldType()) 

60 

61 

62class ListOfUUIDsField(fields.AutoTypedField): 

63 AUTO_TYPE = fields.List(fields.UUID()) 

64 

65 

66class FlexibleDict(fields.FieldType): 

67 @staticmethod 

68 def coerce(obj, attr, value): 

69 if isinstance(value, str): 

70 value = ast.literal_eval(value) 

71 return dict(value) 

72 

73 

74class FlexibleDictField(fields.AutoTypedField): 

75 AUTO_TYPE = FlexibleDict() 

76 

77 # TODO(lucasagomes): In our code we've always translated None to {}, 

78 # this method makes this field to work like this. But probably won't 

79 # be accepted as-is in the oslo_versionedobjects library 

80 def _null(self, obj, attr): 

81 if self.nullable: 81 ↛ 83line 81 didn't jump to line 83 because the condition on line 81 was always true

82 return {} 

83 super(FlexibleDictField, self)._null(obj, attr) 

84 

85 

86class FlexibleListOfDict(fields.FieldType): 

87 @staticmethod 

88 def coerce(obj, attr, value): 

89 if isinstance(value, str): 

90 value = ast.literal_eval(value) 

91 return list(value) 

92 

93 

94class FlexibleListOfDictField(fields.AutoTypedField): 

95 AUTO_TYPE = FlexibleListOfDict() 

96 

97 # TODO(lucasagomes): In our code we've always translated None to {}, 

98 # this method makes this field to work like this. But probably won't 

99 # be accepted as-is in the oslo_versionedobjects library 

100 def _null(self, obj, attr): 

101 if self.nullable: 101 ↛ 103line 101 didn't jump to line 103 because the condition on line 101 was always true

102 return [] 

103 super(FlexibleListOfDictField, self)._null(obj, attr) 

104 

105 

106class Json(fields.FieldType): 

107 def coerce(self, obj, attr, value): 

108 if isinstance(value, str): 

109 loaded = jsonutils.loads(value) 

110 return loaded 

111 return value 

112 

113 def from_primitive(self, obj, attr, value): 

114 return self.coerce(obj, attr, value) 

115 

116 def to_primitive(self, obj, attr, value): 

117 return jsonutils.dumps(value) 

118 

119 

120class JsonField(fields.AutoTypedField): 

121 AUTO_TYPE = Json() 

122 

123# ### Notification fields ### # 

124 

125 

126class BaseWatcherEnum(Enum): 

127 

128 ALL = () 

129 

130 def __init__(self, **kwargs): 

131 super(BaseWatcherEnum, self).__init__(valid_values=self.__class__.ALL) 

132 

133 

134class NotificationPriority(BaseWatcherEnum): 

135 DEBUG = 'debug' 

136 INFO = 'info' 

137 WARNING = 'warning' 

138 ERROR = 'error' 

139 CRITICAL = 'critical' 

140 

141 ALL = (DEBUG, INFO, WARNING, ERROR, CRITICAL) 

142 

143 

144class NotificationPhase(BaseWatcherEnum): 

145 START = 'start' 

146 END = 'end' 

147 ERROR = 'error' 

148 

149 ALL = (START, END, ERROR) 

150 

151 

152class NotificationAction(BaseWatcherEnum): 

153 CREATE = 'create' 

154 UPDATE = 'update' 

155 EXCEPTION = 'exception' 

156 DELETE = 'delete' 

157 

158 STRATEGY = 'strategy' 

159 PLANNER = 'planner' 

160 EXECUTION = 'execution' 

161 

162 CANCEL = 'cancel' 

163 

164 ALL = (CREATE, UPDATE, EXCEPTION, DELETE, STRATEGY, PLANNER, EXECUTION, 

165 CANCEL) 

166 

167 

168class NotificationPriorityField(BaseEnumField): 

169 AUTO_TYPE = NotificationPriority() 

170 

171 

172class NotificationPhaseField(BaseEnumField): 

173 AUTO_TYPE = NotificationPhase() 

174 

175 

176class NotificationActionField(BaseEnumField): 

177 AUTO_TYPE = NotificationAction()