Coverage for portality / models / autocheck.py: 96%

1from portality.dao import DomainObject 

2from portality.lib.seamless import SeamlessMixin 

3from portality.lib.coerce import COERCE_MAP 

4from portality.lib import es_data_mapping 

5from portality.core import app 

6 

7import json 

8from copy import deepcopy 

9 

10 

11AUTOCHECK_STRUCT = { 

12 "fields": { 

13 "id": {"coerce": "unicode"}, 

14 "es_type": {"coerce": "unicode"}, 

15 "created_date": {"coerce": "utcdatetime"}, 

16 "last_updated": {"coerce": "utcdatetime"}, 

17 "application": {"coerce": "unicode"}, 

18 "journal": {"coerce": "unicode"} 

19 }, 

20 "lists": { 

21 "checks": {"contains": "object"} 

22 }, 

23 

24 "structs": { 

25 "checks": { 

26 "fields": { 

27 "id": {"coerce": "unicode"}, 

28 "field": {"coerce": "unicode"}, 

29 "original_value": {"coerce": "unicode"}, 

30 "replaced_value": {"coerce": "unicode"}, 

31 "advice": {"coerce": "unicode"}, 

32 "reference_url": {"coerce": "unicode"}, 

33 "checked_by": {"coerce": "unicode"}, 

34 "dismissed": {"coerce": "bool"}, 

35 "context": {"coerce": "unicode"} 

36 }, 

37 "lists": { 

38 "suggested_value": {"contains": "field", "coerce": "unicode"} 

39 } 

40 } 

41 } 

42} 

43 

44MAPPING_OPTS = { 

45 "dynamic": None, 

46 "coerces": app.config["DATAOBJ_TO_MAPPING_DEFAULTS"] 

47} 

48 

49 

50class Autocheck(SeamlessMixin, DomainObject): 

51 __type__ = "autocheck" 

52 

53 __SEAMLESS_STRUCT__ = [ 

54 AUTOCHECK_STRUCT 

55 ] 

56 

57 __SEAMLESS_COERCE__ = COERCE_MAP 

58 

59 def __init__(self, **kwargs): 

60 # FIXME: hack, to deal with ES integration layer being improperly abstracted 

61 if "_source" in kwargs: 

62 kwargs = kwargs["_source"] 

63 super(Autocheck, self).__init__(raw=kwargs) 

64 

65 def mappings(self): 

66 return es_data_mapping.create_mapping(self.__seamless_struct__.raw, MAPPING_OPTS) 

67 

68 @property 

69 def data(self): 

70 return self.__seamless__.data 

71 

72 @classmethod 

73 def for_application(cls, app_id): 

74 q = ApplicationQuery(app_id) 

75 res = cls.object_query(q.query()) 

76 if len(res) > 0: 

77 return res[0] 

78 return None 

79 

80 @classmethod 

81 def for_journal(cls, journal_id): 

82 q = JournalQuery(journal_id) 

83 res = cls.object_query(q.query()) 

84 if len(res) > 0: 

85 return res[0] 

86 return None 

87 

88 @classmethod 

89 def delete_all_but_latest(cls, journal_id=None, application_id=None): 

90 if journal_id is not None: 

91 q = JournalQuery(journal_id) 

92 elif application_id is not None: 

93 q = ApplicationQuery(application_id) 

94 else: 

95 return 

96 

97 res = cls.object_query(q.query()) 

98 if len(res) > 1: 

99 for r in res[1:]: 

100 r.delete() 

101 

102 @property 

103 def application(self): 

104 return self.__seamless__.get_single("application") 

105 

106 @application.setter 

107 def application(self, val): 

108 self.__seamless__.set_with_struct("application", val) 

109 

110 @property 

111 def journal(self): 

112 return self.__seamless__.get_single("journal") 

113 

114 @journal.setter 

115 def journal(self, val): 

116 self.__seamless__.set_with_struct("journal", val) 

117 

118 def add_check(self, field=None, original_value=None, suggested_value=None, advice=None, reference_url=None, context=None, checked_by=None): 

119 obj = {} 

120 if field is not None: 

121 obj["field"] = field 

122 if original_value is not None: 

123 obj["original_value"] = original_value 

124 if suggested_value is not None: 

125 if not isinstance(suggested_value, list): 

126 suggested_value = [suggested_value] 

127 obj["suggested_value"] = suggested_value 

128 if advice is not None: 

129 obj["advice"] = advice 

130 if reference_url is not None: 

131 obj["reference_url"] = reference_url 

132 if checked_by is not None: 

133 obj["checked_by"] = checked_by 

134 if context is not None: 

135 obj["context"] = json.dumps(context) 

136 

137 # ensure we add the check only once 

138 exists = self.__seamless__.exists_in_list("checks", matchsub=obj) 

139 

140 # now give this check an id and add it 

141 if not exists: 

142 obj["id"] = self.makeid() 

143 self.__seamless__.add_to_list_with_struct("checks", obj) 

144 

145 # return the constructed object, in case the caller could use it 

146 return obj 

147 

148 @property 

149 def checks(self): 

150 annos = self.__seamless__.get_list("checks") 

151 realised_checks = [] 

152 for anno in annos: 

153 anno = deepcopy(anno) 

154 if "context" in anno: 

155 anno["context"] = json.loads(anno["context"]) 

156 realised_checks.append(anno) 

157 return realised_checks 

158 

159 @property 

160 def checks_raw(self): 

161 return self.__seamless__.get_list("checks") 

162 

163 def dismiss(self, check_id): 

164 annos = self.checks_raw 

165 for anno in annos: 

166 if anno.get("id") == check_id: 

167 anno["dismissed"] = True 

168 break 

169 

170 def undismiss(self, check_id): 

171 annos = self.checks_raw 

172 for anno in annos: 

173 if anno.get("id") == check_id: 

174 del anno["dismissed"] 

175 break 

176 

177 

178class ApplicationQuery(object): 

179 def __init__(self, app_id): 

180 self._app_id = app_id 

181 

182 def query(self): 

183 return { 

184 "query" : { 

185 "bool": { 

186 "must": [ 

187 {"term": {"application.exact": self._app_id}} 

188 ] 

189 } 

190 }, 

191 "size": 1, 

192 "sort": { 

193 "created_date": {"order": "desc"} 

194 } 

195 } 

196 

197 

198class JournalQuery(object): 

199 def __init__(self, journal_id): 

200 self._journal_id = journal_id 

201 

202 def query(self): 

203 return { 

204 "query" : { 

205 "bool": { 

206 "must": [ 

207 {"term": {"journal.exact": self._journal_id}} 

208 ] 

209 } 

210 }, 

211 "size": 1, 

212 "sort": { 

213 "created_date": {"order": "desc"} 

214 } 

215 }