import asyncio import os import unittest from unittest.mock import patch from app.schemas.bacnet import BacnetPointReadRequest, BacnetPointSearchRequest from app.services import bacnet_service class FailingBacnetApp: async def read_property(self, *_args): raise RuntimeError("timeout") class FailingBacnetContext: def __init__(self): self.this_application = type("ThisApplication", (), {"app": FailingBacnetApp()})() async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): return False class SearchBacnetApp: async def read_property(self, _address, objid, prop, array_index=None): if str(objid) == "device,12345" and str(prop) == "object-list" and array_index == 0: return 1 if str(objid) == "device,12345" and str(prop) == "object-list" and array_index == 1: return ("analog-input", 1) if str(objid) == "analog-input,1" and str(prop) == "object-name": return "ai-1" if str(objid) == "analog-input,1" and str(prop) == "description": return "temperature" if str(objid) == "analog-input,1" and str(prop) == "present-value": return 12.3 raise RuntimeError("unexpected read") class SearchBacnetContext: def __init__(self): self.this_application = type("ThisApplication", (), {"app": SearchBacnetApp()})() async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb): return False class BacnetServiceTest(unittest.TestCase): def test_read_points_returns_clear_no_response_error(self): original_start = bacnet_service.BAC0.start original_local_ip = bacnet_service.local_bacnet_ip original_local_port = bacnet_service.get_free_udp_port bacnet_service.BAC0.start = lambda **_kwargs: FailingBacnetContext() bacnet_service.local_bacnet_ip = lambda _ip, _port: "127.0.0.1/24" bacnet_service.get_free_udp_port = lambda: 47809 self.addCleanup(setattr, bacnet_service.BAC0, "start", original_start) self.addCleanup(setattr, bacnet_service, "local_bacnet_ip", original_local_ip) self.addCleanup(setattr, bacnet_service, "get_free_udp_port", original_local_port) response = bacnet_service.read_points( BacnetPointReadRequest( ip="192.168.1.10", bacnet_device_id=12345, points=[{"object_type": "analogInput", "object_id": 1}], ) ) self.assertEqual(1, response["code"]) self.assertIn("no response from BACnet device 192.168.1.10:47808", response["msg"]) self.assertEqual([], response["data"]["points"]) def test_search_points_reads_object_list_by_index(self): original_start = bacnet_service.BAC0.start original_local_ip = bacnet_service.local_bacnet_ip original_local_port = bacnet_service.get_free_udp_port bacnet_service.BAC0.start = lambda **_kwargs: SearchBacnetContext() bacnet_service.local_bacnet_ip = lambda _ip, _port: "127.0.0.1/24" bacnet_service.get_free_udp_port = lambda: 47809 self.addCleanup(setattr, bacnet_service.BAC0, "start", original_start) self.addCleanup(setattr, bacnet_service, "local_bacnet_ip", original_local_ip) self.addCleanup(setattr, bacnet_service, "get_free_udp_port", original_local_port) response = bacnet_service.search_points( BacnetPointSearchRequest(ip="192.168.1.10", bacnet_device_id=12345) ) self.assertEqual(0, response["code"]) self.assertEqual( [ { "name": "ai-1", "description": "temperature", "object_type": "AnalogInput", "object_id": 1, "present_value": 12.3, } ], response["data"]["points"], ) def test_ignores_cancelled_broadcast_endpoint_callback(self): class Loop: default_handler_called = False def default_exception_handler(self, _context): self.default_handler_called = True loop = Loop() bacnet_service.ignore_cancelled_bacnet_broadcast_endpoint( loop, { "message": "Exception in callback IPv4DatagramServer.set_broadcast_transport_protocol()", "exception": asyncio.CancelledError(), }, ) self.assertFalse(loop.default_handler_called) def test_builds_bbmd_whois_as_distributed_local_broadcast(self): packet = bacnet_service.bbmd_whois_packet(0, 4_194_303) self.assertEqual(0x81, packet[0]) self.assertEqual(0x09, packet[1]) self.assertEqual(len(packet), int.from_bytes(packet[2:4], "big")) self.assertEqual(b"\x01\x00\x10\x08", packet[4:8]) def test_parses_forwarded_i_am_packet(self): original_sender = b"\xc0\xa8\x01\x0a\xba\xc0" object_id = ((8 << 22) | 12345).to_bytes(4, "big") apdu = b"\x10\x00" + b"\xc4" + object_id + b"\x22\x05\xc4" + b"\x91\x03" + b"\x22\x03\x4a" npdu = b"\x01\x00" + apdu payload = original_sender + npdu packet = b"\x81\x04" + (4 + len(payload)).to_bytes(2, "big") + payload device = bacnet_service.parse_i_am_packet(packet, ("192.168.1.1", 47808)) self.assertEqual( { "bacnet_device_id": 12345, "max_apdu": 1476, "segmentation": "noSegmentation", "vendor_id": 842, "ip": "192.168.1.10", "port": 47808, }, device, ) def test_bbmd_whois_requires_configured_bbmd_ip(self): with patch.dict(os.environ, {}, clear=True): response = bacnet_service.bbmd_whois() self.assertEqual(1, response["code"]) self.assertIn("BACNET_BBMD_IP environment variable is required", response["msg"]) self.assertEqual([], response["data"]["devices"]) def test_builds_bbmd_whois_request_from_environment(self): with patch.dict( os.environ, { "BACNET_BBMD_IP": "192.168.1.1", "BACNET_BBMD_PORT": "47809", "BACNET_BBMD_TTL": "120", "BACNET_BBMD_WHOIS_TIMEOUT": "7.5", }, clear=True, ): request = bacnet_service.bbmd_whois_request_from_env() self.assertEqual("192.168.1.1", request.bbmd_ip) self.assertEqual(47809, request.bbmd_port) self.assertEqual(120, request.ttl) self.assertEqual(7.5, request.timeout) if __name__ == "__main__": unittest.main()