Download tests/security/test_authorization_maturity.py from techprotrade/annator-command-center: direct link, hf CLI and curl.
- Browser
- Download file 31.3 kB
-
https://huggingface.co/spaces/techprotrade/annator-command-center/resolve/main/tests/security/test_authorization_maturity.py
- Command line
-
hf download hf://spaces/techprotrade/annator-command-center/tests/security/test_authorization_maturity.py
-
curl -L -o test_authorization_maturity.py https://huggingface.co/spaces/techprotrade/annator-command-center/resolve/main/tests/security/test_authorization_maturity.py
31.3 kB
| """ | |
| Agent Maturity Authorization Security Tests (SECU-02.1). | |
| Comprehensive tests for agent maturity level permissions: | |
| - STUDENT: Read-only (search, list, get, summarize, present_chart) | |
| - INTERN: Level 1-2 actions (draft, analyze, suggest, generate) | |
| - SUPERVISED: Level 1-3 actions (update, submit_form, send_email) | |
| - AUTONOMOUS: Level 1-4 actions (delete, execute, deploy, payment) | |
| These are CRITICAL security tests ensuring governance enforcement. | |
| Failure indicates a security vulnerability where agents can exceed permissions. | |
| """ | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from sqlalchemy.orm import Session | |
| from tests.factories.agent_factory import ( | |
| StudentAgentFactory, | |
| InternAgentFactory, | |
| SupervisedAgentFactory, | |
| AutonomousAgentFactory | |
| ) | |
| from tests.factories.user_factory import UserFactory, AdminUserFactory | |
| from core.models import AgentStatus, AgentRegistry | |
| from core.agent_governance_service import AgentGovernanceService | |
| from core.governance_cache import get_governance_cache | |
| # ============================================================================ | |
| # STUDENT Agent Tests - Read-Only Only | |
| # ============================================================================ | |
| class TestStudentAgentPermissions: | |
| """Test STUDENT agents can only perform read-only actions.""" | |
| def test_student_can_search(self, db_session: Session): | |
| """STUDENT agents can search (complexity 1, read-only).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "search") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| assert result["agent_status"] == AgentStatus.STUDENT.value | |
| def test_student_can_read(self, db_session: Session): | |
| """STUDENT agents can read (complexity 1, read-only).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "read") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_student_can_list(self, db_session: Session): | |
| """STUDENT agents can list (complexity 1, read-only).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "list") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_student_can_summarize(self, db_session: Session): | |
| """STUDENT agents can summarize (complexity 1, read-only).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "summarize") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_student_can_present_chart(self, db_session: Session): | |
| """STUDENT agents can present charts (complexity 1, read-only).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "present_chart") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_student_blocked_from_stream(self, db_session: Session): | |
| """STUDENT agents CANNOT stream (complexity 2).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "stream") | |
| assert result["allowed"] is False | |
| assert "lacks maturity" in result["reason"].lower() or "blocked" in result["reason"].lower() | |
| assert result["action_complexity"] == 2 | |
| def test_student_blocked_from_analyze(self, db_session: Session): | |
| """STUDENT agents CANNOT analyze (complexity 2).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "analyze") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 2 | |
| def test_student_blocked_from_draft(self, db_session: Session): | |
| """STUDENT agents CANNOT draft (complexity 2).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "draft") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 2 | |
| def test_student_blocked_from_update(self, db_session: Session): | |
| """STUDENT agents CANNOT update state (complexity 3).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "update") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_student_blocked_from_submit_form(self, db_session: Session): | |
| """STUDENT agents CANNOT submit forms (complexity 3).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "submit_form") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_student_blocked_from_send_email(self, db_session: Session): | |
| """STUDENT agents CANNOT send email (complexity 3).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "send_email") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_student_blocked_from_delete(self, db_session: Session): | |
| """STUDENT agents CANNOT delete (complexity 4 - CRITICAL).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "delete") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_student_blocked_from_execute(self, db_session: Session): | |
| """STUDENT agents CANNOT execute commands (complexity 4 - CRITICAL).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "execute") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_student_blocked_from_deploy(self, db_session: Session): | |
| """STUDENT agents CANNOT deploy (complexity 4 - CRITICAL).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "deploy") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_student_blocked_from_payment(self, db_session: Session): | |
| """STUDENT agents CANNOT process payments (complexity 4 - CRITICAL).""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "payment") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_student_confidence_score_range(self, db_session: Session): | |
| """STUDENT agents have confidence score < 0.5.""" | |
| student = StudentAgentFactory(_session=db_session) | |
| assert 0.0 <= student.confidence_score < 0.5 | |
| def test_student_cannot_bypass_governance_via_cache(self, db_session: Session): | |
| """Test STUDENT cannot bypass governance by poisoning cache. | |
| Note: Current governance cache implementation returns cached values | |
| without database verification. This test documents the behavior. | |
| In production, cache poisoning should be prevented via: | |
| 1. Cache signing/encryption | |
| 2. TTL expiration | |
| 3. Database verification for critical actions | |
| """ | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| cache = get_governance_cache() | |
| # First check - cache miss, checks database | |
| result1 = governance.can_perform_action(student.id, "delete") | |
| assert result1["allowed"] is False | |
| # Verify it was cached | |
| cached = cache.get(student.id, "delete") | |
| assert cached is not None | |
| assert cached["allowed"] is False | |
| # Even if we could poison the cache, the cache implementation | |
| # uses in-memory dict with no access control. This is a known | |
| # limitation mitigated by: | |
| # - Process isolation (separate worker processes) | |
| # - Cache TTL (automatic expiration) | |
| # - No external cache API (cache not exposed via endpoints) | |
| # Document: Cache poisoning requires code execution access | |
| # At that point, attacker has higher privileges anyway | |
| # ============================================================================ | |
| # INTERN Agent Tests - Level 1-2 Actions | |
| # ============================================================================ | |
| class TestInternAgentPermissions: | |
| """Test INTERN agents can perform level 1-2 actions.""" | |
| def test_intern_can_search(self, db_session: Session): | |
| """INTERN agents can search (complexity 1).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "search") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_intern_can_analyze(self, db_session: Session): | |
| """INTERN agents can analyze (complexity 2).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "analyze") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_can_suggest(self, db_session: Session): | |
| """INTERN agents can suggest (complexity 2).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "suggest") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_can_draft(self, db_session: Session): | |
| """INTERN agents can draft (complexity 2).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "draft") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_can_generate(self, db_session: Session): | |
| """INTERN agents can generate (complexity 2).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "generate") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_can_stream(self, db_session: Session): | |
| """INTERN agents can stream (complexity 2).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "stream") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_can_submit(self, db_session: Session): | |
| """INTERN agents can submit (complexity 2 - NOT submit_form).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "submit") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_intern_blocked_from_update(self, db_session: Session): | |
| """INTERN agents CANNOT update state (complexity 3).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "update") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_intern_blocked_from_submit_form(self, db_session: Session): | |
| """INTERN agents CANNOT submit forms (complexity 3).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "submit_form") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_intern_blocked_from_send_email(self, db_session: Session): | |
| """INTERN agents CANNOT send email (complexity 3).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "send_email") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 3 | |
| def test_intern_blocked_from_delete(self, db_session: Session): | |
| """INTERN agents CANNOT delete (complexity 4 - CRITICAL).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "delete") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_intern_blocked_from_execute(self, db_session: Session): | |
| """INTERN agents CANNOT execute (complexity 4 - CRITICAL).""" | |
| intern = InternAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(intern.id, "execute") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_intern_confidence_score_range(self, db_session: Session): | |
| """INTERN agents have confidence score 0.5-0.7.""" | |
| intern = InternAgentFactory(_session=db_session) | |
| assert 0.5 <= intern.confidence_score < 0.7 | |
| # ============================================================================ | |
| # SUPERVISED Agent Tests - Level 1-3 Actions | |
| # ============================================================================ | |
| class TestSupervisedAgentPermissions: | |
| """Test SUPERVISED agents can perform level 1-3 actions.""" | |
| def test_supervised_can_search(self, db_session: Session): | |
| """SUPERVISED agents can search (complexity 1).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "search") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_supervised_can_stream(self, db_session: Session): | |
| """SUPERVISED agents can stream (complexity 2).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "stream") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_supervised_can_analyze(self, db_session: Session): | |
| """SUPERVISED agents can analyze (complexity 2).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "analyze") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_supervised_can_update(self, db_session: Session): | |
| """SUPERVISED agents can update state (complexity 3).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "update") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_supervised_can_submit_form(self, db_session: Session): | |
| """SUPERVISED agents can submit forms (complexity 3).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "submit_form") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_supervised_can_send_email(self, db_session: Session): | |
| """SUPERVISED agents can send email (complexity 3).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "send_email") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_supervised_can_create(self, db_session: Session): | |
| """SUPERVISED agents can create (complexity 3).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "create") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_supervised_can_post_message(self, db_session: Session): | |
| """SUPERVISED agents can post messages (complexity 3).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "post_message") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_supervised_blocked_from_delete(self, db_session: Session): | |
| """SUPERVISED agents CANNOT delete (complexity 4 - CRITICAL).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "delete") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_supervised_blocked_from_execute(self, db_session: Session): | |
| """SUPERVISED agents CANNOT execute (complexity 4 - CRITICAL).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "execute") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_supervised_blocked_from_deploy(self, db_session: Session): | |
| """SUPERVISED agents CANNOT deploy (complexity 4 - CRITICAL).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "deploy") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_supervised_blocked_from_payment(self, db_session: Session): | |
| """SUPERVISED agents CANNOT process payments (complexity 4 - CRITICAL).""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(supervised.id, "payment") | |
| assert result["allowed"] is False | |
| assert result["action_complexity"] == 4 | |
| def test_supervised_confidence_score_range(self, db_session: Session): | |
| """SUPERVISED agents have confidence score 0.7-0.9.""" | |
| supervised = SupervisedAgentFactory(_session=db_session) | |
| assert 0.7 <= supervised.confidence_score < 0.9 | |
| # ============================================================================ | |
| # AUTONOMOUS Agent Tests - Full Execution (Level 1-4) | |
| # ============================================================================ | |
| class TestAutonomousAgentPermissions: | |
| """Test AUTONOMOUS agents have full execution权限.""" | |
| def test_autonomous_can_search(self, db_session: Session): | |
| """AUTONOMOUS agents can search (complexity 1).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "search") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 1 | |
| def test_autonomous_can_stream(self, db_session: Session): | |
| """AUTONOMOUS agents can stream (complexity 2).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "stream") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_autonomous_can_analyze(self, db_session: Session): | |
| """AUTONOMOUS agents can analyze (complexity 2).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "analyze") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 2 | |
| def test_autonomous_can_update(self, db_session: Session): | |
| """AUTONOMOUS agents can update (complexity 3).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "update") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_autonomous_can_delete(self, db_session: Session): | |
| """AUTONOMOUS agents can delete (complexity 4 - CRITICAL).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "delete") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 4 | |
| def test_autonomous_can_execute(self, db_session: Session): | |
| """AUTONOMOUS agents can execute (complexity 4 - CRITICAL).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "execute") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 4 | |
| def test_autonomous_can_deploy(self, db_session: Session): | |
| """AUTONOMOUS agents can deploy (complexity 4 - CRITICAL).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "deploy") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 4 | |
| def test_autonomous_can_payment(self, db_session: Session): | |
| """AUTONOMOUS agents can process payments (complexity 4 - CRITICAL).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "payment") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 4 | |
| def test_autonomous_can_send_email(self, db_session: Session): | |
| """AUTONOMOUS agents can send email (complexity 3).""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "send_email") | |
| assert result["allowed"] is True | |
| assert result["action_complexity"] == 3 | |
| def test_autonomous_confidence_score_range(self, db_session: Session): | |
| """AUTONOMOUS agents have confidence score >= 0.9.""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| assert 0.9 <= autonomous.confidence_score <= 1.0 | |
| # ============================================================================ | |
| # Maturity Promotion/Regression Tests | |
| # ============================================================================ | |
| class TestMaturityTransitions: | |
| """Test maturity level changes invalidate cache.""" | |
| def test_promotion_student_to_intern(self, db_session: Session): | |
| """Test promoting STUDENT to INTERN grants new permissions.""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| cache = get_governance_cache() | |
| # Initially blocked from streaming | |
| result1 = governance.can_perform_action(student.id, "stream") | |
| assert result1["allowed"] is False | |
| # Promote to INTERN | |
| student.status = AgentStatus.INTERN.value | |
| student.confidence_score = 0.6 | |
| db_session.commit() | |
| # Invalidate cache | |
| cache.invalidate(student.id) | |
| # Now allowed to stream | |
| result2 = governance.can_perform_action(student.id, "stream") | |
| assert result2["allowed"] is True | |
| def test_regression_autonomous_to_supervised(self, db_session: Session): | |
| """Test demoting AUTONOMOUS to SUPERVISED revokes critical permissions.""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| cache = get_governance_cache() | |
| # Initially allowed to delete | |
| result1 = governance.can_perform_action(autonomous.id, "delete") | |
| assert result1["allowed"] is True | |
| # Demote to SUPERVISED | |
| autonomous.status = AgentStatus.SUPERVISED.value | |
| autonomous.confidence_score = 0.8 | |
| db_session.commit() | |
| # Invalidate cache | |
| cache.invalidate(autonomous.id) | |
| # Now blocked from delete | |
| result2 = governance.can_perform_action(autonomous.id, "delete") | |
| assert result2["allowed"] is False | |
| def test_multiple_maturity_changes(self, db_session: Session): | |
| """Test multiple maturity transitions maintain correct permissions.""" | |
| agent = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| cache = get_governance_cache() | |
| # STUDENT: blocked from analyze | |
| result = governance.can_perform_action(agent.id, "analyze") | |
| assert result["allowed"] is False | |
| # Promote to INTERN | |
| agent.status = AgentStatus.INTERN.value | |
| agent.confidence_score = 0.6 | |
| db_session.commit() | |
| cache.invalidate(agent.id) | |
| # INTERN: allowed to analyze | |
| result = governance.can_perform_action(agent.id, "analyze") | |
| assert result["allowed"] is True | |
| # Promote to SUPERVISED | |
| agent.status = AgentStatus.SUPERVISED.value | |
| agent.confidence_score = 0.8 | |
| db_session.commit() | |
| cache.invalidate(agent.id) | |
| # SUPERVISED: still allowed to analyze | |
| result = governance.can_perform_action(agent.id, "analyze") | |
| assert result["allowed"] is True | |
| # Promote to AUTONOMOUS | |
| agent.status = AgentStatus.AUTONOMOUS.value | |
| agent.confidence_score = 0.95 | |
| db_session.commit() | |
| cache.invalidate(agent.id) | |
| # AUTONOMOUS: still allowed to analyze | |
| result = governance.can_perform_action(agent.id, "analyze") | |
| assert result["allowed"] is True | |
| # ============================================================================ | |
| # Edge Cases and Boundary Tests | |
| # ============================================================================ | |
| class TestMaturityEdgeCases: | |
| """Test edge cases and boundary conditions.""" | |
| def test_nonexistent_agent(self, db_session: Session): | |
| """Test governance check for non-existent agent.""" | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action("nonexistent-agent-id", "delete") | |
| assert result["allowed"] is False | |
| assert "not found" in result["reason"].lower() | |
| def test_invalid_action_type(self, db_session: Session): | |
| """Test governance check for unknown action type.""" | |
| autonomous = AutonomousAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(autonomous.id, "unknown_action_xyz") | |
| # Should handle gracefully (default to blocking or assume safe) | |
| assert "allowed" in result | |
| def test_confidence_score_boundaries(self, db_session: Session): | |
| """Test confidence score boundaries match maturity levels.""" | |
| from tests.factories.agent_factory import AgentFactory | |
| # Test boundary at 0.5 (STUDENT/INTERN) | |
| agent1 = AgentFactory.create( | |
| confidence_score=0.49, | |
| status=AgentStatus.STUDENT.value, | |
| _session=db_session | |
| ) | |
| assert agent1.confidence_score < 0.5 | |
| agent2 = AgentFactory.create( | |
| confidence_score=0.50, | |
| status=AgentStatus.INTERN.value, | |
| _session=db_session | |
| ) | |
| assert agent2.confidence_score >= 0.5 | |
| # Test boundary at 0.7 (INTERN/SUPERVISED) | |
| agent3 = AgentFactory.create( | |
| confidence_score=0.69, | |
| status=AgentStatus.INTERN.value, | |
| _session=db_session | |
| ) | |
| assert agent3.confidence_score < 0.7 | |
| agent4 = AgentFactory.create( | |
| confidence_score=0.70, | |
| status=AgentStatus.SUPERVISED.value, | |
| _session=db_session | |
| ) | |
| assert agent4.confidence_score >= 0.7 | |
| # Test boundary at 0.9 (SUPERVISED/AUTONOMOUS) | |
| agent5 = AgentFactory.create( | |
| confidence_score=0.89, | |
| status=AgentStatus.SUPERVISED.value, | |
| _session=db_session | |
| ) | |
| assert agent5.confidence_score < 0.9 | |
| agent6 = AgentFactory.create( | |
| confidence_score=0.90, | |
| status=AgentStatus.AUTONOMOUS.value, | |
| _session=db_session | |
| ) | |
| assert agent6.confidence_score >= 0.9 | |
| def test_governance_reason_provided(self, db_session: Session): | |
| """Test governance checks provide reason for denial.""" | |
| student = StudentAgentFactory(_session=db_session) | |
| governance = AgentGovernanceService(db_session) | |
| result = governance.can_perform_action(student.id, "delete") | |
| assert "reason" in result | |
| assert len(result["reason"]) > 0 | |
| # Reason should mention maturity or permissions | |
| reason_lower = result["reason"].lower() | |
| assert any(term in reason_lower for term in ["maturity", "permission", "blocked", "lacks"]) | |