data-collector-debugtool/intergration/s7/test_s7_schema.py

import unittest

from pydantic import ValidationError

from app.schemas.s7 import S7ConnectScanRequest, S7PointReadRequest, S7RawReadRequest


class S7SchemaTest(unittest.TestCase):
    def test_device_type_port_and_connection_type_defaults(self):
        request = S7RawReadRequest(
            ip="127.0.0.1",
            rock=0,
            slot=1,
            read={"area": "db", "db": 1, "start": 0, "size": 4},
        )

        self.assertEqual("S7-1200", request.device_type)
        self.assertEqual(102, request.port)
        self.assertEqual("PG", request.tsap_conn_type)
        self.assertEqual("DB", request.read.area)

    def test_tsap_conn_type_is_case_insensitive(self):
        request = S7RawReadRequest(
            ip="127.0.0.1",
            rock=0,
            slot=1,
            tsap_conn_type="basic",
            read={"area": "m", "start": 0, "size": 1},
        )

        self.assertEqual("BASIC", request.tsap_conn_type)
        self.assertEqual("M", request.read.area)

    def test_device_type_is_case_insensitive(self):
        request = S7RawReadRequest(
            device_type="s7-smart200",
            ip="127.0.0.1",
            rock=0,
            slot=1,
            read={"area": "v", "start": 0, "size": 1},
        )

        self.assertEqual("S7-Smart200", request.device_type)
        self.assertEqual("V", request.read.area)

    def test_v_area_requires_smart200_device_type(self):
        with self.assertRaises(ValidationError) as context:
            S7RawReadRequest(
                device_type="S7-1200",
                ip="127.0.0.1",
                rock=0,
                slot=1,
                read={"area": "V", "start": 0, "size": 1},
            )

        self.assertIn("V area is only supported", str(context.exception))

    def test_db_area_requires_positive_db_number(self):
        with self.assertRaises(ValidationError) as context:
            S7RawReadRequest(
                ip="127.0.0.1",
                rock=0,
                slot=1,
                read={"area": "DB", "db": 0, "start": 0, "size": 1},
            )

        self.assertIn("db must be greater than 0", str(context.exception))

    def test_bit_is_only_allowed_for_bool_points(self):
        with self.assertRaises(ValidationError) as context:
            S7PointReadRequest(
                ip="127.0.0.1",
                rock=0,
                slot=1,
                points=[{"area": "DB", "db": 1, "start": 0, "type": "int16", "bit": 0}],
            )

        self.assertIn("bit is only supported for bool points", str(context.exception))

    def test_connect_scan_request_only_requires_ip(self):
        request = S7ConnectScanRequest(ip="127.0.0.1")

        self.assertEqual("127.0.0.1", request.ip)


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