Coverage for agentos/tests/test_hitl.py: 0%

1""" 

2Tests for HITL (Human-in-the-Loop) approval module. 

3""" 

4 

5from agentos.hitl.approver import ( 

6 HumanInTheLoop, 

7 ApprovalRequest, 

8 ApprovalDecision, 

9 ApprovalStatus, 

10 RiskLevel, 

11 ApprovalPolicy, 

12) 

13from agentos.hitl.presets import ( 

14 default_approval_policy, 

15 permissive_approval_policy, 

16 strict_approval_policy, 

17) 

18 

19 

20class TestApprovalRequest: 

21 def test_create_request(self): 

22 req = ApprovalRequest( 

23 request_id="abc-123", 

24 action="delete_file", 

25 description="Delete /tmp/test.txt", 

26 risk_level=RiskLevel.HIGH, 

27 tool_name="file.delete", 

28 ) 

29 assert req.request_id == "abc-123" 

30 assert req.risk_level == RiskLevel.HIGH 

31 assert req.tool_name == "file.delete" 

32 

33 

34class TestApprovalDecision: 

35 def test_approved(self): 

36 d = ApprovalDecision(request_id="x", status=ApprovalStatus.APPROVED) 

37 assert d.is_approved 

38 assert not d.is_rejected 

39 

40 def test_modified_is_approved(self): 

41 d = ApprovalDecision(request_id="x", status=ApprovalStatus.MODIFIED, modified_args={"force": True}) 

42 assert d.is_approved 

43 

44 def test_rejected(self): 

45 d = ApprovalDecision(request_id="x", status=ApprovalStatus.REJECTED, reason="too risky") 

46 assert d.is_rejected 

47 assert not d.is_approved 

48 

49 

50class TestHumanInTheLoop: 

51 def test_low_risk_auto_skipped(self): 

52 hitl = HumanInTheLoop(policy=default_approval_policy()) 

53 req, decision = hitl.request_and_decide( 

54 action="read_file", 

55 description="Read config.yaml", 

56 risk_level=RiskLevel.LOW, 

57 tool_name="file.read", 

58 ) 

59 assert decision.status == ApprovalStatus.SKIPPED 

60 

61 def test_high_risk_needs_approval(self): 

62 hitl = HumanInTheLoop(policy=default_approval_policy()) 

63 hitl.callback = lambda r: ApprovalDecision( 

64 request_id=r.request_id, 

65 status=ApprovalStatus.APPROVED, 

66 reason="OK", 

67 ) 

68 req, decision = hitl.request_and_decide( 

69 action="delete_all", 

70 description="Delete production database", 

71 risk_level=RiskLevel.HIGH, 

72 tool_name="db.drop", 

73 ) 

74 assert decision.is_approved 

75 

76 def test_auto_approve_domain(self): 

77 policy = permissive_approval_policy() 

78 hitl = HumanInTheLoop(policy=policy) 

79 req, decision = hitl.request_and_decide( 

80 action="search", 

81 description="Search web", 

82 risk_level=RiskLevel.MEDIUM, 

83 tool_name="read.web_search", 

84 ) 

85 assert decision.status == ApprovalStatus.APPROVED 

86 

87 def test_blocked_domain(self): 

88 policy = strict_approval_policy() 

89 hitl = HumanInTheLoop(policy=policy) 

90 req, decision = hitl.request_and_decide( 

91 action="format", 

92 description="Format disk", 

93 risk_level=RiskLevel.CRITICAL, 

94 tool_name="delete.format_disk", 

95 ) 

96 assert decision.status == ApprovalStatus.REJECTED 

97 

98 def test_rejected_decision(self): 

99 hitl = HumanInTheLoop() 

100 hitl.callback = lambda r: ApprovalDecision( 

101 request_id=r.request_id, 

102 status=ApprovalStatus.REJECTED, 

103 reason="User said no", 

104 ) 

105 _, decision = hitl.request_and_decide( 

106 action="delete", risk_level=RiskLevel.CRITICAL 

107 ) 

108 assert decision.is_rejected 

109 

110 def test_history(self): 

111 hitl = HumanInTheLoop() 

112 hitl.callback = lambda r: ApprovalDecision( 

113 request_id=r.request_id, 

114 status=ApprovalStatus.APPROVED, 

115 ) 

116 hitl.request_and_decide(action="a1", risk_level=RiskLevel.HIGH) 

117 hitl.request_and_decide(action="a2", risk_level=RiskLevel.LOW) 

118 assert len(hitl.get_history()) == 2 

119 

120 def test_pending_queue(self): 

121 hitl = HumanInTheLoop(policy=ApprovalPolicy(require_approval_for_risk={ 

122 RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH, RiskLevel.CRITICAL 

123 })) 

124 req = hitl.request_approval(action="x", risk_level=RiskLevel.LOW) 

125 assert len(hitl.get_pending()) == 1 

126 

127 decision = ApprovalDecision(request_id=req.request_id, status=ApprovalStatus.APPROVED) 

128 hitl.decide(req.request_id, decision) 

129 assert len(hitl.get_pending()) == 0 

130 assert hitl.get_decision(req.request_id).is_approved 

131 

132 def test_approval_cache(self): 

133 hitl = HumanInTheLoop(policy=ApprovalPolicy(cache_approval_seconds=60)) 

134 hitl.callback = lambda r: ApprovalDecision( 

135 request_id=r.request_id, 

136 status=ApprovalStatus.APPROVED, 

137 ) 

138 # First call triggers callback 

139 req1, d1 = hitl.request_and_decide( 

140 action="read", tool_name="file.read", risk_level=RiskLevel.HIGH 

141 ) 

142 assert d1.is_approved 

143 # Second call should use cache (same tool+action) 

144 req2, d2 = hitl.request_and_decide( 

145 action="read", tool_name="file.read", risk_level=RiskLevel.HIGH 

146 ) 

147 assert d2.is_approved 

148 assert len(hitl.get_history()) == 2 

149 

150 def test_critical_blocked_automatically(self): 

151 policy = strict_approval_policy() 

152 hitl = HumanInTheLoop(policy=policy) 

153 # No callback set, critical risk with blocked domain 

154 req, decision = hitl.request_and_decide( 

155 action="format", 

156 risk_level=RiskLevel.CRITICAL, 

157 tool_name="delete.format_disk", 

158 ) 

159 assert decision.status == ApprovalStatus.REJECTED 

160 

161 def test_max_pending(self): 

162 policy = ApprovalPolicy( 

163 require_approval_for_risk={RiskLevel.LOW}, 

164 max_pending_requests=2, 

165 ) 

166 hitl = HumanInTheLoop(policy=policy) 

167 hitl.request_approval(action="a1", risk_level=RiskLevel.LOW) 

168 hitl.request_approval(action="a2", risk_level=RiskLevel.LOW) 

169 req3 = hitl.request_approval(action="a3", risk_level=RiskLevel.LOW) 

170 d = hitl.get_decision(req3.request_id) 

171 assert d.status == ApprovalStatus.REJECTED 

172 assert "Max pending" in d.reason 

173 

174 

175class TestApprovalPresets: 

176 def test_default(self): 

177 p = default_approval_policy() 

178 assert RiskLevel.HIGH in p.require_approval_for_risk 

179 assert RiskLevel.LOW not in p.require_approval_for_risk 

180 

181 def test_permissive(self): 

182 p = permissive_approval_policy() 

183 assert RiskLevel.CRITICAL in p.require_approval_for_risk 

184 assert RiskLevel.HIGH not in p.require_approval_for_risk 

185 

186 def test_strict(self): 

187 p = strict_approval_policy() 

188 assert RiskLevel.MEDIUM in p.require_approval_for_risk 

189 assert "delete" in p.block_domains