145 lines
4.9 KiB
Python
145 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
from core.models import Bounds, Scene, SceneElement
|
|
from runtime.executor import StepResult
|
|
from runtime.planner import PlannedStep
|
|
from semantic.models import SemanticScene, SemanticWidget
|
|
from world.config import WorldConfig, load_config
|
|
from world.model import WorldModel
|
|
|
|
|
|
def _scene() -> Scene:
|
|
return Scene(
|
|
width=10,
|
|
height=20,
|
|
elements=[
|
|
SceneElement(
|
|
id="send",
|
|
type="button",
|
|
text="Send",
|
|
bounds=Bounds(1, 2, 3, 4),
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _semantic_scene(page: str = "Chat") -> SemanticScene:
|
|
return SemanticScene(
|
|
page=page,
|
|
intents=["send a message"],
|
|
widgets=[SemanticWidget(element_id="send", purpose="send message")],
|
|
)
|
|
|
|
|
|
def _result(step: PlannedStep, *, success: bool = True) -> StepResult:
|
|
return StepResult(step=step, success=success, attempts=1)
|
|
|
|
|
|
def test_world_config_defaults_to_enabled_with_history_size_bound() -> None:
|
|
config = load_config({})
|
|
|
|
assert config.enabled is True
|
|
assert config.history_size == 10
|
|
|
|
|
|
def test_observe_updates_current_page_from_semantic_scene() -> None:
|
|
step = PlannedStep(action="tap", description="tap send")
|
|
model = WorldModel(config=WorldConfig(history_size=2))
|
|
|
|
model.observe(_scene(), _semantic_scene("Chat"), step, _result(step))
|
|
|
|
assert model.state.current_page == "Chat"
|
|
|
|
|
|
def test_observe_leaves_current_page_unchanged_without_semantic_scene() -> None:
|
|
step = PlannedStep(action="tap", description="tap send")
|
|
model = WorldModel(config=WorldConfig(history_size=2))
|
|
model.state.current_page = "Chat"
|
|
|
|
model.observe(_scene(), None, step, _result(step))
|
|
|
|
assert model.state.current_page == "Chat"
|
|
|
|
|
|
def test_observe_updates_current_app_for_successful_launch_and_clear_for_terminate() -> None:
|
|
model = WorldModel(config=WorldConfig(history_size=2))
|
|
launch = PlannedStep(
|
|
action="launch_app",
|
|
description="launch chat",
|
|
args={"app_id": "com.example.chat"},
|
|
)
|
|
terminate = PlannedStep(
|
|
action="terminate_app",
|
|
description="close chat",
|
|
args={"app_id": "com.example.chat"},
|
|
)
|
|
|
|
model.observe(_scene(), None, launch, _result(launch))
|
|
assert model.state.current_app == "com.example.chat"
|
|
|
|
model.observe(_scene(), None, terminate, _result(terminate))
|
|
assert model.state.current_app is None
|
|
|
|
|
|
def test_observe_does_not_update_current_app_on_failed_or_unrelated_step() -> None:
|
|
model = WorldModel(config=WorldConfig(history_size=3))
|
|
model.state.current_app = "com.example.chat"
|
|
failed_launch = PlannedStep(
|
|
action="launch_app",
|
|
description="launch other",
|
|
args={"app_id": "com.example.other"},
|
|
)
|
|
tap = PlannedStep(action="tap", description="tap", args={"x": 1, "y": 2})
|
|
|
|
model.observe(_scene(), None, failed_launch, _result(failed_launch, success=False))
|
|
model.observe(_scene(), None, tap, _result(tap))
|
|
|
|
assert model.state.current_app == "com.example.chat"
|
|
|
|
|
|
def test_observe_merges_explicit_remember_variables_only() -> None:
|
|
model = WorldModel(config=WorldConfig(history_size=3))
|
|
remember_step = PlannedStep(
|
|
action="tap",
|
|
description="select chat",
|
|
args={"remember": {"contact": "Zhang San"}},
|
|
)
|
|
plain_step = PlannedStep(action="tap", description="tap send", args={})
|
|
|
|
model.observe(_scene(), None, remember_step, _result(remember_step))
|
|
model.observe(_scene(), None, plain_step, _result(plain_step))
|
|
|
|
assert model.state.variables == {"contact": "Zhang San"}
|
|
|
|
|
|
def test_observe_skips_malformed_fields_without_raising() -> None:
|
|
model = WorldModel(config=WorldConfig(history_size=2))
|
|
missing_app = PlannedStep(action="launch_app", description="launch", args={})
|
|
bad_remember = PlannedStep(
|
|
action="tap",
|
|
description="tap",
|
|
args={"remember": "not-a-dict"},
|
|
)
|
|
|
|
model.observe(_scene(), None, missing_app, _result(missing_app))
|
|
model.observe(_scene(), None, bad_remember, _result(bad_remember))
|
|
|
|
assert model.state.current_app is None
|
|
assert model.state.variables == {}
|
|
assert len(model.state.history) == 2
|
|
|
|
|
|
def test_observe_appends_semantic_or_raw_scene_history_with_eviction() -> None:
|
|
model = WorldModel(config=WorldConfig(history_size=2))
|
|
first = PlannedStep(action="first", description="first")
|
|
second = PlannedStep(action="second", description="second")
|
|
third = PlannedStep(action="third", description="third")
|
|
|
|
model.observe(_scene(), _semantic_scene("Chat"), first, _result(first))
|
|
model.observe(_scene(), None, second, _result(second))
|
|
model.observe(_scene(), None, third, _result(third))
|
|
|
|
assert [event.action for event in model.state.history] == ["second", "third"]
|
|
assert model.state.history[0].scene_summary.to_dict() == _scene().to_dict()
|
|
assert model.state.history.maxlen == 2
|