File size: 4,199 Bytes
06c11b0 d10d370 | 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 | from __future__ import annotations
import json
def _write_metadata(root, env_id: str, episodes: list[int]) -> None:
root.mkdir(parents=True, exist_ok=True)
payload = {
"env_id": env_id,
"records": [
{"task": env_id, "episode": ep, "seed": 1000 + ep, "difficulty": "easy"}
for ep in episodes
],
}
(root / f"record_dataset_{env_id}_metadata.json").write_text(
json.dumps(payload), encoding="utf-8"
)
def test_fixed_users_login_and_random_task_pool(monkeypatch, reload_module, tmp_path):
metadata_root = tmp_path / "metadata"
_write_metadata(metadata_root, "EnvA", [0, 1, 2])
_write_metadata(metadata_root, "EnvB", [10, 11])
monkeypatch.setenv("ROBOMME_METADATA_ROOT", str(metadata_root))
user_manager_mod = reload_module("user_manager")
monkeypatch.setattr(user_manager_mod.random, "choice", lambda seq: seq[0])
manager = user_manager_mod.UserManager()
success, _msg, status = manager.init_session("uid1")
assert success
assert status["current_task"]["env_id"] in {"EnvA", "EnvB"}
assert status["current_task"]["episode_idx"] in {0, 1, 2, 10, 11}
assert status["is_done_all"] is False
def test_switch_env_and_next_episode_stays_in_same_env(monkeypatch, reload_module, tmp_path):
metadata_root = tmp_path / "metadata"
_write_metadata(metadata_root, "EnvA", [0, 1, 2])
_write_metadata(metadata_root, "EnvB", [10, 11])
monkeypatch.setenv("ROBOMME_METADATA_ROOT", str(metadata_root))
user_manager_mod = reload_module("user_manager")
monkeypatch.setattr(user_manager_mod.random, "choice", lambda seq: seq[-1])
manager = user_manager_mod.UserManager()
success, _msg, _status = manager.init_session("uid2")
assert success
switched = manager.switch_env_and_random_episode("uid2", "EnvA")
assert switched is not None
assert switched["current_task"]["env_id"] == "EnvA"
assert switched["current_task"]["episode_idx"] in {0, 1, 2}
nxt = manager.next_episode_same_env("uid2")
assert nxt is not None
assert nxt["current_task"]["env_id"] == "EnvA"
assert nxt["current_task"]["episode_idx"] in {0, 1, 2}
def test_complete_current_task_increments_completed_count(monkeypatch, reload_module, tmp_path):
metadata_root = tmp_path / "metadata"
_write_metadata(metadata_root, "EnvA", [0, 1])
monkeypatch.setenv("ROBOMME_METADATA_ROOT", str(metadata_root))
user_manager_mod = reload_module("user_manager")
monkeypatch.setattr(user_manager_mod.random, "choice", lambda seq: seq[0])
manager = user_manager_mod.UserManager()
success, _msg, status = manager.init_session("uid3")
assert success
assert status["completed_count"] == 0
updated = manager.complete_current_task(
"uid3",
env_id=status["current_task"]["env_id"],
episode_idx=status["current_task"]["episode_idx"],
status="success",
)
assert updated is not None
assert updated["completed_count"] == 1
assert updated["is_done_all"] is False
def test_init_session_fails_when_metadata_root_missing(monkeypatch, reload_module, tmp_path):
missing_root = tmp_path / "missing-metadata-root"
monkeypatch.setenv("ROBOMME_METADATA_ROOT", str(missing_root))
user_manager_mod = reload_module("user_manager")
manager = user_manager_mod.UserManager()
success, msg, status = manager.init_session("uid-missing")
assert success is False
assert "No available environments" in msg
assert status is None
def test_env_choices_follow_task_name_list_order(monkeypatch, reload_module, tmp_path):
metadata_root = tmp_path / "metadata"
_write_metadata(metadata_root, "VideoPlaceButton", [0])
_write_metadata(metadata_root, "BinFill", [0])
_write_metadata(metadata_root, "PatternLock", [0])
_write_metadata(metadata_root, "StopCube", [0])
monkeypatch.setenv("ROBOMME_METADATA_ROOT", str(metadata_root))
user_manager_mod = reload_module("user_manager")
manager = user_manager_mod.UserManager()
assert manager.env_choices == [
"BinFill",
"StopCube",
"VideoPlaceButton",
"PatternLock",
]
|