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

import struct
import unittest

from common import ApiTestServer, Snap7TestServer, assert_response_contract, patch_connect_scan, s7_device_payload


class S7ApiIntegrationTest(unittest.TestCase):
    def test_raw_read_returns_semantic_communication_from_simulated_server(self):
        db_data = bytearray([0x11, 0x22, 0x33, 0x44, 0x01])
        with Snap7TestServer(db_data) as s7_server, ApiTestServer() as api_server:
            payload = {
                **s7_device_payload(s7_server.port),
                "read": {"area": "DB", "db": 1, "start": 0, "size": 4},
            }

            status, data = api_server.post_json("/api/dc-debugtool/s7/read", payload)

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        self.assertEqual("success", data["msg"])
        self.assertEqual(s7_device_payload(payload["port"]), data["data"]["device"])
        self.assertNotIn("points", data["data"])

        communication = data["data"]["communication"]
        self.assertIn(
            f"Tx:S7_CONNECT ip=127.0.0.1 port={payload['port']} rock=0 slot=1 tsap_conn_type=PG",
            communication,
        )
        self.assertTrue(any(item.startswith("Rx:S7_CONNECTED") for item in communication), communication)
        self.assertIn("Tx:S7_READ area=DB db=1 start=0 size=4", communication)
        self.assertIn("Rx:11 22 33 44", communication)
        self.assertIn("Tx:S7_DISCONNECT", communication)
        self.assertIn("Rx:S7_DISCONNECTED", communication)

    def test_read_points_returns_values_without_communication_from_simulated_server(self):
        db_data = bytearray(struct.pack(">f", 12.34) + b"\x01" + (-123).to_bytes(2, "big", signed=True))
        with Snap7TestServer(db_data) as s7_server, ApiTestServer() as api_server:
            payload = {
                **s7_device_payload(s7_server.port),
                "points": [
                    {"area": "DB", "db": 1, "start": 0, "type": "float32"},
                    {"area": "DB", "db": 1, "start": 4, "type": "bool", "bit": 0},
                    {"area": "DB", "db": 1, "start": 5, "type": "int16"},
                ],
            }

            status, data = api_server.post_json("/api/dc-debugtool/s7/read_points", payload)

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        self.assertEqual(s7_device_payload(payload["port"]), data["data"]["device"])
        self.assertNotIn("communication", data["data"])

        points = data["data"]["points"]
        self.assertEqual(3, len(points))
        self.assertAlmostEqual(12.34, points[0]["value"], places=4)
        self.assertIs(True, points[1]["value"])
        self.assertEqual(0, points[1]["bit"])
        self.assertEqual(-123, points[2]["value"])

    def test_connect_scan_returns_only_available_connections(self):
        successful = {(0, 1, "PG"), (0, 2, "BASIC")}
        with patch_connect_scan(successful), ApiTestServer() as api_server:
            status, data = api_server.post_json("/api/dc-debugtool/s7/connect_scan", {"ip": "127.0.0.1"})

        assert_response_contract(self, status, data)
        self.assertEqual(0, data["code"], data)
        self.assertEqual({"device_type": "S7-1200", "ip": "127.0.0.1", "port": 102}, data["data"]["device"])
        self.assertEqual(
            [
                {"rock": 0, "slot": 1, "tsap_conn_type": "PG"},
                {"rock": 0, "slot": 2, "tsap_conn_type": "BASIC"},
            ],
            data["data"]["available"],
        )
        self.assertNotIn("results", data["data"])

    def test_validation_error_returns_http_200_contract(self):
        with ApiTestServer() as api_server:
            status, data = api_server.post_json(
                "/api/dc-debugtool/s7/read_points",
                {
                    "ip": "127.0.0.1",
                    "rock": 0,
                    "slot": 1,
                    "points": [{"area": "DB", "db": 0, "start": 0, "type": "int16"}],
                },
            )

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


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