← Back to branch summary

~ubuntu-branches/ubuntu/raring/pymodbus/raring-proposed

« back to all changes in this revision

Viewing changes to examples/functional/base_runner.py

  • Committer: Package Import Robot
  • Author(s): W. Martin Borgert
  • Date: 2011-10-26 07:26:28 UTC
  • mfrom: (1.1.1)
  • Revision ID: package-import@ubuntu.com-20111026072628-fvzyi6tnb8iipomp
Tags: 0.9.0+r175-1
* Update from trunk to get a number of upstream fixes.
* Removed examples/tools/ (not present in previous version
  anyway) from source because there are different licenses
  involved. Needs clarification.
* Dont't install unit tests.
* Debian patches not necessary anymore.

Show diffs side-by-side

added added

removed removed

Lines of Context:
examples/functional/base_runner.py
 
1
import os
 
2
import time
 
3
from subprocess import Popen as execute
 
4
from twisted.internet.defer import Deferred
 
5
 
 
6
#---------------------------------------------------------------------------# 
 
7
# configure the client logging
 
8
#---------------------------------------------------------------------------# 
 
9
import logging
 
10
log = logging.getLogger(__name__)
 
11
 
 
12
class Runner(object):
 
13
    '''
 
14
    This is the base runner class for all the integration tests
 
15
    '''
 
16
 
 
17
    def initialize(self, service):
 
18
        ''' Initializes the test environment '''
 
19
        self.fnull  = open(os.devnull, 'w')
 
20
        self.server = execute(service, stdout=self.fnull, stderr=self.fnull)
 
21
        log.debug("%s service started: %s", service, self.server.pid)
 
22
        time.sleep(0.2)
 
23
 
 
24
    def shutdown(self):
 
25
        ''' Cleans up the test environment '''
 
26
        self.server.kill()
 
27
        self.fnull.close()
 
28
        log.debug("service stopped")
 
29
 
 
30
    def testReadWriteCoil(self):
 
31
        rq = self.client.write_coil(1, True)
 
32
        rr = self.client.read_coils(1,1)
 
33
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
34
        self.__validate(rr, lambda r: r.bits[0] == True)
 
35
        
 
36
    def testReadWriteCoils(self):
 
37
        rq = self.client.write_coils(1, [True]*8)
 
38
        rr = self.client.read_coils(1,8)
 
39
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
40
        self.__validate(rr, lambda r: r.bits == [True]*8)
 
41
        
 
42
    def testReadWriteDiscreteRegisters(self):
 
43
        rq = self.client.write_coils(1, [False]*8)
 
44
        rr = self.client.read_discrete_inputs(1,8)
 
45
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
46
        self.__validate(rr, lambda r: r.bits == [False]*8)
 
47
        
 
48
    def testReadWriteHoldingRegisters(self):
 
49
        rq = self.client.write_register(1, 10)
 
50
        rr = self.client.read_holding_registers(1,1)
 
51
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
52
        self.__validate(rr, lambda r: r.registers[0] == 10)
 
53
        
 
54
    def testReadWriteInputRegisters(self):
 
55
        rq = self.client.write_registers(1, [10]*8)
 
56
        rr = self.client.read_input_registers(1,8)
 
57
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
58
        self.__validate(rr, lambda r: r.registers == [10]*8)
 
59
        
 
60
    def testReadWriteRegistersTogether(self):
 
61
        arguments = {
 
62
            'read_address':    1,
 
63
            'read_count':      8,
 
64
            'write_address':   1,
 
65
            'write_registers': [20]*8,
 
66
        }
 
67
        rq = self.client.readwrite_registers(**arguments)
 
68
        rr = self.client.read_input_registers(1,8)
 
69
        self.__validate(rq, lambda r: r.function_code < 0x80)
 
70
        self.__validate(rr, lambda r: r.registers == [20]*8)
 
71
 
 
72
    def __validate(self, result, test):
 
73
        ''' Validate the result whether it is a result or a deferred.
 
74
 
 
75
        :param result: The result to __validate
 
76
        :param callback: The test to __validate
 
77
        '''
 
78
        if isinstance(result, Deferred):
 
79
            deferred.callback(lambda : self.assertTrue(test(result)))
 
80
            deferred.errback(lambda _: self.assertTrue(False))
 
81
        else: self.assertTrue(test(result))
 
82