File: test_events.py

package info (click to toggle)
pymodbus 2.1.0%2Bdfsg-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 2,708 kB
  • sloc: python: 17,594; makefile: 84; sh: 8
file content (77 lines) | stat: -rw-r--r-- 2,558 bytes parent folder | download | duplicates (2)
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
#!/usr/bin/env python
import unittest
from pymodbus.events import *
from pymodbus.exceptions import NotImplementedException
from pymodbus.exceptions import ParameterException

class ModbusEventsTest(unittest.TestCase):
    '''
    This is the unittest for the pymodbus.device module
    '''

    def setUp(self):
        ''' Sets up the test environment '''
        pass

    def tearDown(self):
        ''' Cleans up the test environment '''
        pass

    def testModbusEventBaseClass(self):
        event = ModbusEvent()
        self.assertRaises(NotImplementedException, event.encode)
        self.assertRaises(NotImplementedException, lambda: event.decode(None))

    def testRemoteReceiveEvent(self):
        event = RemoteReceiveEvent()
        event.decode(b'\x70')
        self.assertTrue(event.overrun)
        self.assertTrue(event.listen)
        self.assertTrue(event.broadcast)

    def testRemoteSentEvent(self):
        event = RemoteSendEvent()
        result = event.encode()
        self.assertEqual(result, b'\x40')
        event.decode(b'\x7f')
        self.assertTrue(event.read)
        self.assertTrue(event.slave_abort)
        self.assertTrue(event.slave_busy)
        self.assertTrue(event.slave_nak)
        self.assertTrue(event.write_timeout)
        self.assertTrue(event.listen)

    def testRemoteSentEventEncode(self):
        arguments = {
            'read'          : True,
            'slave_abort'   : True,
            'slave_busy'    : True,
            'slave_nak'     : True,
            'write_timeout' : True,
            'listen'        : True,
        }
        event = RemoteSendEvent(**arguments)
        result = event.encode()
        self.assertEqual(result, b'\x7f')

    def testEnteredListenModeEvent(self):
        event = EnteredListenModeEvent()
        result = event.encode()
        self.assertEqual(result, b'\x04')
        event.decode(b'\x04')
        self.assertEqual(event.value, 0x04)
        self.assertRaises(ParameterException, lambda: event.decode(b'\x00'))

    def testCommunicationRestartEvent(self):
        event = CommunicationRestartEvent()
        result = event.encode()
        self.assertEqual(result, b'\x00')
        event.decode(b'\x00')
        self.assertEqual(event.value, 0x00)
        self.assertRaises(ParameterException, lambda: event.decode(b'\x04'))

#---------------------------------------------------------------------------#
# Main
#---------------------------------------------------------------------------#
if __name__ == "__main__":
    unittest.main()