data-collector-debugtool/intergration/bacnet/test_bacnet_service.py

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()