File: test_bit_write_messages.py

package info (click to toggle)
pymodbus 1.2.0%2Bgit20151013-1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 2,084 kB
  • ctags: 2,155
  • sloc: python: 14,717; makefile: 86
file content (122 lines) | stat: -rw-r--r-- 4,459 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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
#!/usr/bin/env python
'''
Bit Message Test Fixture
--------------------------------
This fixture tests the functionality of all the 
bit based request/response messages:

* Read/Write Discretes
* Read Coils
'''
import unittest
from pymodbus.bit_write_message import *
from pymodbus.exceptions import *
from pymodbus.pdu import ModbusExceptions

from modbus_mocks import MockContext, FakeList

#---------------------------------------------------------------------------#
# Fixture
#---------------------------------------------------------------------------#
class ModbusBitMessageTests(unittest.TestCase):

    #-----------------------------------------------------------------------#
    # Setup/TearDown
    #-----------------------------------------------------------------------#

    def setUp(self):
        '''
        Initializes the test environment and builds request/result
        encoding pairs
        '''
        pass

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

    def testBitWriteBaseRequests(self):
        messages = {
            WriteSingleCoilRequest(1, 0xabcd)      : '\x00\x01\xff\x00',
            WriteSingleCoilResponse(1, 0xabcd)     : '\x00\x01\xff\x00',
            WriteMultipleCoilsRequest(1, [True]*5) : '\x00\x01\x00\x05\x01\x1f',
            WriteMultipleCoilsResponse(1, 5)       : '\x00\x01\x00\x05',
        }
        for request, expected in messages.iteritems():
            self.assertEqual(request.encode(), expected)

    def testWriteMultipleCoilsRequest(self):
        request = WriteMultipleCoilsRequest(1, [True]*5)
        request.decode('\x00\x01\x00\x05\x01\x1f')
        self.assertEqual(request.byte_count, 1)
        self.assertEqual(request.address, 1)
        self.assertEqual(request.values, [True]*5)

    def testInvalidWriteMultipleCoilsRequest(self):
        request = WriteMultipleCoilsRequest(1, None)
        self.assertEquals(request.values, [])

    def testWriteSingleCoilRequestEncode(self):
        request = WriteSingleCoilRequest(1, False)
        self.assertEquals(request.encode(), '\x00\x01\x00\x00')

    def testWriteSingleCoilExecute(self):
        context = MockContext(False, default=True)
        request = WriteSingleCoilRequest(2, True)
        result  = request.execute(context)
        self.assertEqual(result.exception_code, ModbusExceptions.IllegalAddress)

        context.valid = True
        result = request.execute(context)
        self.assertEqual(result.encode(), '\x00\x02\xff\x00')

        context = MockContext(True, default=False)
        request = WriteSingleCoilRequest(2, False)
        result = request.execute(context)
        self.assertEqual(result.encode(), '\x00\x02\x00\x00')

    def testWriteMultipleCoilsExecute(self):
        context = MockContext(False)
        # too many values
        request = WriteMultipleCoilsRequest(2, FakeList(0x123456))
        result  = request.execute(context)
        self.assertEqual(result.exception_code, ModbusExceptions.IllegalValue)

        # bad byte count
        request = WriteMultipleCoilsRequest(2, [0x00]*4)
        request.byte_count = 0x00
        result  = request.execute(context)
        self.assertEqual(result.exception_code, ModbusExceptions.IllegalValue)

        # does not validate
        context.valid = False
        request = WriteMultipleCoilsRequest(2, [0x00]*4)
        result  = request.execute(context)
        self.assertEqual(result.exception_code, ModbusExceptions.IllegalAddress)

        # validated request
        context.valid = True
        result  = request.execute(context)
        self.assertEqual(result.encode(), '\x00\x02\x00\x04')

    def testWriteMultipleCoilsResponse(self):
        response = WriteMultipleCoilsResponse()
        response.decode('\x00\x80\x00\x08')
        self.assertEqual(response.address, 0x80)
        self.assertEqual(response.count, 0x08)

    def testSerializingToString(self):
        requests = [
            WriteSingleCoilRequest(1, 0xabcd),
            WriteSingleCoilResponse(1, 0xabcd),
            WriteMultipleCoilsRequest(1, [True]*5),
            WriteMultipleCoilsResponse(1, 5),
        ]
        for request in requests:
            self.assertTrue(str(request) != None)

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