#!/usr/bin/env python # -*- coding: utf-8 -*- """ Test Device Services -------------------- """ import unittest from bacpypes.debugging import bacpypes_debugging, ModuleLogger, xtob from bacpypes.pdu import Address, LocalBroadcast, PDU from bacpypes.apdu import ( WhoIsRequest, IAmRequest, WhoHasRequest, WhoHasLimits, WhoHasObject, IHaveRequest, DeviceCommunicationControlRequest, SimpleAckPDU, ) from bacpypes.service.device import ( WhoIsIAmServices, WhoHasIHaveServices, DeviceCommunicationControlServices, ) from .helpers import ApplicationNetwork, ApplicationNode # some debugging _debug = 0 _log = ModuleLogger(globals()) @bacpypes_debugging class TestBasic(unittest.TestCase): def test_basic(self): """Test basic configuration of a network.""" if _debug: TestBasic._debug("test_basic") # create a network anet = ApplicationNetwork() # all start states are successful anet.td.start_state.success() anet.iut.start_state.success() # run the group anet.run() @bacpypes_debugging class TestWhoIsIAm(unittest.TestCase): def test_whois_unconstrained(self): """Test an unconstrained WhoIs, all devices respond.""" if _debug: TestWhoIsIAm._debug("test_whois_unconstrained") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) # all start states are successful anet.td.start_state.doc("1-1-0") \ .send(WhoIsRequest(destination=anet.vlan.broadcast_address)).doc("1-1-1") \ .receive(IAmRequest, pduSource=anet.iut.address).doc("1-1-2") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_whois_range_below(self): """Test range below.""" if _debug: TestWhoIsIAm._debug("test_whois_range_below") # create a network anet = ApplicationNetwork() # add the service capability to the iut anet.iut.add_capability(WhoIsIAmServices) # all start states are successful anet.td.start_state.doc("2-1-0") \ .send(WhoIsRequest( destination=anet.vlan.broadcast_address, deviceInstanceRangeLowLimit=0, deviceInstanceRangeHighLimit=9, )).doc("2-1-1") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_whois_range_above(self): """Test range above.""" if _debug: TestWhoIsIAm._debug("test_whois_range_above") # create a network anet = ApplicationNetwork() # add the service capability to the iut anet.iut.add_capability(WhoIsIAmServices) # all start states are successful anet.td.start_state.doc("3-1-0") \ .send(WhoIsRequest( destination=anet.vlan.broadcast_address, deviceInstanceRangeLowLimit=21, deviceInstanceRangeHighLimit=29, )).doc("3-1-1") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_whois_range(self): """Test a WhoIs, included range.""" if _debug: TestWhoIsIAm._debug("test_whois_range") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) # all start states are successful anet.td.start_state.doc("4-1-0") \ .send(WhoIsRequest( destination=anet.vlan.broadcast_address, deviceInstanceRangeLowLimit=19, deviceInstanceRangeHighLimit=21, )).doc("4-1-1") \ .receive(IAmRequest, pduSource=anet.iut.address).doc("4-1-2") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() @bacpypes_debugging class TestWhoHasIHave(unittest.TestCase): def test_who_has_object_by_name(self): """Test an unconstrained WhoIs, all devices respond.""" if _debug: TestWhoIsIAm._debug("test_who_has_object_by_name") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoHasIHaveServices) # all start states are successful anet.td.start_state.doc("5-1-0") \ .send(WhoHasRequest( destination=anet.vlan.broadcast_address, object=WhoHasObject(objectName="iut"), )).doc("5-1-1") \ .receive(IHaveRequest, pduSource=anet.iut.address).doc("5-1-2") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_who_has_object_by_id(self): """Test an unconstrained WhoIs, all devices respond.""" if _debug: TestWhoIsIAm._debug("test_who_has_object_by_id") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoHasIHaveServices) # all start states are successful anet.td.start_state.doc("6-1-0") \ .send(WhoHasRequest( destination=anet.vlan.broadcast_address, object=WhoHasObject(objectIdentifier=('device', 20)), )).doc("6-1-1") \ .receive(IHaveRequest, pduSource=anet.iut.address).doc("6-1-2") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() @bacpypes_debugging class TestDeviceCommunicationControl(unittest.TestCase): def test_default_behavior(self): """Test.""" if _debug: TestDeviceCommunicationControl._debug("test_default_behavior") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) anet.iut.add_capability(DeviceCommunicationControlServices) # test sequence anet.td.start_state.doc("7-1-0") \ .send(WhoIsRequest(destination=anet.vlan.broadcast_address)).doc("7-1-1") \ .receive(IAmRequest, pduSource=anet.iut.address).doc("7-1-2") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_disable(self): """Test.""" if _debug: TestDeviceCommunicationControl._debug("test_disable") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) anet.iut.add_capability(DeviceCommunicationControlServices) # test sequence anet.td.start_state.doc("7-2-0") \ .send(DeviceCommunicationControlRequest( destination=anet.iut.address, enableDisable='disable', )).doc("7-2-1") \ .receive(SimpleAckPDU).doc("7-2-2") \ .send(WhoIsRequest(destination=anet.vlan.broadcast_address)).doc("7-2-3") \ .timeout(10).doc("7-2-4") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_disable_initiation(self): """Test disabling initiation. After the DCC request send the IUT a WhoIsRequest and verify that the IAmRequest makes it back. """ if _debug: TestDeviceCommunicationControl._debug("test_disable") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) anet.iut.add_capability(DeviceCommunicationControlServices) # test sequence anet.td.start_state.doc("7-3-0") \ .send(DeviceCommunicationControlRequest( destination=anet.iut.address, enableDisable='disableInitiation', )).doc("7-3-1") \ .receive(SimpleAckPDU).doc("7-3-2") \ .send(WhoIsRequest(destination=anet.vlan.broadcast_address)).doc("7-3-3") \ .receive(IAmRequest, pduSource=anet.iut.address).doc("7-3-4") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group anet.run() def test_disable_time_duration(self): """Test disabling communication for a specific amount of time in minutes. After turning off communications, wait for 30 seconds and send a request and nothing should come back. Wait an additional 30 seconds and try again, this tim receiving the response.""" if _debug: TestDeviceCommunicationControl._debug("test_disable_time_duration") # create a network anet = ApplicationNetwork() # add the service capability to the IUT anet.iut.add_capability(WhoIsIAmServices) anet.iut.add_capability(DeviceCommunicationControlServices) # test sequence anet.td.start_state.doc("7-4-0") \ .send(DeviceCommunicationControlRequest( destination=anet.iut.address, enableDisable='disable', timeDuration=1, )).doc("7-4-1") \ .receive(SimpleAckPDU).doc("7-4-2") \ .timeout(30).doc("7-4-3") \ .send(WhoIsRequest( destination=anet.vlan.broadcast_address, )).doc("7-4-4") \ .timeout(30).doc("7-4-5") \ .send(WhoIsRequest( destination=anet.vlan.broadcast_address, )).doc("7-4-6") \ .receive(IAmRequest, pduSource=anet.iut.address).doc("7-4-7") \ .success() # no IUT application layer matching anet.iut.start_state.success() # run the group a little longer than a minute anet.run(65)