tests/test_protocol.py - master - evg.drzhnn/Tenzo_HTTPBridge

tests/test_protocol.py
78 строк3 KB
DEV
init
23 июн 2026, 18:39
23 июн 2026, 18:39193097c
import socket
import sys
import unittest
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT / "src"))

import TenzoConnector as connector


class ProtocolTests(unittest.TestCase):
    def test_tenzom_get_weight_packet_matches_upstream(self):
        packet = connector.make_tenzom_packet(1, 0xC3)
        self.assertEqual(packet, bytes.fromhex("ff 01 c3 e3 ff ff"))

    def test_bcd_weight_and_decimal_position(self):
        result = connector.extract_weight_tenzom(bytes.fromhex("50 03 00 12"))
        self.assertEqual(result["weight_kg"], 3.5)
        self.assertTrue(result["stable"])
        self.assertFalse(result["overload"])

    def test_negative_weight(self):
        result = connector.extract_weight_tenzom(bytes.fromhex("50 03 00 92"))
        self.assertEqual(result["weight_kg"], -3.5)
        self.assertEqual(result["comment"], "вес стабилен, отрицательный")

    def test_invalid_bcd_is_rejected(self):
        with self.assertRaises(ValueError):
            connector.extract_weight_tenzom(bytes.fromhex("fa 03 00 12"))

    def test_tenzom_response_packet_is_found(self):
        packet = bytes.fromhex("00 ff 01 c3 50 03 00 12 e3 ff ff 00")
        found = connector.find_tenzom_packet(packet, 1, 0xC3)
        self.assertIsNotNone(found)
        self.assertEqual(found[0], bytes.fromhex("50 03 00 12"))

    def test_643_response(self):
        result = connector.parse_643_response(b"noise=003.500\x04")
        self.assertEqual(result["weight_kg"], 3.5)
        self.assertTrue(result["stable"])

    def test_643_status_can_shift_decimal_and_mark_unstable(self):
        result = connector.parse_643_response(b"=003.500\x01")
        self.assertEqual(result["weight_kg"], 0.35)
        self.assertFalse(result["stable"])

    def test_legacy_protocol_values_are_migrated(self):
        self.assertEqual(connector.normalize_protocol("1"), "tenzom")
        self.assertEqual(connector.normalize_protocol("3"), "tenzom")
        self.assertEqual(connector.normalize_protocol("2"), "643")

    def test_config_rejects_mixed_or_invalid_ports(self):
        with self.assertRaises(ValueError):
            connector.validate_config({"port": 70000})

    def test_fragmented_response_is_collected(self):
        reader, writer = socket.socketpair()
        try:
            writer.sendall(bytes.fromhex("ff 01 c3 50"))
            writer.sendall(bytes.fromhex("03 00 12 e3 ff ff"))
            reader.settimeout(0.2)
            response = connector.receive_response(
                reader,
                terminator=connector.TENZOM_END,
            )
            self.assertEqual(
                response,
                bytes.fromhex("ff 01 c3 50 03 00 12 e3 ff ff"),
            )
        finally:
            reader.close()
            writer.close()


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