kraken/backend/tests/core/peripheral/hand/test_ruishou.py

"""
@Author  :   liujunshen
@File    :   test_ruishou.py
@Time    :   2023/04/04 13:57:03
"""

from collections import namedtuple
import time

import pytest

from core.peripheral.hand.ruishou import Constants
from core.peripheral.hand.ruishou import Protocol
from core.peripheral.hand.ruishou import RuishouClient
from core.peripheral.hand.ruishou import RuishouConnector

buffer_time = 0.3
ParamStruct = namedtuple(
    "ParamStruct",
    "hand_select thumb index_finger middle_finger ring_finger little_finger duration"
)


# ============= 测试解析模块 ================
def test_protocol_get_pack_success():
    protocol = Protocol()
    ret = protocol.get_pck("finish_action")
    assert isinstance(ret, bytearray)


def test_protocol_get_pack_with_fail_cmd_return_none():
    protocol = Protocol()
    ret = protocol.get_pck("error_cmd")
    assert ret is None


def test_protocol_get_motion_control_pack_success():
    params = {
        Constants.SendPckLocation.MOTION_CONTROL_HAND: 2,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_INDEX_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_MIDDLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 10
    }
    protocol = Protocol()
    ret = protocol.get_pck("motion_control", params)
    assert isinstance(ret, bytearray)


def test_protocol_get_motion_control_pack_with_lack_update_dict_return_none():
    params = {
        Constants.SendPckLocation.MOTION_CONTROL_HAND: 2,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 10,
    }
    protocol = Protocol()
    ret = protocol.get_pck("motion_control", params)
    assert ret is None


def test_protocol_get_motion_control_pack_with_error_update_dict_return_none():
    params = {
        11: buffer_time,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 10,
    }
    protocol = Protocol()
    ret = protocol.get_pck("motion_control", params)
    assert ret is None


def test_protocol_unpack_one_cmd_bytes_success():
    protocol = Protocol()
    b = b"\xae\xaf\x05\x01\x00\x00\xff\xff\x01\xff"
    ret = protocol.unpack_bytes(b)
    assert isinstance(ret, list)
    assert len(ret) == 1


def test_protocol_unpack_two_cmd_bytes_success():
    protocol = Protocol()
    b = b"\xae\xaf\x05\x01\x00\x00\xff\xff\x01\xff\xae\xaf\x05\x02\xff\xff\xff\xff\x03\xfe"
    ret = protocol.unpack_bytes(b)
    assert isinstance(ret, list)
    assert len(ret) == 2


def test_protocol_unpack_bytes_with_error_bytes_return_empty_list():
    protocol = Protocol()
    b = b"\x05\x01\x00\x00\xff\xff\x01\xff"
    ret = protocol.unpack_bytes(b)
    assert isinstance(ret, list)
    assert len(ret) == 0


def test_protocol_parse_list_success():
    protocol = Protocol()
    b = b"\xae\xaf\x05\x01\x00\x00\xff\xff"
    unpack_data = protocol.unpack_bytes(b)
    parsed_data = protocol.parse_bytes(unpack_data[0])
    assert isinstance(parsed_data, dict)


# =======以下需要启动设备或模拟测试软件=============


@pytest.mark.ruishou
def test_connector_connect_and_close_success():
    connector = RuishouConnector()
    connector.start_client()
    time.sleep(buffer_time)
    connector.close_client()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_connector_sync_send_control_motion_data_success():
    connector = RuishouConnector()
    connector.start_client()
    params = {
        Constants.SendPckLocation.MOTION_CONTROL_HAND: 2,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_INDEX_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_MIDDLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 5
    }
    res = connector.sync_send_data("motion_control", params)
    assert isinstance(res, dict)
    time.sleep(5)
    connector.close_client()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_connector_sync_send_error_data_return_none():
    connector = RuishouConnector()
    connector.start_client()
    params = {}
    res = connector.sync_send_data("error_data", params)
    assert res is None
    time.sleep(buffer_time)
    connector.close_client()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_connector_stop_operate_success():
    connector = RuishouConnector()
    connector.start_client()
    params = {
        Constants.SendPckLocation.MOTION_CONTROL_HAND: 1,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_INDEX_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_MIDDLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 5
    }
    connector.sync_send_data("motion_control", params)
    time.sleep(3)
    connector.sync_send_data("finish_action")
    connector.close_client()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_connector_sync_send_control_motion_error_params_fail():
    connector = RuishouConnector()
    connector.start_client()
    params = {
        Constants.SendPckLocation.MOTION_CONTROL_HAND: 2,
        Constants.SendPckLocation.MOTION_CONTROL_THUMB_BENDING: 15,
        Constants.SendPckLocation.MOTION_CONTROL_RING_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_LITTLE_FINGER_BENDING: 10,
        Constants.SendPckLocation.MOTION_CONTROL_DURATION: 5
    }
    res = connector.sync_send_data("motion_control", params)
    assert res is None
    time.sleep(buffer_time)
    connector.close_client()
    time.sleep(buffer_time)


# ========== 测试睿手客户端(对业务) ============


@pytest.mark.ruishou
def test_client_init_success():
    client = RuishouClient()
    ret = client.init()
    assert ret["is_connected"]
    time.sleep(buffer_time)
    client.close()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_get_status_success():
    client = RuishouClient()
    client.init()
    status = client.status()
    assert status["is_connected"]
    time.sleep(buffer_time)
    client.close()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_get_status_with_not_init_return_not_connected():
    client = RuishouClient()
    status = client.status()
    assert not status["is_connected"]


@pytest.mark.ruishou
def test_client_get_status_with_close_client_return_not_connected():
    client = RuishouClient()
    client.init()
    time.sleep(buffer_time)
    client.close()
    status = client.status()
    assert not status["is_connected"]
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_start_operate_success():
    client = RuishouClient()
    client.init()
    params = ParamStruct("左手", 10, 10, 10, 10, 10, 6)
    res = client._control_motion(params)
    assert isinstance(res, dict)
    time.sleep(5)
    client.close()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_reconnect_start_operate_success():
    client = RuishouClient()
    client.init()
    params = ParamStruct("左手", 10, 10, 10, 10, 10, 6)
    time.sleep(buffer_time)
    client.close()
    res = client._control_motion(params)
    assert isinstance(res, dict)
    time.sleep(5)
    client.close()
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_close_success():
    client = RuishouClient()
    client.init()
    time.sleep(buffer_time)
    client.close()
    assert not client.connector.is_connected
    time.sleep(buffer_time)


@pytest.mark.ruishou
def test_client_close_with_not_init_success():
    client = RuishouClient()
    client.close()
    assert not client.connector.is_connected