Spaces:
Runtime error
Runtime error
File size: 3,312 Bytes
86c1c19 e37c7b1 86c1c19 e37c7b1 86c1c19 e37c7b1 86c1c19 e37c7b1 86c1c19 e37c7b1 86c1c19 e37c7b1 | 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 | """Unit tests for console environment context resolution."""
from __future__ import annotations
from types import SimpleNamespace
from open_range.protocols import SnapshotSpec
from open_range.server.console import _get_env_context, clear_episode, publish_episode
from open_range.server.environment import RangeEnvironment
from open_range.server.models import RangeState
class _Req:
def __init__(self, app):
self.app = app
def _app_with_state(**kwargs):
return SimpleNamespace(state=SimpleNamespace(**kwargs))
def test_prefers_active_websocket_session_env():
clear_episode()
fallback_env = RangeEnvironment(docker_available=False)
ws_env = RangeEnvironment(docker_available=False)
server = SimpleNamespace(
_sessions={"session_a": ws_env},
_session_info={"session_a": SimpleNamespace(last_activity_at=10.0)},
)
request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
ctx = _get_env_context(request)
assert ctx["env"] is ws_env
assert ctx["state_scope"] == "websocket_session"
assert ctx["session_id"] == "session_a"
assert ctx["warning"] is None
def test_uses_app_state_env_when_no_active_session():
clear_episode()
fallback_env = RangeEnvironment(docker_available=False)
server = SimpleNamespace(_sessions={}, _session_info={})
request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
ctx = _get_env_context(request)
assert ctx["env"] is fallback_env
assert ctx["state_scope"] == "app_state_env"
assert ctx["session_id"] is None
assert isinstance(ctx["warning"], str) and ctx["warning"]
def test_multiple_sessions_selects_most_recent_and_warns():
clear_episode()
older_env = RangeEnvironment(docker_available=False)
newer_env = RangeEnvironment(docker_available=False)
server = SimpleNamespace(
_sessions={"old": older_env, "new": newer_env},
_session_info={
"old": SimpleNamespace(last_activity_at=10.0),
"new": SimpleNamespace(last_activity_at=20.0),
},
)
request = _Req(_app_with_state(openenv_server=server))
ctx = _get_env_context(request)
assert ctx["env"] is newer_env
assert ctx["state_scope"] == "websocket_session"
assert ctx["session_id"] == "new"
assert "active sessions" in (ctx["warning"] or "").lower()
def test_uses_published_episode_before_app_state_fallback():
clear_episode()
snapshot = SnapshotSpec(
topology={"hosts": ["attacker"], "zones": {"dmz": ["web"]}, "tier": 2},
flags=[],
golden_path=[],
task={"red_briefing": "r", "blue_briefing": "b"},
)
publish_episode(
snapshot,
RangeState(episode_id="published_ep", step_count=4, mode="blue", tier=2),
)
fallback_env = RangeEnvironment(docker_available=False)
server = SimpleNamespace(_sessions={}, _session_info={})
request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
ctx = _get_env_context(request)
assert ctx["env"] is None
assert ctx["published_episode"]["snapshot"]["id"] == "published_ep"
assert ctx["state_scope"] == "published_episode"
assert ctx["session_id"] is None
assert "most recent reset/step state" in (ctx["warning"] or "")
clear_episode()
|