shibosoftwaredev/f1c100s-linux-nema17-stepper-controller

firmware/test_motor_control.py

import importlib.util
from contextlib import redirect_stdout
import io
from pathlib import Path
import struct
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import MagicMock, patch

spec = importlib.util.spec_from_file_location("motor_control", Path(__file__).with_name("motor-control.py"))
client = importlib.util.module_from_spec(spec)
spec.loader.exec_module(client)


class CurrentClientTests(unittest.TestCase):
    def test_limit_decode(self):
        value = client.decode_limit(struct.pack("<BHHHH", 3, 200, 2066, 199, 3300))
        self.assertTrue(value["ready"] and value["dac_ok"])
        self.assertEqual(value["requested_mA"], 200)
        self.assertIsNone(client.decode_limit(struct.pack("<BHHHH", 0, 396, 0xffff, 0, 0))["dac_code"])
        for p in (b"", bytes(8), bytes(10), bytes([4])+bytes(8)):
            with self.assertRaises(ValueError):
                client.decode_limit(p)

    def test_set_limit_does_not_send_arm_motion_disable_or_heartbeat(self):
        link = MagicMock()
        link.request.side_effect = [b"", struct.pack("<BHHHH", 3, 200, 2066, 199, 3300)]
        serial = SimpleNamespace(Serial=MagicMock())
        with patch.object(sys, "argv", ["motor-control.py", "--set-current-ma", "200"]), \
                patch.dict(sys.modules, {"serial": serial}), \
                patch.object(client, "MotionLink", return_value=link), redirect_stdout(io.StringIO()):
            client.main()
        self.assertEqual(link.request.call_args_list,
                         [unittest.mock.call(client.CMD_SET_LIMIT, b"\xc8\x00"), unittest.mock.call(client.CMD_LIMIT)])

    def test_limit_read_is_read_only(self):
        link = MagicMock()
        link.request.return_value = struct.pack("<BHHHH", 3, 200, 2066, 199, 3300)
        serial = SimpleNamespace(Serial=MagicMock())
        with patch.object(sys, "argv", ["motor-control.py", "--limit"]), \
                patch.dict(sys.modules, {"serial": serial}), \
                patch.object(client, "MotionLink", return_value=link), redirect_stdout(io.StringIO()):
            client.main()
        link.request.assert_called_once_with(client.CMD_LIMIT)

    def test_bad_limit_is_rejected_before_serial_access(self):
        for n in ("-1", "397", "65535"):
            with patch.object(sys, "argv", ["motor-control.py", "--set-current-ma", n]), \
                    patch.object(client, "MotionLink") as link, \
                    patch.object(sys, "stderr", io.StringIO()), self.assertRaises(SystemExit):
                client.main()
            link.assert_not_called()

    def test_signed_values(self):
        value = client.decode_current(struct.pack("<BiiHHHH", 3, 400, -400, 3537, 558, 3300, 1))
        self.assertEqual(value["phase_a_mA"], 400)
        self.assertEqual(value["phase_b_mA"], -400)
        self.assertEqual(value["sample_age_ms"], 1)

    def test_invalid_and_clipped_not_reported_as_real_current(self):
        value = client.decode_current(struct.pack("<BiiHHHH", 12, -550, 550, 0, 4095, 3300, 30))
        self.assertIsNone(value["phase_a_mA"])
        self.assertIsNone(value["phase_b_mA"])
        self.assertTrue(value["phase_a_clipped"] and value["phase_b_clipped"])

    def test_protocol_lengths_and_flags(self):
        for payload in (b"", bytes(16), bytes(18), bytes([0x80]) + bytes(16),
                        bytes([5]) + bytes(16), bytes([10]) + bytes(16)):
            with self.assertRaises(ValueError):
                client.decode_current(payload)

    def test_legacy_crc_vector(self):
        self.assertEqual(client.crc16_ccitt(b"123456789"), 0x29B1)

    def test_current_cli_sends_only_the_read_command(self):
        link = MagicMock()
        link.request.return_value = struct.pack("<BiiHHHH", 3, 400, -400, 3537, 558, 3300, 1)
        serial = SimpleNamespace(Serial=MagicMock())
        with patch.object(sys, "argv", ["motor-control.py", "--current"]), \
                patch.dict(sys.modules, {"serial": serial}), \
                patch.object(client, "MotionLink", return_value=link), \
                redirect_stdout(io.StringIO()) as output:
            client.main()
        link.request.assert_called_once_with(client.CMD_CURRENT)
        self.assertIn('"phase_a_mA": 400', output.getvalue())

    def test_failed_current_query_sends_no_cleanup_motion_command(self):
        link = MagicMock()
        link.request.side_effect = TimeoutError("no response")
        serial = SimpleNamespace(Serial=MagicMock())
        with patch.object(sys, "argv", ["motor-control.py", "--current"]), \
                patch.dict(sys.modules, {"serial": serial}), \
                patch.object(client, "MotionLink", return_value=link):
            with self.assertRaises(TimeoutError):
                client.main()
        link.request.assert_called_once_with(client.CMD_CURRENT)


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