Coverage for app/crud/app_version_approvals.py: 100.00%

1from uuid import UUID 

2 

3from fastapi import HTTPException, status 

4from sqlalchemy.orm import Session 

5 

6from app.models import App, AppVersionApproval, AppVersionApprovalStatus 

7from app.utils.time import utcnow 

8 

9 

10def submit_version( 

11 db: Session, 

12 app_id: UUID, 

13 version_tag: str, 

14 diff_url: str | None = None, 

15 notes: str | None = None, 

16) -> AppVersionApproval: 

17 """Submit a version for admin review. 

18 

19 Raises 409 if the version already has a PENDING or APPROVED entry — 

20 submitting again would be a no-op at best and confusing at worst. 

21 REJECTED versions can be resubmitted (creates a fresh PENDING row 

22 after the old one is removed). 

23 """ 

24 existing = ( 

25 db.query(AppVersionApproval) 

26 .filter( 

27 AppVersionApproval.appId == app_id, 

28 AppVersionApproval.version_tag == version_tag, 

29 ) 

30 .first() 

31 ) 

32 

33 if existing: 

34 if existing.status == AppVersionApprovalStatus.PENDING: 

35 raise HTTPException( 

36 status_code=status.HTTP_409_CONFLICT, 

37 detail="Version is already pending review", 

38 ) 

39 if existing.status == AppVersionApprovalStatus.APPROVED: 

40 raise HTTPException( 

41 status_code=status.HTTP_409_CONFLICT, 

42 detail="Version is already approved", 

43 ) 

44 # REJECTED → allow resubmission: delete old entry, create fresh one 

45 db.delete(existing) 

46 db.flush() 

47 

48 approval = AppVersionApproval( 

49 appId=app_id, 

50 version_tag=version_tag, 

51 diff_url=diff_url, 

52 notes=notes, 

53 status=AppVersionApprovalStatus.PENDING, 

54 created_at=utcnow(), 

55 ) 

56 db.add(approval) 

57 db.commit() 

58 db.refresh(approval) 

59 return approval 

60 

61 

62def get_pending_approvals(db: Session) -> list[AppVersionApproval]: 

63 """Return all PENDING version approvals for public apps, oldest first.""" 

64 return ( 

65 db.query(AppVersionApproval) 

66 .join(App, App.appId == AppVersionApproval.appId) 

67 .filter( 

68 AppVersionApproval.status == AppVersionApprovalStatus.PENDING, 

69 App.is_private.is_(False), 

70 ) 

71 .order_by(AppVersionApproval.created_at.asc()) 

72 .all() 

73 ) 

74 

75 

76def get_approvals_for_app(db: Session, app_id: UUID) -> list[AppVersionApproval]: 

77 """Return all version approval entries for a given app.""" 

78 return ( 

79 db.query(AppVersionApproval) 

80 .filter(AppVersionApproval.appId == app_id) 

81 .order_by(AppVersionApproval.created_at.desc()) 

82 .all() 

83 ) 

84 

85 

86def has_approved_version(db: Session, app_id: UUID, version_tag: str) -> bool: 

87 """Return True if the given version is approved for this app.""" 

88 return ( 

89 db.query(AppVersionApproval.approvalId) 

90 .filter( 

91 AppVersionApproval.appId == app_id, 

92 AppVersionApproval.version_tag == version_tag, 

93 AppVersionApproval.status == AppVersionApprovalStatus.APPROVED, 

94 ) 

95 .first() 

96 ) is not None 

97 

98 

99def has_any_approved_version(db: Session, app_id: UUID) -> bool: 

100 """Return True if the app has at least one approved version.""" 

101 return ( 

102 db.query(AppVersionApproval.approvalId) 

103 .filter( 

104 AppVersionApproval.appId == app_id, 

105 AppVersionApproval.status == AppVersionApprovalStatus.APPROVED, 

106 ) 

107 .first() 

108 ) is not None 

109 

110 

111def withdraw(db: Session, app_id: UUID, version_tag: str) -> None: 

112 """Delete a PENDING approval entry (owner withdraws submission).""" 

113 approval = _get_approval_or_404(db, app_id, version_tag) 

114 

115 if approval.status != AppVersionApprovalStatus.PENDING: 

116 raise HTTPException( 

117 status_code=status.HTTP_409_CONFLICT, 

118 detail=f"Only pending submissions can be withdrawn (current status: '{approval.status.value}')", 

119 ) 

120 

121 db.delete(approval) 

122 db.commit() 

123 

124 

125def _get_approval_or_404( 

126 db: Session, app_id: UUID, version_tag: str 

127) -> AppVersionApproval: 

128 approval = ( 

129 db.query(AppVersionApproval) 

130 .filter( 

131 AppVersionApproval.appId == app_id, 

132 AppVersionApproval.version_tag == version_tag, 

133 ) 

134 .first() 

135 ) 

136 if not approval: 

137 raise HTTPException( 

138 status_code=status.HTTP_404_NOT_FOUND, 

139 detail="Version approval entry not found", 

140 ) 

141 return approval 

142 

143 

144def approve( 

145 db: Session, 

146 app_id: UUID, 

147 version_tag: str, 

148 admin_id: UUID, 

149) -> AppVersionApproval: 

150 """Approve a PENDING or REJECTED version.""" 

151 approval = _get_approval_or_404(db, app_id, version_tag) 

152 

153 if approval.status == AppVersionApprovalStatus.APPROVED: 

154 raise HTTPException( 

155 status_code=status.HTTP_409_CONFLICT, 

156 detail="Version is already approved", 

157 ) 

158 

159 approval.status = AppVersionApprovalStatus.APPROVED 

160 approval.reviewed_by = admin_id 

161 approval.reviewed_at = utcnow() 

162 approval.rejection_reason = None 

163 db.commit() 

164 db.refresh(approval) 

165 return approval 

166 

167 

168def reject( 

169 db: Session, 

170 app_id: UUID, 

171 version_tag: str, 

172 admin_id: UUID, 

173 rejection_reason: str, 

174) -> AppVersionApproval: 

175 """Reject a PENDING version with a mandatory reason.""" 

176 approval = _get_approval_or_404(db, app_id, version_tag) 

177 

178 if approval.status != AppVersionApprovalStatus.PENDING: 

179 raise HTTPException( 

180 status_code=status.HTTP_409_CONFLICT, 

181 detail=f"Cannot reject a version with status '{approval.status.value}'", 

182 ) 

183 

184 approval.status = AppVersionApprovalStatus.REJECTED 

185 approval.reviewed_by = admin_id 

186 approval.reviewed_at = utcnow() 

187 approval.rejection_reason = rejection_reason 

188 db.commit() 

189 db.refresh(approval) 

190 return approval 

191 

192 

193def revoke( 

194 db: Session, 

195 app_id: UUID, 

196 version_tag: str, 

197 admin_id: UUID, 

198 rejection_reason: str, 

199) -> AppVersionApproval: 

200 """Revoke a previously APPROVED version (sets status back to REJECTED).""" 

201 approval = _get_approval_or_404(db, app_id, version_tag) 

202 

203 if approval.status != AppVersionApprovalStatus.APPROVED: 

204 raise HTTPException( 

205 status_code=status.HTTP_409_CONFLICT, 

206 detail=f"Cannot revoke a version with status '{approval.status.value}'", 

207 ) 

208 

209 approval.status = AppVersionApprovalStatus.REJECTED 

210 approval.reviewed_by = admin_id 

211 approval.reviewed_at = utcnow() 

212 approval.rejection_reason = rejection_reason 

213 db.commit() 

214 db.refresh(approval) 

215 return approval