fix(pad-input): preserve command ordering across clock resets

CMakeLists.txt

@@ -90,6 +90,9 @@ if(BUILD_TESTING)
             ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_nid_names.py)
         add_test(NAME release_package COMMAND ${Python3_EXECUTABLE}
             ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_package_release.py)
+        add_test(NAME pad_input COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_pad_input.py)
+        set_tests_properties(pad_input PROPERTIES TIMEOUT 20)
     endif()
 endif()
 

tools/pad_input.py

@@ -1,6 +1,7 @@
 import argparse
 import os
 from pathlib import Path
+import re
 import tempfile
 import time
 
@@ -14,6 +15,36 @@ BUTTONS = {
 }
 
 
+def previous_sequence(path):
+    try:
+        with path.open("rb") as command:
+            line = command.readline(129)
+    except FileNotFoundError:
+        return 0
+    if os.name == "nt" and line.endswith(b"\r\n"):
+        line = line[:-2] + b"\n"
+    tokens = line.split()
+    if len(line.rstrip(b"\n")) > 127 or len(tokens) not in (3, 7):
+        return 0
+    if any(re.fullmatch(rb"\+?[0-9]+", token) is None for token in tokens):
+        return 0
+    values = [int(token) for token in tokens]
+    if any(value > 2**64 - 1 for value in values):
+        return 0
+    if values[0] == 0 or values[1] & ~0x10fffe or not 1 <= values[2] <= 5000:
+        return 0
+    if any(axis > 255 for axis in values[3:]):
+        return 0
+    return values[0]
+
+
+def next_sequence(path):
+    sequence = max(time.monotonic_ns(), previous_sequence(path) + 1)
+    if sequence > 2**64 - 1:
+        raise ValueError("controller command sequence is exhausted")
+    return sequence
+
+
 def main():
     parser = argparse.ArgumentParser(description="Queue a controller pulse for APS5_PAD_INPUT_FILE.")
     parser.add_argument("path", type=Path)
@@ -30,11 +61,15 @@ def main():
         if not all(0 <= axis <= 255 for axis in axes):
             parser.error("stick axes must be between 0 and 255")
         sticks = "".join(f" {axis}" for axis in axes)
+    try:
+        sequence = next_sequence(args.path)
+    except ValueError as error:
+        parser.error(str(error))
     temporary = None
     try:
         with tempfile.NamedTemporaryFile(mode="w", dir=args.path.parent, delete=False) as output:
             temporary = Path(output.name)
-            output.write(f"{time.monotonic_ns()} {BUTTONS[args.button]} {args.hold_ms}{sticks}\n")
+            output.write(f"{sequence} {BUTTONS[args.button]} {args.hold_ms}{sticks}\n")
         os.replace(temporary, args.path)
     finally:
         if temporary is not None:

tools/tests/test_pad_input.py

@@ -0,0 +1,84 @@
+import contextlib
+import io
+import os
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import pad_input
+
+
+class PadInputTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+        self.path = self.root / "command"
+
+    def queue(self, clock, *arguments):
+        with mock.patch.object(sys, "argv", ["pad_input.py", str(self.path), *arguments]), \
+                mock.patch.object(pad_input.time, "monotonic_ns", return_value=clock), \
+                contextlib.redirect_stdout(io.StringIO()):
+            pad_input.main()
+        return [int(token) for token in self.path.read_text(encoding="utf-8").split()]
+
+    def test_persistent_command_is_advanced_after_clock_reset(self):
+        self.path.write_text("2000 16384 300\n", encoding="utf-8")
+        initialized_sequence = 2000
+        command = self.queue(1000, "down")
+        self.assertGreater(command[0], initialized_sequence)
+        self.assertEqual(command[1:], [64, 300])
+        self.assertEqual(self.queue(1001, "cross"), [command[0] + 1, 16384, 300])
+
+    def test_equal_clock_still_produces_new_pulses(self):
+        first = self.queue(1000, "cross")
+        second = self.queue(1000, "down")
+        self.assertGreater(second[0], first[0])
+        self.assertEqual(second[1:], [64, 300])
+
+    def test_clock_newer_than_existing_command_is_retained(self):
+        self.path.write_text("1000 16384 300\n", encoding="utf-8")
+        self.assertEqual(self.queue(2000, "release", "--hold-ms", "1"), [2000, 0, 1])
+
+    def test_fresh_command_preserves_pulse_and_stick_schema(self):
+        self.assertEqual(self.queue(1000, "cross", "--hold-ms", "5000", "--left-stick", "255", "0"),
+                         [1000, 16384, 5000, 255, 0, 128, 128])
+        self.assertEqual(self.queue(2000, "release", "--right-stick", "0", "255"),
+                         [2000, 0, 300, 128, 128, 0, 255])
+
+    def test_valid_stick_command_and_unsigned_tokens_preserve_sequence(self):
+        self.path.write_text("+2000 +16384 +300 +0 +255 +128 +128\n", encoding="utf-8")
+        self.assertEqual(self.queue(1000, "release"), [2001, 0, 300])
+
+    def test_crlf_at_runtime_line_capacity_matches_host_text_mode(self):
+        self.path.write_bytes(b" " * 113 + b"2000 16384 300\r\n")
+        expected = 2001 if os.name == "nt" else 1000
+        self.assertEqual(self.queue(1000, "down"), [expected, 64, 300])
+
+    def test_empty_and_malformed_commands_can_be_replaced(self):
+        for text in ["", "not a command\n", "18446744073709551615 nonsense\n",
+                     "18446744073709551616 16384 300\n", "2000 1 300\n", "2000 16384 0\n",
+                     "2000 16384 5001\n", "2000 16384 300 0 256 128 128\n",
+                     "2000 16384 300 extra\n", "2_000 16384 300\n", "-2000 16384 300\n",
+                     "0 16384 300\n", " " * 128 + "2000 16384 300\n"]:
+            with self.subTest(text=text):
+                self.path.write_text(text, encoding="utf-8")
+                self.assertEqual(self.queue(1000, "down"), [1000, 64, 300])
+
+    def test_exhausted_sequence_reports_error_and_preserves_command(self):
+        original = "18446744073709551615 16384 300\n"
+        self.path.write_text(original, encoding="utf-8")
+        errors = io.StringIO()
+        with contextlib.redirect_stderr(errors), self.assertRaises(SystemExit) as error:
+            self.queue(1000, "down")
+        self.assertEqual(error.exception.code, 2)
+        self.assertIn("sequence is exhausted", errors.getvalue())
+        self.assertEqual(self.path.read_text(encoding="utf-8"), original)
+        self.assertEqual(list(self.root.iterdir()), [self.path])
+
+
+if __name__ == "__main__":
+    unittest.main()