feat(pose_3d): POST /unlock proxy + non-regression check
This commit is contained in:
1 parent
feaab2aae7
commit
0ab0e5691c
2 files changed
+32
No files matched your search
@@ -5,6 +5,7 @@ import logging
|
|||||||
import re
|
import re
|
||||||
from typing import Any, Literal, Optional
|
from typing import Any, Literal, Optional
|
||||||
|
|
||||||
|
import httpx
|
||||||
from fastapi import APIRouter, HTTPException
|
from fastapi import APIRouter, HTTPException
|
||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
@@ -171,3 +172,22 @@ def delete_pose(name: str) -> dict[str, Any]:
|
|||||||
del poses[name]
|
del poses[name]
|
||||||
set_config_value(["pose_3d", "named_poses"], poses)
|
set_config_value(["pose_3d", "named_poses"], poses)
|
||||||
return {"deleted": name}
|
return {"deleted": name}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/unlock")
|
||||||
|
def post_unlock() -> dict[str, Any]:
|
||||||
|
"""Proxy a POST {bridge}/unlock. Recovery rapido da arms: error_timeout."""
|
||||||
|
cfg = get_config()
|
||||||
|
key = _deps.active_robot_key()
|
||||||
|
base = ((cfg.get("services", {}) or {}).get(key, {}) or {}).get("base_url")
|
||||||
|
if not base:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=502,
|
||||||
|
detail=f"services.{key}.base_url non configurato",
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
r = httpx.post(f"{base.rstrip('/')}/unlock", timeout=3.0)
|
||||||
|
r.raise_for_status()
|
||||||
|
except httpx.HTTPError as e:
|
||||||
|
raise HTTPException(status_code=502, detail=f"bridge unlock: {e}")
|
||||||
|
return r.json()
|
||||||
@@ -187,3 +187,15 @@ def test_go_pose_executes_via_g1_with_stored_values():
|
|||||||
assert captured["side"] == "left"
|
assert captured["side"] == "left"
|
||||||
assert captured["position"] == [0.11, 0.22, 0.33]
|
assert captured["position"] == [0.11, 0.22, 0.33]
|
||||||
assert captured["precise"] is True
|
assert captured["precise"] is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_unlock_proxies_to_bridge():
|
||||||
|
def _post(url, *args, **kw):
|
||||||
|
assert url.endswith("/unlock")
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.json.return_value = {"ok": True, "message": "unlocked"}
|
||||||
|
return resp
|
||||||
|
with patch("presa_mano.api.pose_3d_routes.httpx.post", side_effect=_post):
|
||||||
|
r = client.post("/api/pose_3d/unlock")
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert r.json()["ok"] is True
|
||||||
Reference in new issue
Block a user