Coverage for app/logic/certification.py: 100%

1from peewee import JOIN, DoesNotExist, Case 

2 

3from app.models.certification import Certification 

4from app.models.certificationRequirement import CertificationRequirement 

5from app.models.requirementMatch import RequirementMatch 

6from app.models.eventParticipant import EventParticipant 

7 

8def getCertRequirementsWithCompletion(*, certification, username): 

9 """ 

10 Function to differentiate between simple requirements and requirements completion checking. 

11 See: `getCertRequirements` 

12 """ 

13 return getCertRequirements(certification, username) 

14 

15def getCertRequirements(certification=None, username=None): 

16 """ 

17 Return the requirements for all certifications, or for one if requested. 

18 

19 Keyword arguments: 

20 certification -- The id or object for a certification to request 

21 username -- The username to check for completion 

22 

23 Returns: 

24 A list of dictionaries with all certification data and requirements. If `certification` 

25 is given, returns only a list of requirement objects for the given certification. If  

26 `username` is given, the requirement objects have a `completed` attribute. 

27 """ 

28 reqList = (Certification.select(Certification, CertificationRequirement) 

29 .join(CertificationRequirement, JOIN.LEFT_OUTER, attr="requirement") 

30 .order_by(Certification.id, CertificationRequirement.order.asc(nulls="LAST"))) 

31 

32 if certification: 

33 if username: 

34 # I don't know how to add something to a select, so we have to recreate the whole query :( 

35 completedCase = Case(None, ((EventParticipant.user_id.is_null(True), 0),), 1) 

36 reqList = (Certification 

37 .select(Certification, CertificationRequirement, completedCase.alias("completed")) 

38 .join(CertificationRequirement, JOIN.LEFT_OUTER, attr="requirement") 

39 .join(RequirementMatch, JOIN.LEFT_OUTER) 

40 .join(EventParticipant, JOIN.LEFT_OUTER, on=(RequirementMatch.event == EventParticipant.event)) 

41 .where(EventParticipant.user.is_null(True) | (EventParticipant.user == username)) 

42 .order_by(Certification.id, CertificationRequirement.order.asc(nulls="LAST"))) 

43 

44 # we have to add the is not null check so that `cert.requirement` always exists 

45 reqList = reqList.where(Certification.id == certification, CertificationRequirement.id.is_null(False)) 

46 reqList = reqList.distinct() 

47 

48 certs = [] 

49 for cert in reqList: 

50 if username: 

51 cert.requirement.completed = bool(cert.__dict__['completed']) 

52 certs.append(cert.requirement) 

53 return certs 

54 

55 #return [cert.requirement for cert in reqList] 

56 

57 certs = {} 

58 for cert in reqList: 

59 if cert.id not in certs.keys(): 

60 certs[cert.id] = {"data": cert, "requirements": []} 

61 

62 if getattr(cert, 'requirement', None): 

63 certs[cert.id]["requirements"].append(cert.requirement) 

64 

65 return certs 

66 

67def updateCertRequirements(certId, newRequirements): 

68 """ 

69 Update the certification requirements in the database to match the provided list of requirement data. 

70 

71 The order of the list matters. Any ids that are in the database and not in `newRequirements` will be  

72 removed. IDs that do not exist in the database will be created (and given a new, auto-generated ID). 

73 

74 Arguments: 

75 certId - The id of the certification whose requirements we are updating 

76 newRequirements - a list of dictionaries. Each dictionary needs 'id', 'required', 'frequency', and 'name'. 

77 

78 Returns: 

79 A list of CertificationRequirement objects corresponding to the given `newRequirements` list. 

80 """ 

81 # check for missing ids to remove 

82 saveIds = [requirementData['id'] for requirementData in newRequirements] 

83 CertificationRequirement.delete().where(CertificationRequirement.certification_id == certId, CertificationRequirement.id.not_in(saveIds)).execute() 

84 

85 # update existing and add new requirements 

86 requirements = [] 

87 for order, requirementData in enumerate(newRequirements): 

88 try: 

89 newRequirement = CertificationRequirement.get_by_id(requirementData['id']) 

90 except DoesNotExist: 

91 newRequirement = CertificationRequirement() 

92 

93 newRequirement.certification = certId 

94 newRequirement.isRequired = bool(requirementData['required']) 

95 newRequirement.frequency = requirementData['frequency'] 

96 newRequirement.name = requirementData['name'] 

97 newRequirement.order = order 

98 newRequirement.save() 

99 

100 requirements.append(newRequirement) 

101 

102 return requirements 

103 

104def updateCertRequirementForEvent(event, requirement): 

105 """ 

106 Add a certification requirement to an event.  

107 Replaces the requirement for an event if the event already exists. 

108 

109 Arguments: 

110 event - an Event object or id 

111 requirement - a CertificationRequirement object or id 

112 """ 

113 # delete existing matches for our event 

114 for match in RequirementMatch.select().where(RequirementMatch.event == event): 

115 match.delete_instance() 

116 

117 RequirementMatch.create(event=event, requirement=requirement)