76 lines
2.0 KiB
Python
76 lines
2.0 KiB
Python
"""Worker-side protocol guards."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from line_laser_modbus.models import ModeCommand, Pose6D
|
|
|
|
from line_laser_hmi.worker import ModbusWorker
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, mode: ModeCommand, *, cache_count: int = 1) -> None:
|
|
self.mode = mode
|
|
self.cache_count = cache_count
|
|
self.target_writes = 0
|
|
self.correction_writes = 0
|
|
|
|
def read_mode(self) -> ModeCommand:
|
|
return self.mode
|
|
|
|
def read_available_cache_count(self) -> int:
|
|
return self.cache_count
|
|
|
|
def write_target_pose(self, pose: Pose6D, *, timestamp: int = 0) -> None:
|
|
self.target_writes += 1
|
|
|
|
def write_correction(self, pose: Pose6D, *, timestamp: int = 0) -> None:
|
|
self.correction_writes += 1
|
|
|
|
|
|
def test_worker_rejects_target_pose_outside_teaching_mode():
|
|
"""Target pose writes are only allowed in pre-weld teaching mode."""
|
|
|
|
worker = ModbusWorker()
|
|
client = _FakeClient(ModeCommand.MANUAL_TEACHING)
|
|
worker._client = client
|
|
|
|
worker.write_target(Pose6D.zeros(), 0)
|
|
|
|
assert client.target_writes == 0
|
|
|
|
|
|
def test_worker_allows_target_pose_in_teaching_mode():
|
|
"""Target pose writes pass when mode 2 has available cache."""
|
|
|
|
worker = ModbusWorker()
|
|
client = _FakeClient(ModeCommand.PRE_WELD_TEACHING)
|
|
worker._client = client
|
|
|
|
worker.write_target(Pose6D.zeros(), 0)
|
|
|
|
assert client.target_writes == 1
|
|
|
|
|
|
def test_worker_rejects_correction_outside_tracking_modes():
|
|
"""Correction writes are only allowed in mode 3 or 4."""
|
|
|
|
worker = ModbusWorker()
|
|
client = _FakeClient(ModeCommand.PRE_WELD_TEACHING)
|
|
worker._client = client
|
|
|
|
worker.write_correction(Pose6D.zeros(), 0)
|
|
|
|
assert client.correction_writes == 0
|
|
|
|
|
|
def test_worker_allows_correction_in_replay_mode():
|
|
"""Trajectory replay is allowed to receive small correction writes."""
|
|
|
|
worker = ModbusWorker()
|
|
client = _FakeClient(ModeCommand.TRAJECTORY_REPLAY)
|
|
worker._client = client
|
|
|
|
worker.write_correction(Pose6D.zeros(), 0)
|
|
|
|
assert client.correction_writes == 1
|