Method command

tests/utils/api_client.py:121–167

Send a command and return the result. Args: command: Command name (e.g., "io.getStatus") params: Optional parameters dict timeout: Optional timeout override Returns: Result dict on success Raises: APIErro

(
        self,
        command: str,
        params: Optional[dict] = None,
        timeout: Optional[float] = None,
    )

        self._socket.sendall(data.encode("utf-8"))

    def command(
        self,
        command: str,
        params: Optional[dict] = None,
        timeout: Optional[float] = None,
    ) -> dict:
        """
        Send a command and return the result.

        Args:
            command: Command name (e.g., "io.getStatus")
            params: Optional parameters dict
            timeout: Optional timeout override

        Returns:
            Result dict on success

        Raises:
            APIError: If command fails
            TimeoutError: If response not received in time
        """
        request_id = str(uuid.uuid4())
        message = {"type": "command", "id": request_id, "command": command}

        if params:
            message["params"] = params

        self._send_message(message)

        deadline = time.time() + (timeout or self.timeout)

        while True:
            remaining = deadline - time.time()
            if remaining <= 0:
                raise TimeoutError("Timeout waiting for response")

            response = self._recv_message(remaining)

            if response.get("type") == "response" and response.get("id") == request_id:
                if response.get("success"):
                    return response.get("result", {})
                else:
                    error = response.get("error", {})
                    raise APIError(
                        error.get("code", "UNKNOWN"),
                        error.get("message", "Unknown error"),
                    )

    def batch(
        self, commands: list[dict], timeout: Optional[float] = None