Modbus-rtu/HostComputer/PythonTest/test_plsr_test.py

import unittest

import plsr_test


class FakeSerial:
    def __init__(self, response):
        self.response = response
        self.rx = bytearray()
        self.writes = []

    @property
    def in_waiting(self):
        return len(self.rx)

    def reset_input_buffer(self):
        self.rx.clear()

    def write(self, data):
        self.writes.append(bytes(data))
        self.rx.extend(bytes.fromhex("40 00") + self.response)
        return len(data)

    def flush(self):
        pass

    def read(self, size):
        chunk_size = min(size, 3, len(self.rx))
        data = bytes(self.rx[:chunk_size])
        del self.rx[:chunk_size]
        return data


class ProtocolTests(unittest.TestCase):
    def test_known_crc(self):
        self.assertEqual(plsr_test.crc16(bytes.fromhex("01 03 00 00 00 0A")), 0xCDC5)

    def test_status_request(self):
        frame = plsr_test.read_holding_request(1, 0, 32)
        self.assertEqual(frame.hex(" "), "01 03 00 00 00 20 44 12")

    def test_write_register_request(self):
        frame = plsr_test.write_register_request(1, 100, 1)
        self.assertEqual(frame, plsr_test.with_crc(bytes.fromhex("01 06 00 64 00 01")))

    def test_parser_recovers_from_stale_bytes(self):
        response = plsr_test.with_crc(bytes.fromhex("01 03 04 00 05 00 06"))
        buffer = bytearray(bytes.fromhex("40 00 00 03 40") + response)
        self.assertEqual(plsr_test.extract_response(buffer, 1, 3), response)

    def test_parser_waits_for_partial_frame(self):
        response = plsr_test.with_crc(bytes.fromhex("01 03 04 00 05 00 06"))
        buffer = bytearray(response[:5])
        self.assertIsNone(plsr_test.extract_response(buffer, 1, 3))
        buffer.extend(response[5:])
        self.assertEqual(plsr_test.extract_response(buffer, 1, 3), response)

    def test_snapshot_combines_high_and_low_words(self):
        snapshot = plsr_test.Snapshot(5, 2, 10, 0, 0, (1 << 16) | 2, 4, 1)
        self.assertEqual(snapshot.total_pulses, 65538)
        self.assertEqual(snapshot.state_name, "Done")

    def test_spec_snapshot_reads_stop_reason_and_signed_total(self):
        class SnapshotClient(plsr_test.ModbusRtuClient):
            def read_holding(self, address, count):
                if address == plsr_test.REG_MONITOR_TOTAL_PULSES:
                    return [0xFFFF, 0xFFFE, 0, 2, 6, 1, 0, 1]
                return [2, 1]

        snapshot = SnapshotClient("FAKE", protocol=plsr_test.MAP_SPEC).read_snapshot()
        self.assertEqual(snapshot.total_pulses, -2)
        self.assertEqual(snapshot.stop_reason, 1)
        self.assertEqual(snapshot.state_name, "Stopped")

    def test_client_handles_fragmented_noisy_response(self):
        response = plsr_test.with_crc(bytes.fromhex("01 03 04 00 05 00 06"))
        fake = FakeSerial(response)
        client = plsr_test.ModbusRtuClient("FAKE", timeout=0.1, retries=0)
        client._serial = fake
        self.assertEqual(client.read_holding(0, 2), [5, 6])
        self.assertEqual(len(fake.writes), 1)

    def test_signed_double_word_round_trip(self):
        self.assertEqual(plsr_test.encode_i32(-2147483648), (0x8000, 0x0000))
        self.assertEqual(plsr_test.encode_i32(-1), (0xFFFF, 0xFFFF))
        self.assertEqual(plsr_test.decode_i32(0x7FFF, 0xFFFF), 2147483647)
        self.assertEqual(plsr_test.decode_i32(0x8000, 0), -2147483648)

    def test_spec_configuration_validation(self):
        cfg = plsr_test.PlsrConfiguration(
            pulse_output=3, direction_output=3, segment_count=10,
            segments=[plsr_test.SegmentParameters(100000, -1, 4, 10, 20, 0) for _ in range(10)],
        )
        cfg.validate()
        zero_speed_cfg = plsr_test.PlsrConfiguration(
            start_speed_hz=0,
            end_speed_hz=0,
            segments=[plsr_test.SegmentParameters(0, 1)] + [plsr_test.SegmentParameters() for _ in range(9)],
        )
        zero_speed_cfg.validate()
        with self.assertRaisesRegex(ValueError, "frequency must be 1..100000 Hz"):
            plsr_test.PlsrConfiguration(default_speed_hz=0).validate()
        with self.assertRaises(ValueError):
            plsr_test.PlsrConfiguration(segment_count=11).validate()

    def test_send_and_run_modes_accept_only_zero_or_one(self):
        config = plsr_test.PlsrConfiguration(send_mode=1, run_mode=1)
        config.validate()
        with self.assertRaises(ValueError):
            plsr_test.PlsrConfiguration(send_mode=2).validate()
        with self.assertRaises(ValueError):
            plsr_test.PlsrConfiguration(run_mode=2).validate()

    def test_spec_segment_request_uses_fc16_and_signed_count(self):
        frame = plsr_test.write_registers_request(
            1, plsr_test.REG_SEGMENT_BASE,
            [100000 >> 16, 100000 & 0xFFFF, 0xFFFF, 0xFFFF, 4, 10, 20, 0],
        )
        self.assertEqual(frame[1], 0x10)
        self.assertEqual(frame[2:6], bytes.fromhex("11 00 00 08"))
        self.assertEqual(frame[6], 16)

    def test_spec_commands_have_independent_control_bits(self):
        client = plsr_test.ModbusRtuClient("FAKE", protocol=plsr_test.MAP_SPEC)
        writes = []
        client.write_register = lambda address, value: writes.append((address, value))

        client.command(plsr_test.COMMAND_STOP)
        client.command(plsr_test.COMMAND_IMMEDIATE_STOP)
        client.command(plsr_test.COMMAND_CLEAR_COUNT)
        client.command(plsr_test.COMMAND_RESET)
        client.command(plsr_test.COMMAND_RESTART)

        self.assertEqual(writes, [
            (plsr_test.REG_CONTROL, plsr_test.CONTROL_STOP),
            (plsr_test.REG_CONTROL, plsr_test.CONTROL_IMMEDIATE_STOP),
            (plsr_test.REG_CONTROL, plsr_test.CONTROL_CLEAR_COUNT),
            (plsr_test.REG_CONTROL, plsr_test.CONTROL_RESET),
            (plsr_test.REG_CONTROL, plsr_test.CONTROL_RESTART),
        ])

    def test_running_frequency_updates_current_segment_only(self):
        client = plsr_test.ModbusRtuClient(
            "FAKE", protocol=plsr_test.MAP_SPEC
        )
        snapshots = [
            plsr_test.Snapshot(
                2, 1, 1000, 0, 0, 25, 0, 0, 3, plsr_test.MAP_SPEC
            ),
            plsr_test.Snapshot(
                2, 1, 5000, 0, 0, 25, 0, 0, 3, plsr_test.MAP_SPEC
            ),
        ]
        writes = []
        client.read_snapshot = lambda: snapshots.pop(0)
        client.write_u32 = lambda address, value: writes.append((address, value))

        result = client.set_running_frequency(5000)

        expected_address = (
            plsr_test.REG_SEGMENT_BASE + 2 * plsr_test.REG_SEGMENT_STRIDE
        )
        self.assertEqual(writes, [(expected_address, 5000)])
        self.assertEqual(result.current_frequency, 5000)


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