Coverage for apps/workflow/rules_api.py: 82%

1from __future__ import annotations 

2 

3from ninja import Router, Schema 

4 

5from apps.accounts.auth import requires 

6from apps.workflow.models import AutomationRule 

7from domain.result import AppError, DomainError 

8from domain.rules import SEED_RULES 

9 

10router = Router(tags=["automation-rules"]) 

11 

12 

13class RuleOut(Schema): 

14 id: str 

15 name: str 

16 trigger: str 

17 conditions: list[dict] = [] 

18 actions: list[dict] = [] 

19 active: bool 

20 system: bool 

21 

22 

23class NewRuleInput(Schema): 

24 name: str 

25 trigger: str 

26 target_stage: str | None = None 

27 action_type: str = "notify" 

28 message: str | None = None 

29 fields: list[str] | None = None 

30 document_classification_id: str | None = None 

31 task_title: str | None = None 

32 

33 

34class RulePatch(Schema): 

35 name: str | None = None 

36 active: bool | None = None 

37 

38 

39def _to_rule_out(r: AutomationRule) -> dict: 

40 return { 

41 "id": r.rule_id, 

42 "name": r.name, 

43 "trigger": r.trigger, 

44 "conditions": r.conditions or [], 

45 "actions": r.actions or [], 

46 "active": r.is_active, 

47 "system": r.rule_id.startswith("ORG-"), 

48 } 

49 

50 

51def _seed_rules_if_empty(): 

52 if AutomationRule.objects.count() == 0: 

53 for r in SEED_RULES: 

54 conds = [ 

55 {"kind": c.kind, "field": c.field_name, "value": c.value, "values": list(c.values)} 

56 for c in r.conditions 

57 ] 

58 acts = [ 

59 { 

60 "type": a.action_type, 

61 "message": a.message, 

62 "fields": list(a.fields), 

63 "document_classification_id": a.document_classification_id, 

64 "task_title": a.task_title, 

65 } 

66 for a in r.actions 

67 ] 

68 AutomationRule.objects.create( 

69 rule_id=r.id, 

70 name=r.name, 

71 trigger=r.trigger, 

72 conditions=conds, 

73 actions=acts, 

74 is_active=r.active, 

75 ) 

76 

77 

78@router.get("/rules", response=list[RuleOut], auth=requires("LOAN_VIEW")) 

79def list_rules(request): 

80 _seed_rules_if_empty() 

81 return [_to_rule_out(r) for r in AutomationRule.objects.all().order_by("rule_id")] 

82 

83 

84@router.post("/rules", response={201: RuleOut}, auth=requires("LOAN_EDIT")) 

85def create_rule(request, payload: NewRuleInput): 

86 if not payload.name or not payload.name.strip(): 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 raise DomainError([AppError("VALIDATION", "Rule name is required.", "name")]) 

88 

89 _seed_rules_if_empty() 

90 custom_count = AutomationRule.objects.filter(rule_id__startswith="CUSTOM-").count() + 1 

91 rule_id = f"CUSTOM-{custom_count:05d}" 

92 while AutomationRule.objects.filter(rule_id=rule_id).exists(): 92 ↛ 93line 92 didn't jump to line 93 because the condition on line 92 was never true

93 custom_count += 1 

94 rule_id = f"CUSTOM-{custom_count:05d}" 

95 

96 conditions = [] 

97 if payload.target_stage: 97 ↛ 104line 97 didn't jump to line 104 because the condition on line 97 was always true

98 conditions.append({ 

99 "kind": "fieldIn", 

100 "field": "targetStage", 

101 "values": [payload.target_stage], 

102 }) 

103 

104 actions: list[dict] = [] 

105 action_type = payload.action_type.lower() 

106 if action_type == "block": 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true

107 actions.append({"type": "Stop", "message": payload.message or "Blocked by rule."}) 

108 elif action_type == "requirefields": 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 actions.append({"type": "RequireFields", "fields": payload.fields or []}) 

110 elif action_type == "senddocument": 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true

111 actions.append({ 

112 "type": "SendDocument", 

113 "document_classification_id": payload.document_classification_id or "LOAN_ORIGINATION_AGREEMENT", 

114 }) 

115 elif action_type == "createtask": 115 ↛ 119line 115 didn't jump to line 119 because the condition on line 115 was always true

116 actions.append({"type": "CreateTask", "task_title": payload.task_title or payload.message or "Follow up"}) 

117 else: 

118 # notify 

119 actions.append({"type": "CreateTask", "task_title": payload.message or "Automation notice"}) 

120 

121 rule = AutomationRule.objects.create( 

122 rule_id=rule_id, 

123 name=payload.name.strip(), 

124 trigger=payload.trigger, 

125 conditions=conditions, 

126 actions=actions, 

127 is_active=True, 

128 ) 

129 return 201, _to_rule_out(rule) 

130 

131 

132@router.patch("/rules/{rule_id}", response=RuleOut, auth=requires("LOAN_EDIT")) 

133def update_rule(request, rule_id: str, payload: RulePatch): 

134 _seed_rules_if_empty() 

135 rule = AutomationRule.objects.filter(rule_id=rule_id).first() 

136 if rule is None: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true

137 raise DomainError([AppError("NOT_FOUND", f"Rule {rule_id} not found.")]) 

138 

139 if payload.name is not None: 139 ↛ 143line 139 didn't jump to line 143 because the condition on line 139 was always true

140 if not payload.name.strip(): 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true

141 raise DomainError([AppError("VALIDATION", "Rule name is required.", "name")]) 

142 rule.name = payload.name.strip() 

143 if payload.active is not None: 143 ↛ 146line 143 didn't jump to line 146 because the condition on line 143 was always true

144 rule.is_active = payload.active 

145 

146 rule.save() 

147 return _to_rule_out(rule) 

148 

149 

150@router.delete("/rules/{rule_id}", response={204: None}, auth=requires("LOAN_EDIT")) 

151def delete_rule(request, rule_id: str): 

152 _seed_rules_if_empty() 

153 rule = AutomationRule.objects.filter(rule_id=rule_id).first() 

154 if rule is None: 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 raise DomainError([AppError("NOT_FOUND", f"Rule {rule_id} not found.")]) 

156 if rule.rule_id.startswith("ORG-"): 

157 raise DomainError( 

158 [AppError("RULE_PROTECTED", "Built-in ORG rules can't be deleted — toggle them off instead.")], 

159 status_code=422, 

160 ) 

161 rule.delete() 

162 return 204, None