More tweaks to dectalk driver. Think we're nearly there. Hopefully.

tests/unit/test_dectalk_usb_driver.py

@@ -85,6 +85,19 @@ class BlockingUsbDevice(FakeUsbDevice):
         return result
 
 
+class CancelDuringSpeechUsbDevice(FakeUsbDevice):
+    def __init__(self):
+        super().__init__()
+        self.first_normal_write_started = threading.Event()
+        self.release_first_normal_write = threading.Event()
+
+    def write(self, endpoint, data, timeout=None):
+        if endpoint == 0x02 and not self.first_normal_write_started.is_set():
+            self.first_normal_write_started.set()
+            self.release_first_normal_write.wait(timeout=1.0)
+        return super().write(endpoint, data, timeout=timeout)
+
+
 def build_environment():
     return {
         "runtime": {
@@ -197,6 +210,36 @@ def test_dectalk_usb_cancel_bypasses_blocked_normal_write(monkeypatch):
         speech_driver.shutdown()
 
 
+def test_dectalk_usb_cancel_stops_current_speech_upload(monkeypatch):
+    device = CancelDuringSpeechUsbDevice()
+    usb_core = FakeUsbCore(device)
+    usb_util = FakeUsbUtil()
+    speech_driver = dectalkUsbDriver.driver()
+    monkeypatch.setattr(
+        speech_driver,
+        "_load_usb_modules",
+        lambda: (usb_core, usb_util),
+    )
+    speech_driver.initialize(build_environment())
+    normal_thread = threading.Thread(
+        target=speech_driver._write_bytes,
+        args=(b"x" * 192, "speech"),
+    )
+    try:
+        normal_thread.start()
+        assert device.first_normal_write_started.wait(timeout=0.2)
+        speech_driver.cancel()
+    finally:
+        device.release_first_normal_write.set()
+        normal_thread.join(timeout=1.0)
+        speech_driver.shutdown()
+
+    normal_writes = [
+        payload for endpoint, payload, _ in device.writes if endpoint == 0x02
+    ]
+    assert normal_writes == [b"x" * 64]
+
+
 def test_dectalk_usb_driver_reports_missing_pyusb(monkeypatch):
     speech_driver = dectalkUsbDriver.driver()
 

tests/unit/test_hardware_speech_drivers.py

@@ -70,6 +70,18 @@ def test_dectalk_driver_speaks_printable_text(serial_pair):
         speech_driver.shutdown()
 
 
+def test_dectalk_driver_neutralizes_inline_commands_in_speech(
+    serial_pair,
+):
+    speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
+    try:
+        speech_driver.speak("status [:dv g5 10] done")
+        expected = b"status [ :dv g5 10] done\x0b"
+        assert read_available(master_fd, len(expected)) == expected
+    finally:
+        speech_driver.shutdown()
+
+
 def test_dectalk_driver_writes_settings_and_cancel(serial_pair):
     speech_driver, master_fd = initialized_driver(dectalkDriver, serial_pair)
     try: