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

import unittest

from common import assert_bacnet_device, assert_response_contract, device_payload, post_json


class BacnetApiIntegrationTest(unittest.TestCase):
    def test_validation_error_returns_http_200_contract(self):
        status, data = post_json(
            "/api/dc-debugtool/bacnet/read_points",
            {**device_payload(), "points": []},
        )

        assert_response_contract(self, status, data)
        self.assertEqual(1, data["code"], data)
        self.assertIn("points", data["msg"])
        self.assertEqual([], data["data"]["points"])

    def test_search_points_returns_bacnet_points(self):
        payload = device_payload()

        status, data = post_json("/api/dc-debugtool/bacnet/search_points", payload)

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        assert_bacnet_device(self, payload, data)
        points = data["data"]["points"]
        self.assertIsInstance(points, list)
        self.assertTrue(points, data)
        for point in points:
            self.assertIn("name", point)
            self.assertIn("description", point)
            self.assertIn("object_type", point)
            self.assertIn("object_id", point)
            self.assertIn("present_value", point)

    def test_read_points_uses_search_result(self):
        payload = device_payload()
        status, search_data = post_json("/api/dc-debugtool/bacnet/search_points", payload)
        assert_response_contract(self, status, search_data)
        self.assertEqual(0, search_data["code"], search_data)
        points = search_data["data"]["points"]
        self.assertTrue(points, search_data)

        source_point = next((point for point in points if point.get("present_value") is not None), points[0])
        read_payload = {
            **payload,
            "points": [
                {
                    "object_type": source_point["object_type"],
                    "object_id": source_point["object_id"],
                }
            ],
        }

        status, data = post_json("/api/dc-debugtool/bacnet/read_points", read_payload)

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        assert_bacnet_device(self, payload, data)
        self.assertEqual(1, len(data["data"]["points"]))
        point = data["data"]["points"][0]
        self.assertEqual(source_point["object_type"], point["object_type"])
        self.assertEqual(source_point["object_id"], point["object_id"])
        self.assertIn("present_value", point)


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