chenlin
f3a8a38acd
web-platform-ci / Standalone decision service (no cloud credentials) (push) Waiting to run
web-platform-ci / TypeScript, lint, unit, build (push) Waiting to run
web-platform-ci / Playwright E2E (push) Waiting to run
lekiwi-compatibility / cpu-compatibility (push) Waiting to run
web-platform-ci / Standalone decision service (no cloud credentials) (pull_request) Waiting to run
web-platform-ci / TypeScript, lint, unit, build (pull_request) Waiting to run
web-platform-ci / Playwright E2E (pull_request) Waiting to run
lekiwi-compatibility / cpu-compatibility (pull_request) Waiting to run
集成同源 BYOK 会话隔离、精简模型设置、官方订阅入口和 HTTPS 发布运维；保留本地训练/调参与控制能力。同步 npm 版本及 CHANGELOG，记录公网真实 API 验收仍待用户凭据。
66 lines
2.7 KiB
Python
66 lines
2.7 KiB
Python
import copy
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from mujoco_control_bridge import protocol as p
|
|
from mujoco_control_bridge.client import SimRobotClient
|
|
|
|
FIXTURE = json.loads(
|
|
(Path(__file__).resolve().parents[2] / "contracts/fixtures/single-joint.json").read_text()
|
|
)
|
|
|
|
|
|
class ProtocolTests(unittest.TestCase):
|
|
def test_shared_fixture(self):
|
|
descriptor = p.descriptor(FIXTURE["descriptor"])
|
|
p.observation(FIXTURE["observation"], descriptor)
|
|
action = p.action(FIXTURE["validAction"], descriptor)
|
|
p.same_identity(action, FIXTURE["identity"])
|
|
for invalid in FIXTURE["invalidActions"]:
|
|
with self.subTest(invalid["label"]), self.assertRaises(p.RobotError) as caught:
|
|
action = p.action({**FIXTURE["validAction"], **invalid["patch"]}, descriptor)
|
|
p.same_identity(action, FIXTURE["identity"])
|
|
self.assertEqual(caught.exception.code, invalid["code"])
|
|
|
|
def test_finite_clamp_and_extra_fields(self):
|
|
for value in (float("nan"), float("inf"), True, "0", 10**1000):
|
|
with self.subTest(value=str(value)[:20]), self.assertRaises(p.RobotError):
|
|
p.values({"slider.position": value}, FIXTURE["descriptor"]["actionChannels"])
|
|
self.assertEqual(
|
|
p.values({"slider.position": 20}, FIXTURE["descriptor"]["actionChannels"], True),
|
|
{"slider.position": 1},
|
|
)
|
|
for raw in ('{"x":1,"x":2}', '{"x":NaN}', '{"x":Infinity}', "["):
|
|
with self.assertRaises(p.RobotError):
|
|
p.loads(raw)
|
|
|
|
def test_unknown_units_capabilities_versions(self):
|
|
for mutation in ("unit", "frame", "capability", "fingerprint"):
|
|
descriptor = copy.deepcopy(FIXTURE["descriptor"])
|
|
if mutation == "unit":
|
|
descriptor["actionChannels"][0]["unit"] = "deg"
|
|
elif mutation == "frame":
|
|
descriptor["frame"] = "unknown"
|
|
elif mutation == "fingerprint":
|
|
descriptor["modelFingerprint"] = "not-sha"
|
|
else:
|
|
descriptor["capabilities"]["training"] = "true"
|
|
with self.subTest(mutation), self.assertRaises(p.RobotError):
|
|
p.descriptor(descriptor)
|
|
|
|
def test_only_local_endpoints_and_token_not_in_url(self):
|
|
for endpoint in (
|
|
"https://127.0.0.1",
|
|
"http://evil.test",
|
|
"http://user:token@127.0.0.1",
|
|
"http://localhost/?token=secret",
|
|
"http://localhost/path",
|
|
):
|
|
with self.subTest(endpoint), self.assertRaises(ValueError):
|
|
SimRobotClient(endpoint, token="test-only-not-a-real-secret")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|