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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
#!/usr/bin/env python
import unittest
from mock import patch, Mock
from pymodbus.device import ModbusDeviceIdentification
from pymodbus.server.async import ModbusTcpProtocol, ModbusUdpProtocol
from pymodbus.server.async import ModbusServerFactory
from pymodbus.server.async import StartTcpServer, StartUdpServer, StartSerialServer
from pymodbus.exceptions import ConnectionException, NotImplementedException
from pymodbus.exceptions import ParameterException
from pymodbus.bit_read_message import ReadCoilsRequest, ReadCoilsResponse
#---------------------------------------------------------------------------#
# Fixture
#---------------------------------------------------------------------------#
class AsynchronousServerTest(unittest.TestCase):
'''
This is the unittest for the pymodbus.server.async module
'''
#-----------------------------------------------------------------------#
# Setup/TearDown
#-----------------------------------------------------------------------#
def setUp(self):
'''
Initializes the test environment
'''
values = dict((i, '') for i in range(10))
identity = ModbusDeviceIdentification(info=values)
def tearDown(self):
''' Cleans up the test environment '''
pass
#-----------------------------------------------------------------------#
# Test Modbus Server Factory
#-----------------------------------------------------------------------#
def testModbusServerFactory(self):
''' Test the base class for all the clients '''
factory = ModbusServerFactory(store=None)
self.assertEqual(factory.control.Identity.VendorName, '')
identity = ModbusDeviceIdentification(info={0x00: 'VendorName'})
factory = ModbusServerFactory(store=None, identity=identity)
self.assertEqual(factory.control.Identity.VendorName, 'VendorName')
#-----------------------------------------------------------------------#
# Test Modbus TCP Server
#-----------------------------------------------------------------------#
def testTCPServerDisconnect(self):
protocol = ModbusTcpProtocol()
protocol.connectionLost('because of an error')
#-----------------------------------------------------------------------#
# Test Modbus UDP Server
#-----------------------------------------------------------------------#
def testUdpServerInitialize(self):
protocol = ModbusUdpProtocol(store=None)
self.assertEqual(protocol.control.Identity.VendorName, '')
identity = ModbusDeviceIdentification(info={0x00: 'VendorName'})
protocol = ModbusUdpProtocol(store=None, identity=identity)
self.assertEqual(protocol.control.Identity.VendorName, 'VendorName')
#-----------------------------------------------------------------------#
# Test Modbus Server Startups
#-----------------------------------------------------------------------#
def testTcpServerStartup(self):
''' Test that the modbus tcp async server starts correctly '''
with patch('twisted.internet.reactor') as mock_reactor:
StartTcpServer(context=None, console=True)
self.assertEqual(mock_reactor.listenTCP.call_count, 2)
self.assertEqual(mock_reactor.run.call_count, 1)
def testUdpServerStartup(self):
''' Test that the modbus udp async server starts correctly '''
with patch('twisted.internet.reactor') as mock_reactor:
StartUdpServer(context=None)
self.assertEqual(mock_reactor.listenUDP.call_count, 1)
self.assertEqual(mock_reactor.run.call_count, 1)
def testSerialServerStartup(self):
''' Test that the modbus serial async server starts correctly '''
with patch('twisted.internet.reactor') as mock_reactor:
StartSerialServer(context=None, port='/dev/ptmx')
self.assertEqual(mock_reactor.run.call_count, 1)
#---------------------------------------------------------------------------#
# Main
#---------------------------------------------------------------------------#
if __name__ == "__main__":
unittest.main()
|
