Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions tesla_fleet_api/teslemetry/vehicle.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,14 @@ async def closure(
json=data,
)

async def open_rear_trunk(self) -> dict[str, Any]:
"""Opens the rear trunk; unlike actuate_trunk, never closes it."""
return await self.closure(rear_trunk=ClosureState.OPEN)

async def close_rear_trunk(self) -> dict[str, Any]:
"""Closes the rear trunk; unlike actuate_trunk, never opens it."""
return await self.closure(rear_trunk=ClosureState.CLOSE)

async def open_tonneau(self) -> dict[str, Any]:
"""Opens the tonneau cover."""
return await self.closure(tonneau=ClosureState.OPEN)
Expand Down
54 changes: 54 additions & 0 deletions tests/test_teslemetry_vehicle_custom_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -425,3 +425,57 @@ async def open_tonneau(self) -> dict[str, bool]:
await VehicleRouter(ble, cloud).open_tonneau()
self.assertEqual(ble.calls, 1)
request.assert_awaited_once()


class RearTrunkTests(IsolatedAsyncioTestCase):
"""Rear trunk methods mirror the ``VehicleBluetooth`` names via ``closure``."""

async def _assert_rear_trunk(self, method: str, body: str) -> None:
vehicle, request = _make_vehicle()
result = await getattr(vehicle, method)()
request.assert_awaited_once_with(
Method.POST,
f"api/1/vehicles/{VIN}/custom_command/closure",
json={"rearTrunk": body},
)
self.assertEqual(result, {"response": {"result": True}})

async def test_open_rear_trunk(self) -> None:
await self._assert_rear_trunk("open_rear_trunk", "open")

async def test_close_rear_trunk(self) -> None:
await self._assert_rear_trunk("close_rear_trunk", "close")

async def test_router_prefers_ble_and_falls_back_to_cloud(self) -> None:
from tesla_fleet_api.router import VehicleRouter

class _Ble:
vin = VIN

def __init__(self, fail: bool) -> None:
self.fail = fail
self.calls = 0

async def close_rear_trunk(self) -> dict[str, bool]:
self.calls += 1
if self.fail:
raise ConnectionError("ble down")
return {"ble": True}

cloud, request = _make_vehicle()

ble = _Ble(fail=False)
self.assertEqual(
await VehicleRouter(ble, cloud).close_rear_trunk(), {"ble": True}
)
request.assert_not_awaited()

ble = _Ble(fail=True)
result = await VehicleRouter(ble, cloud).close_rear_trunk()
self.assertEqual(ble.calls, 1)
self.assertEqual(result, {"response": {"result": True}})
request.assert_awaited_once_with(
Method.POST,
f"api/1/vehicles/{VIN}/custom_command/closure",
json={"rearTrunk": "close"},
)
Loading