chenlin
7519309a27
web-platform-ci / Standalone decision service (no cloud credentials) (push) Has been cancelled
web-platform-ci / TypeScript, lint, unit, build (push) Has been cancelled
web-platform-ci / Playwright E2E (push) Has been cancelled
lekiwi-compatibility / cpu-compatibility (push) Has been cancelled
76 lines
3.6 KiB
Python
76 lines
3.6 KiB
Python
import json
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from decision_server.protocol import LANGUAGE_SKILLS, LANGUAGE_VERSION
|
|
from decision_server.providers import jev
|
|
from decision_server.tests.test_service import observation
|
|
|
|
|
|
class JevContextTests(unittest.IsolatedAsyncioTestCase):
|
|
async def decide(self, candidate="close", phase="descend", failure="none", **changes):
|
|
obs = observation()
|
|
obs.update(version=LANGUAGE_VERSION, phase=phase)
|
|
values = (
|
|
dict(
|
|
choice=candidate,
|
|
grasp="empty",
|
|
diagnosis="none",
|
|
recovery="continue",
|
|
noul="allow",
|
|
score="partial",
|
|
reason="none",
|
|
)
|
|
| changes
|
|
)
|
|
upstream = {"answers": {k: {"choice": v} for k, v in values.items()}}
|
|
request = dict(observation=obs, candidates=[candidate, "stop"], failure=failure)
|
|
with patch.object(jev, "post", AsyncMock(return_value=upstream)) as post:
|
|
result, _ = await jev.decide(
|
|
None, SimpleNamespace(model="fixture", protocol="openrouter-decisions"), request
|
|
)
|
|
return result, post.call_args.args[3], request
|
|
|
|
async def test_next_skill_context_and_descriptive_criteria(self):
|
|
result, body, request = await self.decide()
|
|
state = json.loads(body["state"])
|
|
self.assertEqual(state["observation"], request["observation"])
|
|
self.assertEqual(state["candidates"], ["close", "stop"])
|
|
self.assertEqual(state["transition"], {"nextSkill": "close", "completedSkill": "descend"})
|
|
self.assertEqual(result["choice"], "close")
|
|
choice = body["questions"]["choice"]
|
|
self.assertEqual(set(choice["criteria"]), {"close", "stop"})
|
|
self.assertIn("0.012 m", choice["criteria"]["close"])
|
|
self.assertIn("Zero finger forces", choice["criteria"]["close"])
|
|
self.assertIn("freezes physics", choice["instructions"])
|
|
self.assertIn("NEXT", body["questions"]["noul"]["instructions"])
|
|
self.assertEqual(body["provider"], {"allow_fallbacks": False})
|
|
for skill in LANGUAGE_SKILLS:
|
|
self.assertGreater(len(jev.SKILL_CRITERIA[skill]), len(skill))
|
|
self.assertIn("secure=false is expected", jev.SKILL_CRITERIA["verify"])
|
|
self.assertIn("secure=true", jev.SKILL_CRITERIA["carry"])
|
|
self.assertIn("onGoalSupport=true", jev.SKILL_CRITERIA["release"])
|
|
|
|
async def test_initial_entry_and_failed_recovery_are_not_completed_skills(self):
|
|
for candidate, phase, failure in [
|
|
("stow", "stow", "none"),
|
|
("open", "close", "empty_grasp"),
|
|
("carry", "stow", "none"),
|
|
]:
|
|
_, body, _ = await self.decide(candidate, phase, failure)
|
|
self.assertIsNone(json.loads(body["state"])["transition"]["completedSkill"])
|
|
_, body, _ = await self.decide("approach", "stow")
|
|
self.assertEqual(json.loads(body["state"])["transition"]["completedSkill"], "stow")
|
|
|
|
async def test_never_rewrites_veto_or_stop_even_with_reason_none(self):
|
|
for changes in [dict(noul="deny"), dict(noul="uncertain"), dict(choice="stop")]:
|
|
with self.subTest(changes=changes):
|
|
result, _, _ = await self.decide(**changes)
|
|
for key, value in changes.items():
|
|
self.assertEqual(result[key], value)
|
|
self.assertEqual(result["reason"], "none")
|
|
|
|
def test_legacy_question_shape_is_unchanged(self):
|
|
self.assertEqual(jev.question(["ok"], "Choose ok.")["criteria"], {"ok": "ok"})
|