diff --git a/CMakeLists.txt b/CMakeLists.txt index aa3c450dd..c3e5de8e2 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -92,6 +92,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() diff --git a/tools/pad_input.py b/tools/pad_input.py index e68f3185b..af7238e89 100644 --- a/tools/pad_input.py +++ b/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: diff --git a/tools/tests/test_pad_input.py b/tools/tests/test_pad_input.py new file mode 100644 index 000000000..9771ead3d --- /dev/null +++ b/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()