Coverage for rfpy/model/acl.py: 95%

1import logging 

2from typing import TYPE_CHECKING, Optional 

3 

4from sqlalchemy import ( 

5 DateTime, 

6 text, 

7 Integer, 

8 ForeignKey, 

9 UniqueConstraint, 

10 TIMESTAMP, 

11) 

12from sqlalchemy.dialects import mysql 

13from sqlalchemy.orm import ( 

14 relationship, 

15 backref, 

16 foreign, 

17 Mapped, 

18 mapped_column, 

19 DynamicMapped, 

20) 

21 

22from .meta import Base 

23from rfpy.auth import ROLES 

24from rfpy.model.humans import CustomRole 

25 

26if TYPE_CHECKING: 

27 from rfpy.model.humans import User 

28 from rfpy.model.project import Project 

29 from rfpy.model.issue import Issue 

30 from rfpy.model.humans import Organisation 

31 from rfpy.model.questionnaire import Section 

32 

33 

34log = logging.getLogger(__name__) 

35 

36 

37class Participant(Base): 

38 __tablename__ = "project_org_roles" 

39 __table_args__ = (UniqueConstraint("project_id", "org_id"),) + Base.__table_args__ 

40 

41 project_id: Mapped[int] = mapped_column( 

42 mysql.INTEGER(11), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False 

43 ) 

44 

45 org_id: Mapped[str] = mapped_column( 

46 mysql.VARCHAR(length=50), 

47 ForeignKey("organisations.id", onupdate="CASCADE"), 

48 nullable=False, 

49 server_default=text("''"), 

50 ) 

51 role: Mapped[Optional[str]] = mapped_column(mysql.VARCHAR(length=255), default=None, nullable=True) 

52 organisation: Mapped["Organisation"] = relationship( 

53 "Organisation", innerjoin=True, backref="participants" 

54 ) 

55 project: Mapped["Project"] = relationship( 

56 "Project", back_populates="participants_query" 

57 ) 

58 

59 permissions: DynamicMapped["ProjectPermission"] = relationship( 

60 "ProjectPermission", 

61 lazy="dynamic", 

62 back_populates="participant", 

63 cascade="all, delete", 

64 passive_deletes=True, 

65 ) 

66 custom_role: Mapped[CustomRole] = relationship( 

67 CustomRole, foreign_keys=[role], primaryjoin=CustomRole.id == foreign(role) 

68 ) 

69 

70 def __repr__(self) -> str: 

71 args = (self.org_id, self.project_id, self.role) 

72 return "%s participant in project %s with role %s" % args 

73 

74 @property 

75 def role_permissions(self): 

76 return ROLES[self.role] 

77 

78 @property 

79 def role_name(self): 

80 if self.role in ROLES: 

81 return self.role 

82 elif self.custom_role: 

83 return self.custom_role.name 

84 else: 

85 logging.error("Participant # %s has no role configured", self.id) 

86 

87 def as_dict(self) -> dict: 

88 org_dict = self.organisation.as_dict() 

89 return {"organisation": org_dict, "role": self.role_name} 

90 

91 

92class ProjectPermission(Base): 

93 __tablename__ = "project_permissions" 

94 __table_args__ = ( 

95 UniqueConstraint("project_org_role_id", "user_id"), 

96 ) + Base.__table_args__ 

97 

98 project_org_role_id: Mapped[Optional[int]] = mapped_column( 

99 mysql.INTEGER(11), ForeignKey("project_org_roles.id", ondelete="CASCADE"), nullable=True 

100 ) 

101 

102 user_id: Mapped[Optional[str]] = mapped_column( 

103 mysql.VARCHAR(length=50), ForeignKey("users.id", ondelete="CASCADE"), nullable=True 

104 ) 

105 

106 user: Mapped["User"] = relationship("User", back_populates="project_permissions") 

107 

108 participant: Mapped["Participant"] = relationship( 

109 "Participant", back_populates="permissions" 

110 ) 

111 project: Mapped["Project"] = relationship( 

112 "Project", 

113 secondary="project_org_roles", 

114 viewonly=True, 

115 backref=backref("permissions", lazy="dynamic"), 

116 ) 

117 

118 def __repr__(self) -> str: 

119 return "Permission for user %s on project_org_role %s" % ( 

120 self.user_id, 

121 self.project_org_role_id, 

122 ) 

123 

124 

125class SectionPermission(Base): 

126 __tablename__ = "section_permissions" 

127 __table_args__ = ( 

128 UniqueConstraint("project_permissions_id", "user_id", "section_id"), 

129 ) + Base.__table_args__ 

130 

131 section_id: Mapped[Optional[int]] = mapped_column( 

132 mysql.INTEGER(11), ForeignKey("sections.id", ondelete="CASCADE"), nullable=True 

133 ) 

134 user_id: Mapped[Optional[str]] = mapped_column( 

135 mysql.VARCHAR(length=50), ForeignKey("users.id", ondelete="CASCADE"), nullable=True 

136 ) 

137 project_permissions_id: Mapped[int] = mapped_column( 

138 mysql.INTEGER(11), 

139 ForeignKey("project_permissions.id", ondelete="CASCADE"), 

140 nullable=False, 

141 ) 

142 

143 section: Mapped["Section"] = relationship( 

144 "Section", backref=backref("permissions"), viewonly=True 

145 ) 

146 user: Mapped["User"] = relationship("User", back_populates="section_permissions") 

147 project_permission: Mapped["ProjectPermission"] = relationship( 

148 "ProjectPermission", backref=backref("section_permissions") 

149 ) 

150 

151 def __repr__(self) -> str: 

152 return "Permission for user %s on section %s" % (self.user_id, self.section_id) 

153 

154 

155class TokenExpired(Exception): 

156 pass 

157 

158 

159class UserAdminToken(Base): 

160 __tablename__ = "user_admin_tokens" 

161 __mapper_args__ = {"polymorphic_on": "kind", "polymorphic_identity": None} 

162 id = None # type: ignore 

163 token: Mapped[str] = mapped_column(mysql.VARCHAR(length=255), primary_key=True) 

164 user_id: Mapped[Optional[str]] = mapped_column( 

165 mysql.VARCHAR(length=50), ForeignKey("users.id", ondelete="CASCADE") 

166 ) 

167 email_address: Mapped[Optional[str]] = mapped_column(mysql.VARCHAR(length=100)) 

168 buyer_org_id: Mapped[Optional[str]] = mapped_column("buyer_org_id", mysql.VARCHAR(length=100)) 

169 manager_id: Mapped[Optional[str]] = mapped_column(mysql.VARCHAR(length=50)) 

170 issue_id: Mapped[Optional[int]] = mapped_column( 

171 Integer, ForeignKey("issues.id", ondelete="CASCADE") 

172 ) 

173 kind: Mapped[str] = mapped_column(mysql.VARCHAR(length=50), nullable=False) 

174 created_date: Mapped[DateTime] = mapped_column( 

175 TIMESTAMP, nullable=False, server_default=text("CURRENT_TIMESTAMP") 

176 ) 

177 expiry_date: Mapped[DateTime] = mapped_column(DateTime, nullable=False) 

178 

179 user: Mapped["User"] = relationship("User", back_populates="tokens") 

180 issue: Mapped["Issue"] = relationship("Issue") 

181 

182 def __repr__(self) -> str: 

183 kind = self.kind if self.kind else "Undefined" 

184 return "%s Token expires: %s" % (kind, self.expiry_date) 

185 

186 

187class LoginToken(UserAdminToken): 

188 __mapper_args__ = {"polymorphic_identity": "COLLAB_LOGIN"} 

189 

190 def __repr__(self) -> str: 

191 return "Login Token for user: %s, expires: %s" % ( 

192 self.user_id, 

193 self.expiry_date, 

194 ) 

195 

196 

197class UserRegistrationToken(UserAdminToken): 

198 __mapper_args__ = {"polymorphic_identity": "USER_REGISTRATION"} 

199 

200 def __repr__(self) -> str: 

201 return "User Registration for %s, expires: %s" % ( 

202 self.email_address, 

203 self.expiry_date, 

204 ) 

205 

206 

207class IssueRegistrationToken(UserAdminToken): 

208 __mapper_args__ = {"polymorphic_identity": "ISSUE_REGISTRATION"} 

209 

210 def __repr__(self) -> str: 

211 return "Issue Registration for %s, expires: %s" % ( 

212 self.email_address, 

213 self.expiry_date, 

214 ) 

215 

216 

217class PasswordResetToken(UserAdminToken): 

218 __mapper_args__ = {"polymorphic_identity": "PASSWORD_RESET"} 

219 

220 def __repr__(self) -> str: 

221 return "Password Reset for user: %s, expires: %s" % ( 

222 self.user_id, 

223 self.expiry_date, 

224 )