fixed platform not moving problem

apps/autocal/tests/test_controller_sequence.py

@@ -0,0 +1,286 @@
+"""The controller-command sequence: what is actually sent, and when.
+
+These are the regressions that hardware found and the UI suite could not:
+
+* ``ZAux_Execute`` must be handed a real reply buffer. Passing ``NULL`` is not
+  an error the library reports — it writes through the null pointer and takes the
+  process down, so the assertion has to be on the *arguments*.
+* Re-initialisation must stop and start the task the program actually runs on,
+  and a firmware refusal that arrives only in the reply text must not be
+  reported as a re-initialisation in progress.
+* A session that adopts a program already running on the controller must still
+  be able to re-initialise it.
+
+No test here touches a controller or a DLL.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import time
+from pathlib import Path
+
+import pytest
+
+from orisys_autocal.motion import (
+    DEFAULT_BAS_TASK,
+    INIT_OK,
+    INIT_RUNNING,
+    ConnectionSpec,
+    DemoTransport,
+    MotionService,
+    clamp_bas_task,
+)
+from orisys_autocal.motion.transport import _run_filename
+
+#: The shipped programs, used as the file the service is asked to download.
+#: Real files rather than per-test temporaries: pytest's ``tmp_path`` needs to
+#: enumerate a directory outside the workspace, which a confined run denies.
+_BAS_DIR = Path(__file__).resolve().parents[1] / "config" / "bas"
+SHIPPED_ECAT = _BAS_DIR / "ECAT.bas"
+SHIPPED_RTEX = _BAS_DIR / "RTEX.bas"
+
+# ------------------------------------------------------------------ D1: the buffer
+
+
+class _RecordingFunc:
+    """A ``ZAux_Execute`` double: records its arguments, writes a reply."""
+
+    def __init__(self, reply: bytes = b"", code: int = 0) -> None:
+        self.calls: list[tuple] = []
+        self.reply = reply
+        self.code = code
+
+    def __call__(self, handle, command, response, length):
+        size = int(getattr(length, "value", length))
+        self.calls.append((handle, command, response, size))
+        if response is None:
+            raise AssertionError("ZAux_Execute was called with a NULL reply buffer")
+        if self.reply:
+            ctypes.memmove(response, self.reply, min(len(self.reply), size - 1))
+        return self.code
+
+
+class _FakeDll:
+    def __init__(self, execute: _RecordingFunc) -> None:
+        self.ZAux_Execute = execute
+
+
+@pytest.fixture
+def client_with_fake_dll():
+    from orisys_app_shared.zmotion import ZmotionClient
+
+    client = ZmotionClient()
+    execute = _RecordingFunc(reply=b"Online command error, not find.\n")
+    client._dll = _FakeDll(execute)  # noqa: SLF001 - the seam under test
+    client._handle = ctypes.c_void_p(0x1234)  # noqa: SLF001 - pretend connected
+    return client, execute
+
+
+def test_execute_passes_a_real_reply_buffer(client_with_fake_dll):
+    """The null-buffer bug: a NULL psResponse is a native crash, not an error."""
+    client, execute = client_with_fake_dll
+
+    result = client.execute('RUN "ECAT.bas",4')
+
+    assert result.code == 0
+    _handle, command, response, length = execute.calls[0]
+    assert command == b'RUN "ECAT.bas",4'
+    assert response is not None
+    assert length >= 1024
+
+
+def test_execute_returns_the_controllers_reply_text(client_with_fake_dll):
+    """The firmware explains a refused RUN only in the reply."""
+    client, _execute = client_with_fake_dll
+
+    result = client.execute('RUN "nope.bas",4')
+
+    assert "not find" in str(result.value)
+
+
+def test_transport_execute_hands_the_reply_back():
+    from orisys_autocal.motion.transport import NativeTransport
+
+    class _StubClient:
+        dll_dir = "stub"
+
+        def execute(self, command):
+            from orisys_app_shared.zmotion import ZmotionValue
+
+            return ZmotionValue(0, "controller says hello")
+
+    assert NativeTransport(_StubClient()).execute("?X") == "controller says hello"
+
+
+# ------------------------------------------------- D2: the re-init command shape
+
+
+class RecordingTransport(DemoTransport):
+    """Demo platform that keeps every online command it was sent."""
+
+    def __init__(self, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.commands: list[str] = []
+
+    def execute(self, command: str) -> str:
+        self.commands.append(command)
+        return super().execute(command)
+
+
+def _initialise(service, timeout: float = 5.0) -> None:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline and service.bus.init_status != INIT_OK:
+        service.poll([0])
+        time.sleep(0.02)
+
+
+def test_reinit_stops_and_runs_the_configured_task():
+    """Task 2 is not the vendor's task; stopping it restarts nothing."""
+    transport = RecordingTransport()
+    service = MotionService(transport, bas_task=4)
+    service.connect(ConnectionSpec(mode="demo"))
+    service.download_bas(str(SHIPPED_ECAT), 1)
+    _initialise(service)
+
+    transport.commands.clear()
+    result = service.reinit_bus()
+
+    assert result.ok, result.message
+    assert transport.commands == ["STOPTASK 4", 'RUN "ECAT.bas",4']
+    assert service.bus.init_status == INIT_RUNNING
+
+
+def test_reinit_task_number_comes_from_configuration():
+    transport = RecordingTransport()
+    service = MotionService(transport, bas_task=6)
+    service.connect(ConnectionSpec(mode="demo"))
+    service.download_bas(str(SHIPPED_RTEX), 1)
+    _initialise(service)
+
+    transport.commands.clear()
+    assert service.reinit_bus().ok
+    assert transport.commands == ["STOPTASK 6", 'RUN "RTEX.bas",6']
+
+
+def test_a_controller_refusal_in_the_reply_is_not_reported_as_running():
+    """``ZAux_Execute`` returns 0 for a RUN naming a program it does not have."""
+    transport = RecordingTransport()
+    service = MotionService(transport, bas_task=4)
+    service.connect(ConnectionSpec(mode="demo"))
+    service.download_bas(str(SHIPPED_ECAT), 1)
+    _initialise(service)
+
+    # The controller has ECAT.bas; pretend the operator's file was renamed.
+    transport._bas_file = "something-else.bas"  # noqa: SLF001 - the stub's state
+    result = service.reinit_bus()
+
+    assert not result.ok
+    assert "not find" in result.message
+    # And the bus must not be left claiming an initialisation is under way.
+    assert service.bus.init_status != INIT_RUNNING
+
+
+# --------------------------------------------------- D3: the adopted-program path
+
+
+def test_reinit_works_for_a_program_the_session_adopted():
+    """The controller does not report the file name; the UI's remembered path does."""
+
+    class AdoptedProgram(RecordingTransport):
+        def user_var(self, name):
+            return {"BUS_TYPE": 0.0, "Bus_TotalAxisnum": 4.0, "Bus_InitStatus": 1.0}[name]
+
+    transport = AdoptedProgram()
+    service = MotionService(transport, bas_task=4)
+    assert service.connect(
+        ConnectionSpec(mode="demo"), bas_path=r"C:\somewhere\ECAT.bas"
+    ).ok
+
+    # This stub never changes its answer, so let the 25 s settle window lapse
+    # rather than waiting for it: what is under test is the command, not the wait.
+    service._init_timeout = 0.01  # noqa: SLF001 - shortening the vendor window
+    time.sleep(0.02)
+
+    transport.commands.clear()
+    result = service.reinit_bus()
+
+    assert result.ok, result.message
+    assert transport.commands == ["STOPTASK 4", 'RUN "ECAT.bas",4']
+
+
+def test_adopted_program_without_a_known_file_says_so():
+    """The old message claimed no program was loaded while the bus panel said OK."""
+
+    class AdoptedProgram(DemoTransport):
+        def user_var(self, name):
+            return {"BUS_TYPE": 0.0, "Bus_TotalAxisnum": 4.0, "Bus_InitStatus": 1.0}[name]
+
+    service = MotionService(AdoptedProgram())
+    assert service.connect(ConnectionSpec(mode="demo")).ok
+
+    result = service.reinit_bus()
+
+    assert not result.ok
+    assert "文件名" in result.message or "file name" in result.message
+
+
+def test_reinit_needs_the_bus_not_to_be_mid_run():
+    service = MotionService(DemoTransport())
+    service.connect(ConnectionSpec(mode="demo"))
+    service.download_bas(str(SHIPPED_ECAT), 1)
+
+    assert not service.reinit_bus().ok  # still inside the demo's init delay
+
+
+# ------------------------------------------------------- task-number hygiene
+
+
+@pytest.mark.parametrize("value", [0, -1, 9, 100, None, "x", ""])
+def test_a_nonsense_task_number_falls_back_to_the_vendor_default(value):
+    assert clamp_bas_task(value) == DEFAULT_BAS_TASK
+
+
+@pytest.mark.parametrize("value", [1, 2, 4, 8, "4"])
+def test_a_plausible_task_number_survives(value):
+    assert clamp_bas_task(value) == int(value)
+
+
+@pytest.mark.parametrize(
+    "command,expected",
+    [
+        ('RUN "ECAT.bas",4', "ECAT.bas"),
+        ('RUN "a/b c/R TEX.bas",2', "a/b c/R TEX.bas"),
+        ("STOPTASK 4", ""),
+    ],
+)
+def test_run_filename_parsing(command, expected):
+    assert _run_filename(command) == expected
+
+
+# ------------------------------------------------------- the simulated controller
+
+
+def test_the_demo_rejects_a_program_it_does_not_have():
+    """The simulation must misbehave the way the firmware does, or tests lie."""
+    transport = DemoTransport()
+    transport.open(ConnectionSpec(mode="demo"))
+    transport.bas_download(str(SHIPPED_ECAT), 1)
+
+    assert transport.execute('RUN "ECAT.bas",4') == ""
+    assert "not find" in transport.execute('RUN "other.bas",4')
+
+
+def test_the_demo_keeps_the_downloaded_file_name():
+    transport = DemoTransport()
+    transport.open(ConnectionSpec(mode="demo"))
+    transport.bas_download(str(SHIPPED_RTEX), 1)
+
+    assert transport._bas_file == "RTEX.bas"  # noqa: SLF001 - the stub's state
+
+
+def test_a_missing_connection_is_still_reported():
+    service = MotionService(DemoTransport())
+    result = service.reinit_bus()
+    assert not result.ok
+    assert result.message