add some reject testing code to the sandbox

sandbox/i_am_reject_test.py

@@ -0,0 +1,151 @@
+#!/usr/bin/python
+
+"""
+BACpypes Test
+-------------
+"""
+
+import os
+import sys
+
+from bacpypes.debugging import bacpypes_debugging, ModuleLogger, xtob
+from bacpypes.consolelogging import ArgumentParser
+
+from bacpypes.pdu import Address
+from bacpypes.comm import bind
+
+from bacpypes.apdu import APDU, IAmRequest
+
+from bacpypes.app import LocalDeviceObject, Application
+from bacpypes.appservice import ApplicationServiceAccessPoint, StateMachineAccessPoint
+
+from tests.state_machine import ServerStateMachine
+
+# some debugging
+_debug = 0
+_log = ModuleLogger(globals())
+
+# defaults for testing
+BACPYPES_TEST = ""
+BACPYPES_TEST_OPTION = ""
+
+# parsed test options
+test_options = None
+
+
+#
+#   TestApplication
+#
+
+@bacpypes_debugging
+class TestApplication(Application):
+
+    def __init__(self, localDevice, localAddress, aseID=None):
+        if _debug: TestApplication._debug("__init__ %r %r aseID=%r", localDevice, localAddress, aseID)
+        Application.__init__(self, localDevice, localAddress, aseID)
+
+#
+#   setUp
+#
+
+@bacpypes_debugging
+def setUp(argv=None):
+    global test_options
+
+    # create an argument parser
+    parser = ArgumentParser(description=__doc__)
+
+    # add an option
+    parser.add_argument(
+        '--option', help="this is an option",
+        default=os.getenv("BACPYPES_TEST_OPTION") or BACPYPES_TEST_OPTION,
+        )
+
+    # get the debugging args and parse them
+    arg_str = os.getenv("BACPYPES_TEST") or BACPYPES_TEST
+    test_options = parser.parse_args(argv or arg_str.split())
+
+    if _debug: setUp._debug("setUp")
+    if _debug: setUp._debug("    - test_options: %r", test_options)
+
+#
+#   tearDown
+#
+
+@bacpypes_debugging
+def tearDown():
+    if _debug: tearDown._debug("tearDown")
+
+#
+#   main
+#
+
+setUp(sys.argv[1:])
+
+# make a device object
+test_device = LocalDeviceObject(
+    objectName='test_device',
+    objectIdentifier=599,
+    maxApduLengthAccepted=1024,
+    segmentationSupported='segmentedBoth',
+    vendorIdentifier=15,
+    )
+
+# make a test address
+test_address = Address(1)
+
+# create a client state machine, trapped server, and bind them together
+test_application = TestApplication(test_device, test_address)
+
+# include a application decoder
+test_asap = ApplicationServiceAccessPoint()
+
+# pass the device object to the state machine access point so it
+# can know if it should support segmentation
+test_smap = StateMachineAccessPoint(test_device)
+
+# state machine
+test_server = ServerStateMachine()
+
+# bind everything together
+bind(test_application, test_asap, test_smap, test_server)
+
+# ==============================================================================
+
+i_am_request = IAmRequest(
+    iAmDeviceIdentifier=('device', 100),
+    maxAPDULengthAccepted=1024,
+    segmentationSupported='segmentedBoth',
+    vendorID=15,
+    )
+print("i_am_request")
+i_am_request.debug_contents()
+print("")
+
+test_apdu = APDU()
+i_am_request.encode(test_apdu)
+
+print("test_apdu")
+test_apdu.debug_contents()
+print("")
+
+print("modify test_apdu")
+test_apdu.pduData = test_apdu.pduData[5:]
+test_apdu.debug_contents()
+print("")
+
+# make a send transition from start to success
+test_server.start_state.send(test_apdu).success()
+
+# run the machine
+print("running")
+test_server.run()
+print("")
+
+# ==============================================================================
+
+# check for success
+assert not test_server.running
+assert test_server.current_state.is_success_state
+
+tearDown()

sandbox/read_property_reject_test.py

@@ -0,0 +1,199 @@
+#!/usr/bin/python
+
+"""
+BACpypes Test
+-------------
+"""
+
+import os
+import sys
+
+from bacpypes.debugging import bacpypes_debugging, ModuleLogger, xtob
+from bacpypes.consolelogging import ArgumentParser
+
+from bacpypes.pdu import Address
+from bacpypes.comm import bind
+
+from bacpypes.apdu import APDU, ReadPropertyRequest, \
+    ComplexAckPDU, RejectPDU, AbortPDU
+
+from bacpypes.app import LocalDeviceObject, Application
+from bacpypes.appservice import ApplicationServiceAccessPoint, StateMachineAccessPoint
+
+from tests.state_machine import Server, StateMachine
+
+# some debugging
+_debug = 0
+_log = ModuleLogger(globals())
+
+# defaults for testing
+BACPYPES_TEST = ""
+BACPYPES_TEST_OPTION = ""
+
+# parsed test options
+test_options = None
+
+
+#
+#   TestApplication
+#
+
+@bacpypes_debugging
+class TestApplication(Application):
+
+    def __init__(self, localDevice, localAddress, aseID=None):
+        if _debug: TestApplication._debug("__init__ %r %r aseID=%r", localDevice, localAddress, aseID)
+        Application.__init__(self, localDevice, localAddress, aseID)
+
+#
+#   setUp
+#
+
+@bacpypes_debugging
+def setUp(argv=None):
+    global test_options
+
+    # create an argument parser
+    parser = ArgumentParser(description=__doc__)
+
+    # add an option
+    parser.add_argument(
+        '--option', help="this is an option",
+        default=os.getenv("BACPYPES_TEST_OPTION") or BACPYPES_TEST_OPTION,
+        )
+
+    # get the debugging args and parse them
+    arg_str = os.getenv("BACPYPES_TEST") or BACPYPES_TEST
+    test_options = parser.parse_args(argv or arg_str.split())
+
+    if _debug: setUp._debug("setUp")
+    if _debug: setUp._debug("    - test_options: %r", test_options)
+
+#
+#   tearDown
+#
+
+@bacpypes_debugging
+def tearDown():
+    if _debug: tearDown._debug("tearDown")
+
+
+#
+#   MatchingStateMachine
+#
+
+
+@bacpypes_debugging
+class MatchingStateMachine(Server, StateMachine):
+
+    def __init__(self):
+        if _debug: MatchingStateMachine._debug("__init__")
+
+        Server.__init__(self)
+        StateMachine.__init__(self)
+
+    def send(self, pdu):
+        if _debug: MatchingStateMachine._debug("send %r", pdu)
+        self.response(pdu)
+
+    def indication(self, pdu):
+        if _debug: MatchingStateMachine._debug("indication %r", pdu)
+        self.receive(pdu)
+
+    def match_pdu(self, pdu, transition_pdu):
+        if _debug: MatchingStateMachine._debug("match_pdu %r %r", pdu, transition_pdu)
+
+        # instance types match is enough
+        is_instance = isinstance(pdu, transition_pdu)
+        if _debug: MatchingStateMachine._debug("    - is_instance: %r", is_instance)
+
+        return is_instance
+
+#
+#   main
+#
+
+setUp(sys.argv[1:])
+
+# make a device object
+test_device = LocalDeviceObject(
+    objectName='test_device',
+    objectIdentifier=100,
+    maxApduLengthAccepted=1024,
+    segmentationSupported='segmentedBoth',
+    vendorIdentifier=15,
+    )
+
+# make a test address
+test_address = Address(1)
+
+# create a client state machine, trapped server, and bind them together
+test_application = TestApplication(test_device, test_address)
+
+print("objects: " + str(test_application.objectIdentifier))
+
+# get the services supported
+services_supported = test_application.get_services_supported()
+if _debug: _log.debug("    - services_supported: %r", services_supported)
+
+# let the device object know
+test_device.protocolServicesSupported = services_supported.value
+
+# include a application decoder
+test_asap = ApplicationServiceAccessPoint()
+
+# pass the device object to the state machine access point so it
+# can know if it should support segmentation
+test_smap = StateMachineAccessPoint(test_device)
+
+# state machine
+test_server = MatchingStateMachine()
+
+# bind everything together
+bind(test_application, test_asap, test_smap, test_server)
+
+# ==============================================================================
+
+read_property_request = ReadPropertyRequest(
+    objectIdentifier=('device', 100),
+    propertyIdentifier='objectName',
+    )
+read_property_request.pduSource = Address(2)
+# read_property_request.pduDestination = Address(1)
+read_property_request.apduInvokeID = 1
+
+print("read_property_request")
+read_property_request.debug_contents()
+print("")
+
+test_apdu = APDU()
+read_property_request.encode(test_apdu)
+
+print("test_apdu")
+test_apdu.debug_contents()
+print("")
+
+if 0:
+    print("modify test_apdu")
+    test_apdu.pduData = test_apdu.pduData[:5]
+    test_apdu.debug_contents()
+    print("")
+
+# make a send transition from start to success
+start_state = test_server.start_state.doc("start_state")
+response_state = start_state.send(test_apdu).doc("response_state")
+success_state = response_state.receive(ComplexAckPDU).doc("success_state")
+success_state.success()
+
+# run the machine
+print("running")
+test_server.run()
+print("")
+
+# ==============================================================================
+
+# check for success
+assert not test_server.running
+assert test_server.current_state.is_success_state
+
+tearDown()

sandbox/who_is_reject_test.py

@@ -0,0 +1,150 @@
+#!/usr/bin/python
+
+"""
+BACpypes Test
+-------------
+"""
+
+import os
+import sys
+
+from bacpypes.debugging import bacpypes_debugging, ModuleLogger, xtob
+from bacpypes.consolelogging import ArgumentParser
+
+from bacpypes.pdu import Address
+from bacpypes.comm import bind
+
+from bacpypes.apdu import APDU, WhoIsRequest
+
+from bacpypes.app import LocalDeviceObject, Application
+from bacpypes.appservice import ApplicationServiceAccessPoint, StateMachineAccessPoint
+
+from tests.state_machine import ServerStateMachine
+
+# some debugging
+_debug = 0
+_log = ModuleLogger(globals())
+
+# defaults for testing
+BACPYPES_TEST = ""
+BACPYPES_TEST_OPTION = ""
+
+# parsed test options
+test_options = None
+
+
+#
+#   TestApplication
+#
+
+@bacpypes_debugging
+class TestApplication(Application):
+
+    def __init__(self, localDevice, localAddress, aseID=None):
+        if _debug: TestApplication._debug("__init__ %r %r aseID=%r", localDevice, localAddress, aseID)
+        Application.__init__(self, localDevice, localAddress, aseID)
+
+#
+#   setUp
+#
+
+@bacpypes_debugging
+def setUp(argv=None):
+    global test_options
+
+    # create an argument parser
+    parser = ArgumentParser(description=__doc__)
+
+    # add an option
+    parser.add_argument(
+        '--option', help="this is an option",
+        default=os.getenv("BACPYPES_TEST_OPTION") or BACPYPES_TEST_OPTION,
+        )
+
+    # get the debugging args and parse them
+    arg_str = os.getenv("BACPYPES_TEST") or BACPYPES_TEST
+    test_options = parser.parse_args(argv or arg_str.split())
+
+    if _debug: setUp._debug("setUp")
+    if _debug: setUp._debug("    - test_options: %r", test_options)
+
+#
+#   tearDown
+#
+
+@bacpypes_debugging
+def tearDown():
+    if _debug: tearDown._debug("tearDown")
+
+#
+#   main
+#
+
+setUp(sys.argv[1:])
+
+# make a device object
+test_device = LocalDeviceObject(
+    objectName='test_device',
+    objectIdentifier=100,
+    maxApduLengthAccepted=1024,
+    segmentationSupported='segmentedBoth',
+    vendorIdentifier=15,
+    )
+
+# make a test address
+test_address = Address(1)
+
+# create a client state machine, trapped server, and bind them together
+test_application = TestApplication(test_device, test_address)
+
+# include a application decoder
+test_asap = ApplicationServiceAccessPoint()
+
+# pass the device object to the state machine access point so it
+# can know if it should support segmentation
+test_smap = StateMachineAccessPoint(test_device)
+
+# state machine
+test_server = ServerStateMachine()
+
+# bind everything together
+bind(test_application, test_asap, test_smap, test_server)
+
+# ==============================================================================
+
+who_is_request = WhoIsRequest(
+    deviceInstanceRangeLowLimit=0,
+    deviceInstanceRangeHighLimit=4194303,
+    )
+print("who_is_request")
+who_is_request.debug_contents()
+print("")
+
+test_apdu = APDU()
+who_is_request.encode(test_apdu)
+
+print("test_apdu")
+test_apdu.debug_contents()
+print("")
+
+print("modify test_apdu")
+test_apdu.pduData = test_apdu.pduData[:-1]
+# test_apdu.pduData = xtob('7509006869207468657265') # CharacterString("hi there")
+test_apdu.debug_contents()
+print("")
+
+# make a send transition from start to success
+test_server.start_state.send(test_apdu).success()
+
+# run the machine
+print("running")
+test_server.run()
+print("")
+
+# ==============================================================================
+
+# check for success
+assert not test_server.running
+assert test_server.current_state.is_success_state
+
+tearDown()