File size: 3,403 Bytes
4058302 | 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 | """Environment-level integration tests (require openenv installed)."""
from __future__ import annotations
import importlib
import pytest
openenv = pytest.importorskip(
"openenv.core.env_server",
reason="openenv-core not installed; skipping environment tests.",
)
environment_module = importlib.import_module("server.environment")
models_module = importlib.import_module("models")
IncidentCommandCenterEnvironment = environment_module.IncidentCommandCenterEnvironment
IncidentAction = models_module.IncidentAction
def test_reset_returns_valid_observation() -> None:
env = IncidentCommandCenterEnvironment()
obs = env.reset(task_name="easy", seed=123)
assert obs.done is False
assert obs.incident_id
assert obs.budget_remaining > 0
assert obs.sla_minutes_remaining > 0
assert "inspect_logs" in obs.available_actions
assert obs.investigation_targets
assert obs.customer_tier in {"free", "standard", "premium", "enterprise"}
def test_reset_is_seeded_deterministic() -> None:
env = IncidentCommandCenterEnvironment()
a = env.reset(task_name="medium", seed=7)
b = env.reset(task_name="medium", seed=7)
assert a.incident_id == b.incident_id
assert a.investigation_targets == b.investigation_targets
def test_inspect_logs_step_returns_reward_components() -> None:
env = IncidentCommandCenterEnvironment()
obs = env.reset(task_name="easy", seed=1)
log_target = next(iter(obs.investigation_targets.get("logs", []) or [""]))
result = env.step(
IncidentAction(
actor="triage_agent",
action_type="inspect_logs",
target=log_target or "payments-api",
)
)
assert isinstance(result.reward_components, dict)
assert "step_cost" in result.reward_components
def test_wrong_actor_incurs_penalty() -> None:
env = IncidentCommandCenterEnvironment()
env.reset(task_name="easy", seed=1)
res = env.step(
IncidentAction(
actor="triage_agent",
action_type="close_incident",
root_cause="unknown",
)
)
assert res.reward_components.get("wrong_actor_penalty", 0.0) < 0
def test_budget_exhaustion_terminates_episode() -> None:
env = IncidentCommandCenterEnvironment()
env.reset(task_name="easy", seed=2)
done = False
steps = 0
while not done and steps < 200:
res = env.step(
IncidentAction(actor="triage_agent", action_type="inspect_logs", target="foo")
)
done = bool(res.done)
steps += 1
assert done, "Episode should terminate when budget/SLA is exhausted"
def test_close_correct_root_cause_awards_positive_reward() -> None:
env = IncidentCommandCenterEnvironment()
obs = env.reset(task_name="easy", seed=3)
incident = env._incidents[env.state.current_incident_index] # type: ignore[attr-defined]
expected_root_cause = incident.root_cause
env.step(
IncidentAction(
actor="investigator_agent",
action_type="apply_fix",
resolution_summary=" ".join(incident.accepted_fix_keywords[0]),
)
)
res = env.step(
IncidentAction(
actor="ops_manager_agent",
action_type="close_incident",
root_cause=expected_root_cause,
)
)
assert any(v > 0 for v in res.reward_components.values()), res.reward_components
|