Coverage for portality/models/editors.py: 92%

1from portality.dao import DomainObject, ScrollInitialiseException 

2from portality.models import Account 

3 

4 

5class EditorGroup(DomainObject): 

6 """ 

7 ~~EditorGroup:Model~~ 

8 """ 

9 __type__ = "editor_group" 

10 

11 @classmethod 

12 def group_exists_by_name(cls, name): 

13 q = EditorGroupQuery(name) 

14 res = cls.query(q=q.query()) 

15 ids = [hit.get("_source", {}).get("id") for hit in res.get("hits", {}).get("hits", []) if "_source" in hit] 

16 if len(ids) == 0: 

17 return None 

18 if len(ids) > 0: 

19 return ids[0] 

20 

21 @classmethod 

22 def _groups_by_x(cls, **kwargs): 

23 """ Generalised editor groups by maned / editor / associate """ 

24 # ~~-> EditorGroupMember:Query~~ 

25 q = EditorGroupMemberQuery(**kwargs) 

26 try: 

27 return [g for g in cls.iterate(q.query(), page_size=100)] 

28 except ScrollInitialiseException: 

29 # If there's no mapping for editor groups, the account definitely has no assignments. 

30 return [] 

31 

32 @classmethod 

33 def groups_by_maned(cls, maned): 

34 return cls._groups_by_x(maned=maned) 

35 

36 @classmethod 

37 def groups_by_editor(cls, editor): 

38 return cls._groups_by_x(editor=editor) 

39 

40 @classmethod 

41 def groups_by_associate(cls, associate): 

42 return cls._groups_by_x(associate=associate) 

43 

44 @property 

45 def name(self): 

46 return self.data.get("name") 

47 

48 def set_name(self, val): 

49 self.data["name"] = val 

50 

51 @property 

52 def maned(self): 

53 return self.data.get("maned") 

54 

55 def set_maned(self, val): 

56 self.data["maned"] = val 

57 

58 def get_maned_account(self): 

59 return Account.pull(self.maned) 

60 

61 @property 

62 def editor(self): 

63 return self.data.get("editor") 

64 

65 def set_editor(self, val): 

66 self.data["editor"] = val 

67 

68 def get_editor_account(self): 

69 return Account.pull(self.editor) 

70 

71 @property 

72 def associates(self): 

73 return self.data.get("associates", []) 

74 

75 def set_associates(self, val): 

76 if not isinstance(val, list): 

77 val = [val] 

78 self.data["associates"] = val 

79 

80 def add_associate(self, val): 

81 if "associates" not in self.data: 

82 self.data["associates"] = [] 

83 self.data["associates"].append(val) 

84 

85 def get_associate_accounts(self): 

86 accs = [] 

87 for a in self.associates: 

88 acc = Account.pull(a) 

89 accs.append(acc) 

90 return accs 

91 

92 def is_member(self, account_name): 

93 """ Determine if an account is a member of this Editor Group """ 

94 all_eds = self.associates + [self.editor] 

95 return account_name in all_eds 

96 

97 

98class EditorGroupQuery(object): 

99 def __init__(self, name): 

100 self.name = name 

101 

102 def query(self): 

103 q = { 

104 "track_total_hits" : True, 

105 "query": {"term": {"name.exact": self.name}} 

106 } 

107 return q 

108 

109 

110class EditorGroupMemberQuery(object): 

111 """ 

112 ~~EditorGroupMember:Query->Elasticsearch:Technology~~ 

113 """ 

114 def __init__(self, editor=None, associate=None, maned=None): 

115 self.editor = editor 

116 self.associate = associate 

117 self.maned = maned 

118 

119 def query(self): 

120 q = { 

121 "track_total_hits": True, 

122 "query": {"bool": {"should": []}}, 

123 "sort": {"name.exact": {"order" : "asc"}} 

124 } 

125 if self.editor is not None: 

126 et = {"term": {"editor.exact": self.editor}} 

127 q["query"]["bool"]["should"].append(et) 

128 if self.associate is not None: 

129 at = {"term": {"associates.exact": self.associate}} 

130 q["query"]["bool"]["should"].append(at) 

131 if self.maned is not None: 

132 mt = {"term" : {"maned.exact" : self.maned}} 

133 q["query"]["bool"]["should"].append(mt) 

134 return q