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

import asyncio
import unittest

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)


if __name__ == "__main__":
    unittest.main()