Coverage for portality / models / admin_alert.py: 98%

1from portality.core import app 

2from portality.dao import DomainObject 

3from portality.lib import es_data_mapping 

4from portality.lib.coerce import COERCE_MAP 

5from portality.lib.seamless import SeamlessMixin 

6from portality.lib import dates 

7from portality.models import Account 

8 

9STRUCT = { 

10 "fields": { 

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

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

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

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

15 "message": {"coerce": "unicode"}, 

16 "source": {"coerce": "unicode"}, 

17 "state": {"coerce": "unicode", "allowed_values": ["new", "in_progress", "closed"]} 

18 }, 

19 "lists": { 

20 "audit": {"contains": "object"} 

21 }, 

22 "structs": { 

23 "audit": { 

24 "fields": { 

25 "user": {"coerce": "unicode"}, 

26 "date": {"coerce": "utcdatetime"}, 

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

28 "to_state": {"coerce": "unicode"} 

29 } 

30 } 

31 } 

32} 

33 

34MAPPING_OPTS = { 

35 "dynamic": None, 

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

37} 

38 

39class AdminAlert(SeamlessMixin, DomainObject): 

40 __type__ = "admin_alert" 

41 

42 __SEAMLESS_STRUCT__ = STRUCT 

43 __SEAMLESS_COERCE__ = COERCE_MAP 

44 

45 STATE_NEW = "new" 

46 STATE_IN_PROGRESS = "in_progress" 

47 STATE_CLOSED = "closed" 

48 

49 def __init__(self, **kwargs): 

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

51 if "_source" in kwargs: 

52 kwargs = kwargs["_source"] 

53 super(AdminAlert, self).__init__(raw=kwargs) 

54 

55 def mappings(self): 

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

57 

58 @property 

59 def data(self): 

60 return self.__seamless__.data 

61 

62 @property 

63 def source(self): 

64 return self.__seamless__.get_single("source") 

65 

66 @source.setter 

67 def source(self, val): 

68 self.__seamless__.set_with_struct("source", val) 

69 

70 @property 

71 def message(self): 

72 return self.__seamless__.get_single("message") 

73 

74 @message.setter 

75 def message(self, val): 

76 self.__seamless__.set_with_struct("message", val) 

77 

78 @property 

79 def state(self): 

80 return self.__seamless__.get_single("state") 

81 

82 @state.setter 

83 def state(self, val): 

84 self.__seamless__.set_with_struct("state", val) 

85 

86 @property 

87 def audit(self): 

88 return self.__seamless__.get_list("audit") 

89 

90 def change_state(self, new_state, user: Account, date=None): 

91 if new_state not in [self.STATE_NEW, self.STATE_IN_PROGRESS, self.STATE_CLOSED]: 

92 raise ValueError(f"Invalid state: {new_state}") 

93 old_state = self.state 

94 self.state = new_state 

95 audit_entry = { 

96 "user": user.id, 

97 "date": date or dates.now(), 

98 "from_state": old_state, 

99 "to_state": new_state 

100 } 

101 self.__seamless__.add_to_list_with_struct("audit", audit_entry) 

102 

103 self.save()