Coverage for watcher/api/controllers/v1/types.py: 79%

1# Copyright 2013 Red Hat, Inc. 

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 

16from oslo_serialization import jsonutils 

17from oslo_utils import strutils 

18import wsme 

19from wsme import types as wtypes 

20 

21from watcher._i18n import _ 

22from watcher.common import exception 

23from watcher.common import utils 

24 

25 

26class UuidOrNameType(wtypes.UserType): 

27 """A simple UUID or logical name type.""" 

28 

29 basetype = wtypes.text 

30 name = 'uuid_or_name' 

31 

32 @staticmethod 

33 def validate(value): 

34 if not (utils.is_uuid_like(value) or utils.is_hostname_safe(value)): 

35 raise exception.InvalidUuidOrName(name=value) 

36 return value 

37 

38 @staticmethod 

39 def frombasetype(value): 

40 if value is None: 

41 return None 

42 return UuidOrNameType.validate(value) 

43 

44 

45class IntervalOrCron(wtypes.UserType): 

46 """A simple int value or cron syntax type""" 

47 

48 basetype = wtypes.text 

49 name = 'interval_or_cron' 

50 

51 @staticmethod 

52 def validate(value): 

53 if not (utils.is_int_like(value) or utils.is_cron_like(value)): 53 ↛ 54line 53 didn't jump to line 54 because the condition on line 53 was never true

54 raise exception.InvalidIntervalOrCron(name=value) 

55 return value 

56 

57 @staticmethod 

58 def frombasetype(value): 

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

60 return None 

61 return IntervalOrCron.validate(value) 

62 

63 

64interval_or_cron = IntervalOrCron() 

65 

66 

67class NameType(wtypes.UserType): 

68 """A simple logical name type.""" 

69 

70 basetype = wtypes.text 

71 name = 'name' 

72 

73 @staticmethod 

74 def validate(value): 

75 if not utils.is_hostname_safe(value): 

76 raise exception.InvalidName(name=value) 

77 return value 

78 

79 @staticmethod 

80 def frombasetype(value): 

81 if value is None: 

82 return None 

83 return NameType.validate(value) 

84 

85 

86class UuidType(wtypes.UserType): 

87 """A simple UUID type.""" 

88 

89 basetype = wtypes.text 

90 name = 'uuid' 

91 

92 @staticmethod 

93 def validate(value): 

94 if not utils.is_uuid_like(value): 

95 raise exception.InvalidUUID(uuid=value) 

96 return value 

97 

98 @staticmethod 

99 def frombasetype(value): 

100 if value is None: 

101 return None 

102 return UuidType.validate(value) 

103 

104 

105class BooleanType(wtypes.UserType): 

106 """A simple boolean type.""" 

107 

108 basetype = wtypes.text 

109 name = 'boolean' 

110 

111 @staticmethod 

112 def validate(value): 

113 try: 

114 return strutils.bool_from_string(value, strict=True) 

115 except ValueError as e: 

116 # raise Invalid to return 400 (BadRequest) in the API 

117 raise exception.Invalid(e) 

118 

119 @staticmethod 

120 def frombasetype(value): 

121 if value is None: 

122 return None 

123 return BooleanType.validate(value) 

124 

125 

126class JsonType(wtypes.UserType): 

127 """A simple JSON type.""" 

128 

129 basetype = wtypes.text 

130 name = 'json' 

131 

132 def __str__(self): 

133 # These are the json serializable native types 

134 return ' | '.join(map(str, (wtypes.text, int, float, 

135 BooleanType, list, dict, None))) 

136 

137 @staticmethod 

138 def validate(value): 

139 try: 

140 jsonutils.dumps(value, default=None) 

141 except TypeError: 

142 raise exception.Invalid(_('%s is not JSON serializable') % value) 

143 else: 

144 return value 

145 

146 @staticmethod 

147 def frombasetype(value): 

148 return JsonType.validate(value) 

149 

150 

151uuid = UuidType() 

152boolean = BooleanType() 

153jsontype = JsonType() 

154 

155 

156class MultiType(wtypes.UserType): 

157 """A complex type that represents one or more types. 

158 

159 Used for validating that a value is an instance of one of the types. 

160 

161 :param types: Variable-length list of types. 

162 

163 """ 

164 

165 def __init__(self, *types): 

166 self.types = types 

167 

168 def __str__(self): 

169 return ' | '.join(map(str, self.types)) 

170 

171 def validate(self, value): 

172 for t in self.types: 

173 if t is wsme.types.text and isinstance(value, wsme.types.bytes): 

174 value = value.decode() 

175 if isinstance(value, t): 

176 return value 

177 else: 

178 raise ValueError( 

179 _("Wrong type. Expected '%(type)s', got '%(value)s'"), 

180 type=self.types, value=type(value) 

181 ) 

182 

183 

184class JsonPatchType(wtypes.Base): 

185 """A complex type that represents a single json-patch operation.""" 

186 

187 path = wtypes.wsattr(wtypes.StringType(pattern=r'^(/[\w-]+)+$'), 

188 mandatory=True) 

189 op = wtypes.wsattr(wtypes.Enum(str, 'add', 'replace', 'remove'), 

190 mandatory=True) 

191 value = wsme.wsattr(jsontype, default=wtypes.Unset) 

192 

193 @staticmethod 

194 def internal_attrs(): 

195 """Returns a list of internal attributes. 

196 

197 Internal attributes can't be added, replaced or removed. This 

198 method may be overwritten by derived class. 

199 

200 """ 

201 return ['/created_at', '/id', '/links', '/updated_at', 

202 '/deleted_at', '/uuid'] 

203 

204 @staticmethod 

205 def mandatory_attrs(): 

206 """Returns a list of mandatory attributes. 

207 

208 Mandatory attributes can't be removed from the document. This 

209 method should be overwritten by derived class. 

210 

211 """ 

212 return [] 

213 

214 @staticmethod 

215 def validate(patch): 

216 _path = '/{0}'.format(patch.path.split('/')[1]) 

217 if _path in patch.internal_attrs(): 

218 msg = _("'%s' is an internal attribute and can not be updated") 

219 raise wsme.exc.ClientSideError(msg % patch.path) 

220 

221 if patch.path in patch.mandatory_attrs() and patch.op == 'remove': 

222 msg = _("'%s' is a mandatory attribute and can not be removed") 

223 raise wsme.exc.ClientSideError(msg % patch.path) 

224 

225 if patch.op != 'remove': 

226 if patch.value is wsme.Unset: 

227 msg = _("'add' and 'replace' operations needs value") 

228 raise wsme.exc.ClientSideError(msg) 

229 

230 ret = {'path': patch.path, 'op': patch.op} 

231 if patch.value is not wsme.Unset: 

232 ret['value'] = patch.value 

233 return ret