Spaces:
Sleeping
Sleeping
File size: 5,826 Bytes
9c1645f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """
Tests for SupportOpsArena environment core functionality.
"""
import pytest
from env.environment import SupportOpsArena
from env.actions import ActionType
from env.state import TaskLevel
from adversary.adversary import AdaptiveAdversary
class TestEnvironmentBasics:
"""Test basic environment functionality."""
@pytest.mark.asyncio
async def test_environment_initialization(self):
"""Test environment can be created."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
assert env is not None
@pytest.mark.asyncio
async def test_reset_easy(self):
"""Test reset for easy task."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
obs = await env.reset(task_level="easy", seed=42)
assert obs.task_level == TaskLevel.EASY
assert obs.step_count == 0
assert len(obs.system_logs) > 0
assert obs.escalation_allowed == False
@pytest.mark.asyncio
async def test_reset_medium(self):
"""Test reset for medium task."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
obs = await env.reset(task_level="medium", seed=42)
assert obs.task_level == TaskLevel.MEDIUM
assert obs.step_count == 0
@pytest.mark.asyncio
async def test_reset_hard(self):
"""Test reset for hard task."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
obs = await env.reset(task_level="hard", seed=42)
assert obs.task_level == TaskLevel.HARD
assert obs.step_count == 0
@pytest.mark.asyncio
async def test_step_basic_action(self):
"""Test taking a basic step."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
obs = await env.reset(task_level="easy", seed=42)
result = await env.step(ActionType.INSPECT_LOGS)
assert result.observation.step_count == 1
assert result.reward != 0.0 # Should get diagnostic reward
assert result.done == False
@pytest.mark.asyncio
async def test_invalid_task_level(self):
"""Test reset with invalid task level raises error."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
with pytest.raises(ValueError):
await env.reset(task_level="invalid")
@pytest.mark.asyncio
async def test_episode_done_on_resolve(self):
"""Test episode ends when ticket is resolved."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
await env.reset(task_level="easy", seed=42)
# Take a few diagnostic actions
await env.step(ActionType.INSPECT_LOGS)
await env.step(ActionType.CHECK_AUTHENTICATION)
await env.step(ActionType.INSPECT_NETWORK)
# Resolve ticket
result = await env.step(ActionType.RESOLVE_TICKET)
assert result.done == True
assert "root_cause" in result.info
@pytest.mark.asyncio
async def test_state_method(self):
"""Test state() method returns full state."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
await env.reset(task_level="easy", seed=42)
state = await env.state()
assert state.observation is not None
assert state.hidden is not None
assert state.episode_id is not None
@pytest.mark.asyncio
async def test_escalation_unlock(self):
"""Test escalation is unlocked after enough diagnostics."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
obs = await env.reset(task_level="easy", seed=42)
assert obs.escalation_allowed == False
# Take 3 diagnostic actions (unlock threshold for easy)
result1 = await env.step(ActionType.INSPECT_LOGS)
result2 = await env.step(ActionType.CHECK_AUTHENTICATION)
result3 = await env.step(ActionType.INSPECT_NETWORK)
# Should now be unlocked
assert result3.observation.escalation_allowed == True
@pytest.mark.asyncio
async def test_step_without_reset_raises_error(self):
"""Test that step without reset raises error."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
with pytest.raises(RuntimeError):
await env.step(ActionType.INSPECT_LOGS)
class TestRewardSystem:
"""Test reward calculation."""
@pytest.mark.asyncio
async def test_diagnostic_action_reward(self):
"""Test diagnostic actions give positive reward."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
await env.reset(task_level="easy", seed=42)
result = await env.step(ActionType.INSPECT_LOGS)
assert result.reward > 0 # Should get diagnostic bonus
assert "DIAGNOSTIC_ACTION" in str(result.info.get("reward_breakdown", {}))
@pytest.mark.asyncio
async def test_redundant_action_penalty(self):
"""Test redundant actions are penalized."""
adversary = AdaptiveAdversary()
env = SupportOpsArena(adversary=adversary)
await env.reset(task_level="easy", seed=42)
# Take same action twice
result1 = await env.step(ActionType.INSPECT_LOGS)
result2 = await env.step(ActionType.INSPECT_LOGS)
# Second time should have penalty
assert "REDUNDANT_ACTION" in str(result2.info.get("reward_breakdown", {}))
|