File: test_tcp.py

package info (click to toggle)
python-umodbus 1.0.4-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 456 kB
  • sloc: python: 1,944; makefile: 166; sh: 5
file content (67 lines) | stat: -rw-r--r-- 1,973 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
""" The module umodbus.server is mainly covered through system tests. Only
those parts which can't be tested by system tests should be tested using
unit tests.
"""
import struct
import pytest

from umodbus.exceptions import ServerDeviceFailureError
from umodbus.client.tcp import read_coils
from umodbus.server.tcp import RequestHandler


@pytest.fixture
def meta_data():
    return {
        'transaction_id': 1337,
        'protocol_id': 0,
        'length': 1,
        'unit_id': 5,
    }


@pytest.fixture
def mbap_header():
    transaction_id = 1337
    protocol_id = 0
    length = 1
    unit_id = 5

    return struct.pack('>HHHB', transaction_id, protocol_id, length, unit_id)


@pytest.fixture
def request_handler(monkeypatch):
    # handle() is called when after creating a RequestHandler. This
    # causes an error because it wants to read from a socket. Mock it out so
    # this error doesn't occur.
    monkeypatch.setattr(RequestHandler, 'handle', lambda _: None)
    return RequestHandler(None, None, None)


def test_handle_raising_exception():
    """ Test tests RequestHandler.handle() which is called when an instance
    of RequestHandler is created. This method should reraise exception if one
    occurs.
    """
    with pytest.raises(AttributeError):
        RequestHandler(None, None, None)


def test_request_handler_get_meta_data(request_handler, mbap_header,
                                       meta_data):
    assert request_handler.get_meta_data(mbap_header) == meta_data


def test_request_handler_get_meta_data_raising_error(request_handler):
    with pytest.raises(ServerDeviceFailureError):
        request_handler.get_meta_data(b'')


def def_test_get_request_pdu(request_handler, mbap_header):
    pdu = read_coils(1, 1, 1)
    assert request_handler.get_request_pdu(mbap_header + pdu) == pdu


def test_response_adu(request_handler, mbap_header, meta_data):
    assert len(request_handler.create_response_adu(meta_data, b'')) == 7