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
68
69
70
71
72
73
74
75
76
77
78
79
|
"""Test example server/client async.
This is a thorough test of the
- client_async.py
- server_async.py
examples.
These are basis for most examples and thus tested separately
"""
import asyncio
import pytest
from examples.client_async import (
main,
run_a_few_calls,
run_async_client,
setup_async_client,
)
@pytest.mark.parametrize(
("use_comm", "use_framer"),
[
("tcp", "socket"),
("tcp", "rtu"),
("tls", "tls"),
("udp", "socket"),
("udp", "rtu"),
("serial", "rtu"),
],
)
class TestClientServerAsyncExamples:
"""Test Client server async examples."""
@staticmethod
@pytest.fixture(name="use_port")
def get_port_in_class(base_ports):
"""Return next port."""
base_ports[__class__.__name__] += 1
return base_ports[__class__.__name__]
async def test_combinations(self, mock_server, mock_clc):
"""Run async client and server."""
assert mock_server
await main(cmdline=mock_clc)
async def test_client_no_calls(self, mock_server, mock_clc):
"""Run async client and server."""
assert mock_server
test_client = setup_async_client(cmdline=mock_clc)
await run_async_client(test_client, modbus_calls=None)
async def test_server_no_client(self, mock_server):
"""Run async server without client."""
assert mock_server
async def test_server_client_twice(self, mock_server, use_comm, mock_clc):
"""Run async server without client."""
assert mock_server
if use_comm == "serial":
# Serial do not allow mmulti point.
return
test_client = setup_async_client(cmdline=mock_clc)
await run_async_client(test_client, modbus_calls=run_a_few_calls)
await asyncio.sleep(0.5)
await run_async_client(test_client, modbus_calls=run_a_few_calls)
async def test_client_no_server(self, mock_clc):
"""Run async client without server."""
test_client = setup_async_client(cmdline=mock_clc)
with pytest.raises((AssertionError, asyncio.TimeoutError)):
await run_async_client(test_client, modbus_calls=run_a_few_calls)
async def test_illegal_commtype():
"""Run async client and server."""
with pytest.raises(RuntimeError):
setup_async_client(cmdline=["--comm", "unknown", "--framer", "rtu", "--port", "5912"]
)
|
