From 15ab79ffa41344e67b4d35cbb7fac6784d019cd9 Mon Sep 17 00:00:00 2001 From: Joel Bender Date: Thu, 25 Feb 2016 01:10:09 -0500 Subject: [PATCH] add some reject testing code to the sandbox --- sandbox/i_am_reject_test.py | 151 ++++++++++++++++++++ sandbox/read_property_reject_test.py | 199 +++++++++++++++++++++++++++ sandbox/who_is_reject_test.py | 150 ++++++++++++++++++++ 3 files changed, 500 insertions(+) create mode 100755 sandbox/i_am_reject_test.py create mode 100755 sandbox/read_property_reject_test.py create mode 100755 sandbox/who_is_reject_test.py diff --git a/sandbox/i_am_reject_test.py b/sandbox/i_am_reject_test.py new file mode 100755 index 0000000..09a8fce --- /dev/null +++ b/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() diff --git a/sandbox/read_property_reject_test.py b/sandbox/read_property_reject_test.py new file mode 100755 index 0000000..cc59919 --- /dev/null +++ b/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() diff --git a/sandbox/who_is_reject_test.py b/sandbox/who_is_reject_test.py new file mode 100755 index 0000000..6ec4423 --- /dev/null +++ b/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()