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 ApprovalDecision, 

7 ApprovalPolicy, 

8 ApprovalRequest, 

9 ApprovalStatus, 

10 HumanInTheLoop, 

11 RiskLevel, 

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( 

42 request_id="x", status=ApprovalStatus.MODIFIED, modified_args={"force": True} 

43 ) 

44 assert d.is_approved 

45 

46 def test_rejected(self): 

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

48 assert d.is_rejected 

49 assert not d.is_approved 

50 

51 

52class TestHumanInTheLoop: 

53 def test_low_risk_auto_skipped(self): 

54 hitl = HumanInTheLoop(policy=default_approval_policy()) 

55 req, decision = hitl.request_and_decide( 

56 action="read_file", 

57 description="Read config.yaml", 

58 risk_level=RiskLevel.LOW, 

59 tool_name="file.read", 

60 ) 

61 assert decision.status == ApprovalStatus.SKIPPED 

62 

63 def test_high_risk_needs_approval(self): 

64 hitl = HumanInTheLoop(policy=default_approval_policy()) 

65 hitl.callback = lambda r: ApprovalDecision( 

66 request_id=r.request_id, 

67 status=ApprovalStatus.APPROVED, 

68 reason="OK", 

69 ) 

70 req, decision = hitl.request_and_decide( 

71 action="delete_all", 

72 description="Delete production database", 

73 risk_level=RiskLevel.HIGH, 

74 tool_name="db.drop", 

75 ) 

76 assert decision.is_approved 

77 

78 def test_auto_approve_domain(self): 

79 policy = permissive_approval_policy() 

80 hitl = HumanInTheLoop(policy=policy) 

81 req, decision = hitl.request_and_decide( 

82 action="search", 

83 description="Search web", 

84 risk_level=RiskLevel.MEDIUM, 

85 tool_name="read.web_search", 

86 ) 

87 assert decision.status == ApprovalStatus.APPROVED 

88 

89 def test_blocked_domain(self): 

90 policy = strict_approval_policy() 

91 hitl = HumanInTheLoop(policy=policy) 

92 req, decision = hitl.request_and_decide( 

93 action="format", 

94 description="Format disk", 

95 risk_level=RiskLevel.CRITICAL, 

96 tool_name="delete.format_disk", 

97 ) 

98 assert decision.status == ApprovalStatus.REJECTED 

99 

100 def test_rejected_decision(self): 

101 hitl = HumanInTheLoop() 

102 hitl.callback = lambda r: ApprovalDecision( 

103 request_id=r.request_id, 

104 status=ApprovalStatus.REJECTED, 

105 reason="User said no", 

106 ) 

107 _, decision = hitl.request_and_decide(action="delete", risk_level=RiskLevel.CRITICAL) 

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( 

122 policy=ApprovalPolicy( 

123 require_approval_for_risk={ 

124 RiskLevel.LOW, 

125 RiskLevel.MEDIUM, 

126 RiskLevel.HIGH, 

127 RiskLevel.CRITICAL, 

128 } 

129 ) 

130 ) 

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

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

133 

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

135 hitl.decide(req.request_id, decision) 

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

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

138 

139 def test_approval_cache(self): 

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

141 hitl.callback = lambda r: ApprovalDecision( 

142 request_id=r.request_id, 

143 status=ApprovalStatus.APPROVED, 

144 ) 

145 # First call triggers callback 

146 req1, d1 = hitl.request_and_decide( 

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

148 ) 

149 assert d1.is_approved 

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

151 req2, d2 = hitl.request_and_decide( 

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

153 ) 

154 assert d2.is_approved 

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

156 

157 def test_critical_blocked_automatically(self): 

158 policy = strict_approval_policy() 

159 hitl = HumanInTheLoop(policy=policy) 

160 # No callback set, critical risk with blocked domain 

161 req, decision = hitl.request_and_decide( 

162 action="format", 

163 risk_level=RiskLevel.CRITICAL, 

164 tool_name="delete.format_disk", 

165 ) 

166 assert decision.status == ApprovalStatus.REJECTED 

167 

168 def test_max_pending(self): 

169 policy = ApprovalPolicy( 

170 require_approval_for_risk={RiskLevel.LOW}, 

171 max_pending_requests=2, 

172 ) 

173 hitl = HumanInTheLoop(policy=policy) 

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

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

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

177 d = hitl.get_decision(req3.request_id) 

178 assert d.status == ApprovalStatus.REJECTED 

179 assert "Max pending" in d.reason 

180 

181 

182class TestApprovalPresets: 

183 def test_default(self): 

184 p = default_approval_policy() 

185 assert RiskLevel.HIGH in p.require_approval_for_risk 

186 assert RiskLevel.LOW not in p.require_approval_for_risk 

187 

188 def test_permissive(self): 

189 p = permissive_approval_policy() 

190 assert RiskLevel.CRITICAL in p.require_approval_for_risk 

191 assert RiskLevel.HIGH not in p.require_approval_for_risk 

192 

193 def test_strict(self): 

194 p = strict_approval_policy() 

195 assert RiskLevel.MEDIUM in p.require_approval_for_risk 

196 assert "delete" in p.block_domains