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

import unittest

from pydantic import ValidationError

from app.schemas.bacnet import BacnetBBMDWhoIsRequest, BacnetPointReadRequest, BacnetPointSearchRequest, bac0_object_type


class BacnetSchemaTest(unittest.TestCase):
    def test_port_defaults_to_bacnet_ip_port(self):
        request = BacnetPointSearchRequest(ip="192.168.75.240", bacnet_device_id=12345)

        self.assertEqual(47808, request.port)

    def test_search_request_only_requires_device_fields(self):
        request = BacnetPointSearchRequest(ip="192.168.75.240", bacnet_device_id=12345)

        self.assertEqual("192.168.75.240", request.ip)
        self.assertEqual(12345, request.bacnet_device_id)

    def test_ip_must_be_valid_address(self):
        with self.assertRaises(ValidationError) as context:
            BacnetPointSearchRequest(ip="not-an-ip", bacnet_device_id=12345)

        self.assertIn("ip must be a valid IP address", str(context.exception))

    def test_device_id_range_is_validated(self):
        with self.assertRaises(ValidationError) as context:
            BacnetPointSearchRequest(ip="192.168.75.240", bacnet_device_id=4_194_304)

        self.assertIn("bacnet_device_id", str(context.exception))

    def test_points_must_not_be_empty(self):
        with self.assertRaises(ValidationError) as context:
            BacnetPointReadRequest(ip="192.168.75.240", bacnet_device_id=12345, points=[])

        self.assertIn("points", str(context.exception))

    def test_object_type_is_normalized(self):
        request = BacnetPointReadRequest(
            ip="192.168.75.240",
            bacnet_device_id=12345,
            points=[{"object_type": "analog-input", "object_id": 1}],
        )

        self.assertEqual("AnalogInput", request.points[0].object_type)
        self.assertEqual("analogInput", bac0_object_type(request.points[0].object_type))

    def test_bbmd_whois_defaults_to_full_device_range(self):
        request = BacnetBBMDWhoIsRequest(bbmd_ip="192.168.1.1")

        self.assertEqual(47808, request.bbmd_port)
        self.assertEqual(60, request.ttl)
        self.assertEqual(0, request.low_limit)
        self.assertEqual(4_194_303, request.high_limit)

    def test_bbmd_whois_range_must_be_ordered(self):
        with self.assertRaises(ValidationError) as context:
            BacnetBBMDWhoIsRequest(bbmd_ip="192.168.1.1", low_limit=10, high_limit=1)

        self.assertIn("low_limit must be less than or equal to high_limit", str(context.exception))


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