dzxj_dtu/backend/tests/test_nextion_display.py

import os
import sys
import time
from unittest.mock import MagicMock, Mock, PropertyMock, patch

import pytest

# 添加项目根目录到 Python 路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from modules.nextion_display import EVENT_HEADER, TERMINATOR, NextionDisplay, parse_nextion_events


@pytest.fixture
def display():
    return NextionDisplay()


def test_parse_button_event():
    data = bytes([EVENT_HEADER, 0x00, 0x05, 0x01]) + TERMINATOR
    events, remaining = parse_nextion_events(data)
    assert events == [{'type': 'button', 'page': 0, 'comp': 5, 'event': 1}]
    assert remaining == b''


def test_parse_multiple_events():
    data = (
        bytes([EVENT_HEADER, 0x00, 0x05, 0x01]) + TERMINATOR +
        bytes([EVENT_HEADER, 0x00, 0x06, 0x01]) + TERMINATOR
    )
    events, remaining = parse_nextion_events(data)
    assert len(events) == 2
    assert events[1] == {'type': 'button', 'page': 0, 'comp': 6, 'event': 1}
    assert remaining == b''


def test_parse_event_with_trailing_partial_bytes():
    data = bytes([EVENT_HEADER, 0x00, 0x05, 0x01]) + TERMINATOR + bytes([EVENT_HEADER])
    events, remaining = parse_nextion_events(data)
    assert events == [{'type': 'button', 'page': 0, 'comp': 5, 'event': 1}]
    assert remaining == bytes([EVENT_HEADER])


def test_encode_nextion_cmd():
    display = NextionDisplay()
    encoded = display._encode_cmd('t2.txt="终端1"')
    assert encoded.endswith(TERMINATOR)
    assert encoded.startswith(b't2.txt=')


def test_connect_success(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial) as mock_cls:
        ok, msg = display.connect('/dev/ttyUSB0', 115200)

    assert ok is True
    assert '115200' in msg
    mock_cls.assert_called_once_with('/dev/ttyUSB0', 115200, timeout=0.1)
    assert display.connected is True
    assert display.port == '/dev/ttyUSB0'
    display.disconnect()


def test_connect_failure(display):
    with patch('modules.nextion_display.serial.Serial', side_effect=Exception("open fail")):
        ok, msg = display.connect('/dev/ttyUSB0', 9600)

    assert ok is False
    assert 'open fail' in msg
    assert display.connected is False
    assert display.ser is None


def test_disconnect_when_connected(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)

    ok, msg = display.disconnect()
    assert ok is True
    assert msg == "已断开"
    assert display.connected is False
    assert display.ser is None
    assert display.port is None
    assert display.baudrate is None


def test_disconnect_when_not_connected(display):
    ok, msg = display.disconnect()
    assert ok is True
    assert msg == "已断开"
    assert display.connected is False


def test_send_cmd_success(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)

    ok = display.send_cmd('t2.txt="test"')
    assert ok is True
    mock_serial.write.assert_called()
    display.disconnect()


def test_send_cmd_failure_not_connected(display):
    ok = display.send_cmd('t2.txt="test"')
    assert ok is False


def test_send_cmd_failure_exception(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0
    mock_serial.write.side_effect = Exception("write fail")

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)

    ok = display.send_cmd('t2.txt="test"')
    assert ok is False
    assert display.connected is False
    display.disconnect()


def test_send_raw_success(display):
    raw = b'\x00\x01\x02' + TERMINATOR
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)

    ok = display.send_raw(raw)
    assert ok is True
    mock_serial.write.assert_called_with(raw)
    display.disconnect()


def test_get_status(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 115200)

    status = display.get_status()
    assert status['connected'] is True
    assert status['port'] == '/dev/ttyUSB0'
    assert status['baudrate'] == 115200
    display.disconnect()


def test_read_loop_dispatches_button_callback(display):
    callback = Mock()
    display.set_button_callback(callback)

    event_bytes = bytes([EVENT_HEADER, 0x00, 0x05, 0x01]) + TERMINATOR

    mock_serial = MagicMock()
    mock_serial.is_open = True
    type(mock_serial).in_waiting = PropertyMock(
        side_effect=[len(event_bytes)] + [0] * 100
    )
    mock_serial.read.return_value = event_bytes

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)

    time.sleep(0.05)
    display.disconnect()
    callback.assert_called_once_with(0, 5, 1)


def test_reentrant_connect_no_deadlock(display):
    mock_serial = MagicMock()
    mock_serial.is_open = True
    mock_serial.in_waiting = 0

    with patch('modules.nextion_display.serial.Serial', return_value=mock_serial):
        display.connect('/dev/ttyUSB0', 9600)
        display.connect('/dev/ttyUSB1', 115200)

    assert display.port == '/dev/ttyUSB1'
    assert display.baudrate == 115200
    assert display.connected is True
    display.disconnect()