233 lines
9.0 KiB
Python
233 lines
9.0 KiB
Python
import asyncio
|
|
from contextlib import asynccontextmanager
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from lingkong_motor_service.api import create_app
|
|
from lingkong_motor_service.config import Settings
|
|
from lingkong_motor_service.controller import MotorCommunicationError
|
|
from lingkong_motor_service.protocol import CONTROL_FRAME_ID, MIXED_COMMAND_FRAME_ID
|
|
|
|
from .fakes import FakeTransport
|
|
|
|
|
|
@asynccontextmanager
|
|
async def client_for(app):
|
|
async with app.router.lifespan_context(app):
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
yield client
|
|
|
|
|
|
def test_hardware_gate_is_closed_by_default() -> None:
|
|
async def scenario() -> None:
|
|
app = create_app(Settings())
|
|
|
|
async with client_for(app) as client:
|
|
assert (await client.get("/healthz")).json() == {
|
|
"status": "ok",
|
|
"hardware_enabled": False,
|
|
"can_connected": False,
|
|
}
|
|
assert (await client.get("/readyz")).status_code == 503
|
|
response = await client.post("/v1/motor/open")
|
|
assert (await client.post("/v1/motor/enable")).status_code == 404
|
|
assert (await client.post("/v1/motor/disable")).status_code == 404
|
|
openapi = (await client.get("/openapi.json")).json()
|
|
assert "/v1/motor/enable" not in openapi["paths"]
|
|
assert "/v1/motor/disable" not in openapi["paths"]
|
|
|
|
assert response.status_code == 409
|
|
assert "MOTOR_HARDWARE_ENABLED=true" in response.json()["detail"]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_hardware_service_auto_enables_and_shutdown_stops_then_disables() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
app = create_app(
|
|
Settings(hardware_enabled=True),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
async with client_for(app) as client:
|
|
ready = await client.get("/readyz")
|
|
motor_status = await client.get("/v1/motor/status")
|
|
|
|
assert ready.status_code == 200
|
|
assert ready.json() == {"ready": True}
|
|
assert motor_status.json()["state"] == "ready"
|
|
with transport.lock:
|
|
startup_frames = [
|
|
(frame.arbitration_id, frame.data.hex()) for frame in transport.frames
|
|
]
|
|
assert startup_frames == [
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8800880000000000"),
|
|
]
|
|
|
|
with transport.lock:
|
|
shutdown_frames = [
|
|
(frame.arbitration_id, frame.data.hex()) for frame in transport.frames[-5:]
|
|
]
|
|
assert shutdown_frames == [
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8000800000000000"),
|
|
]
|
|
assert transport.is_open is False
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_http_open_and_close_can_switch_directions() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
app = create_app(
|
|
Settings(hardware_enabled=True),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
async with client_for(app) as client:
|
|
opening = await client.post("/v1/motor/open")
|
|
await asyncio.sleep(0.06)
|
|
still_opening = await client.get("/v1/motor/status")
|
|
closing = await client.post("/v1/motor/close")
|
|
reopened = await client.post("/v1/motor/open")
|
|
stopped = await client.post("/v1/motor/stop")
|
|
|
|
assert opening.status_code == 200
|
|
assert opening.json()["code"] == 200
|
|
assert opening.json()["action"] == "open"
|
|
assert opening.json()["state"] == "opening"
|
|
assert opening.json()["control_raw"] == 400
|
|
assert opening.json()["control_raw_by_motor"] == {"1": 400, "2": -400}
|
|
assert opening.json()["control_mode"] == "open_loop"
|
|
assert opening.json()["streaming"] is True
|
|
assert opening.json()["broadcast_frequency_hz"] == 10
|
|
assert still_opening.json()["state"] == "opening"
|
|
assert still_opening.json()["streaming"] is True
|
|
assert closing.status_code == 200
|
|
assert closing.json()["code"] == 200
|
|
assert closing.json()["action"] == "close"
|
|
assert closing.json()["state"] == "closing"
|
|
assert closing.json()["control_raw"] == -400
|
|
assert closing.json()["control_raw_by_motor"] == {"1": -400, "2": 400}
|
|
assert "not verified" in closing.json()["transmission"]
|
|
assert reopened.status_code == 200
|
|
assert reopened.json()["code"] == 200
|
|
assert reopened.json()["action"] == "open"
|
|
assert reopened.json()["state"] == "opening"
|
|
assert reopened.json()["control_raw"] == 400
|
|
assert reopened.json()["control_raw_by_motor"] == {"1": 400, "2": -400}
|
|
assert stopped.status_code == 200
|
|
assert stopped.json()["code"] == 200
|
|
assert stopped.json()["state"] == "stopped"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_automatic_enable_and_motion_do_not_require_feedback_replies() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
app = create_app(
|
|
Settings(hardware_enabled=True),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
async with client_for(app) as client:
|
|
opening = await client.post("/v1/motor/open")
|
|
stopped = await client.post("/v1/motor/stop")
|
|
|
|
assert opening.status_code == 200
|
|
assert opening.json()["code"] == 200
|
|
assert opening.json()["streaming"] is True
|
|
assert stopped.status_code == 200
|
|
assert stopped.json()["code"] == 200
|
|
assert stopped.json()["streaming"] is False
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_one_http_stop_sends_two_explicit_stop_rounds() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
app = create_app(
|
|
Settings(hardware_enabled=True),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
async with client_for(app) as client:
|
|
start = len(transport.frames)
|
|
|
|
stopped = await client.post("/v1/motor/stop")
|
|
|
|
with transport.lock:
|
|
frames = [
|
|
(frame.arbitration_id, frame.data.hex()) for frame in transport.frames[start:]
|
|
]
|
|
timeouts_s = transport.tx_queue_timeouts_s[start:]
|
|
assert stopped.status_code == 200
|
|
assert stopped.json()["code"] == 200
|
|
assert stopped.json()["state"] == "stopped"
|
|
assert frames == [
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
]
|
|
assert timeouts_s == [None, 0.1, None, 0.1]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_auto_enable_failure_aborts_startup_and_closes_transport() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
transport.fail_next = True
|
|
app = create_app(
|
|
Settings(hardware_enabled=True),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
with pytest.raises(MotorCommunicationError, match="injected CAN failure"):
|
|
async with client_for(app):
|
|
raise AssertionError("startup failure must prevent HTTP service readiness")
|
|
|
|
with transport.lock:
|
|
cleanup_frames = [
|
|
(frame.arbitration_id, frame.data.hex()) for frame in transport.frames
|
|
]
|
|
cleanup_timeouts = list(transport.tx_queue_timeouts_s)
|
|
assert cleanup_frames == [
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
(CONTROL_FRAME_ID, "0000000000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8100810000000000"),
|
|
(MIXED_COMMAND_FRAME_ID, "8000800000000000"),
|
|
]
|
|
assert cleanup_timeouts == [None, 0.1, None, 0.1, 0.1]
|
|
assert transport.is_open is False
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_api_key_is_required_when_configured() -> None:
|
|
async def scenario() -> None:
|
|
transport = FakeTransport()
|
|
app = create_app(
|
|
Settings(hardware_enabled=True, api_token="test-secret"),
|
|
transport_factory=lambda settings: transport,
|
|
)
|
|
|
|
async with client_for(app) as client:
|
|
assert (await client.get("/v1/motor/status")).status_code == 401
|
|
response = await client.get("/v1/motor/status", headers={"X-API-Key": "test-secret"})
|
|
assert response.status_code == 200
|
|
|
|
asyncio.run(scenario())
|