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

import asyncio
import unittest

from app.schemas.bacnet import BacnetPointReadRequest
from app.services import bacnet_service


class FailingBacnetContext:
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        return False

    async def read(self, _args):
        raise RuntimeError("timeout")


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