Spaces:
Runtime error
Runtime error
File size: 2,955 Bytes
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 97 98 99 100 101 | """Focused console bridge tests without TestClient transport."""
from __future__ import annotations
import json
from types import SimpleNamespace
from unittest.mock import patch
from open_range.protocols import SnapshotSpec
from open_range.server.app import create_app
from open_range.server.console import (
api_episode,
api_history,
api_snapshot,
clear_episode,
clear_history,
get_history,
)
from open_range.server.environment import RangeEnvironment
from open_range.server.models import RangeAction
_TEST_SNAPSHOT = SnapshotSpec(
topology={"hosts": ["attacker", "siem"]},
flags=[],
golden_path=[],
task={
"red_briefing": "Console bridge test mode.",
"blue_briefing": "Console bridge test mode.",
},
)
def _request(app):
return SimpleNamespace(app=app)
def _json_response_payload(response) -> dict | list:
return json.loads(response.body.decode())
def test_http_reset_publishes_console_snapshot_and_episode():
clear_episode()
clear_history()
with patch(
"open_range.server.environment.RangeEnvironment._select_snapshot",
return_value=_TEST_SNAPSHOT,
), patch(
"open_range.server.environment.RangeEnvironment._ensure_clean_reset_path",
):
app = create_app()
env = app.state.openenv_server._env_factory()
try:
env.reset(episode_id="http_console_ep")
finally:
env.close()
snapshot = _json_response_payload(_run(api_snapshot(_request(app))))
episode = _json_response_payload(_run(api_episode(_request(app))))
assert snapshot["id"] == "http_console_ep"
assert snapshot["hosts"] == ["attacker", "siem"]
assert snapshot["state_scope"] == "published_episode"
assert episode["step_count"] == 0
assert episode["mode"] == "red"
assert episode["state_scope"] == "published_episode"
def test_environment_reset_clears_history_and_records_reset():
clear_episode()
clear_history()
env = RangeEnvironment(docker_available=False)
env.reset(snapshot=_TEST_SNAPSHOT, episode_id="console_reset_ep")
history = get_history()
assert len(history) == 1
assert history[0]["command"] == "reset"
assert history[0]["mode"] == "system"
assert history[0]["episode_id"] == "console_reset_ep"
def test_environment_meta_steps_record_console_history():
clear_episode()
clear_history()
env = RangeEnvironment(docker_available=False)
env.reset(snapshot=_TEST_SNAPSHOT, episode_id="console_meta_ep")
env.step(RangeAction(command="submit_finding suspicious scan on web", mode="blue"))
history = _json_response_payload(_run(api_history()))
assert len(history) == 2
assert history[0]["command"] == "submit_finding suspicious scan on web"
assert history[0]["mode"] == "blue"
assert history[1]["command"] == "reset"
def _run(awaitable):
import asyncio
return asyncio.run(awaitable)
|