60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""Pipe-level regressions for the native preferences harness response reader."""
|
|
import os
|
|
import select
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from native_preferences import ResponseReader
|
|
|
|
|
|
class ResponseReaderTests(unittest.TestCase):
|
|
def setUp(self):
|
|
readFd, writeFd = os.pipe()
|
|
self.stream = os.fdopen(readFd, encoding="utf-8")
|
|
self.writer = os.fdopen(writeFd, "wb", buffering=0)
|
|
self.addCleanup(self.stream.close)
|
|
self.addCleanup(self.writer.close)
|
|
self.reader = ResponseReader(self.stream)
|
|
|
|
def test_batched_preview_and_save(self):
|
|
self.writer.write(b'{"Preview": true}\n{"Save": true}\n')
|
|
self.assertEqual(self.reader.request(), {"Preview": True})
|
|
self.assertFalse(select.select([self.stream], [], [], 0)[0])
|
|
self.assertEqual(self.reader.request(timeout=0), {"Save": True})
|
|
|
|
def test_missing_response_times_out(self):
|
|
with self.assertRaises(TimeoutError):
|
|
self.reader.request(timeout=0.01)
|
|
|
|
def test_partial_response_times_out_and_can_resume(self):
|
|
self.writer.write(b'{"Name": "\xc3')
|
|
with self.assertRaises(TimeoutError):
|
|
self.reader.request(timeout=0.01)
|
|
self.writer.write(b'\xa9"}\n')
|
|
self.assertEqual(self.reader.request(), {"Name": "\u00e9"})
|
|
|
|
def test_partial_reads_share_one_deadline(self):
|
|
self.writer.write(b'{"Save":')
|
|
with patch("native_preferences.time.monotonic", side_effect=[10, 10, 16]):
|
|
with self.assertRaises(TimeoutError):
|
|
self.reader.request(timeout=5)
|
|
|
|
def test_eof_is_reported_after_complete_responses(self):
|
|
self.writer.write(b'{"Save": true}\n{"Close":')
|
|
self.writer.close()
|
|
self.assertEqual(self.reader.request(), {"Save": True})
|
|
with self.assertRaises(EOFError):
|
|
self.reader.request()
|
|
|
|
def test_remaining_output_includes_read_ahead_and_pipe_bytes(self):
|
|
self.writer.write(b'{"Preview": true}\n{"Name": "\xc3')
|
|
self.assertEqual(self.reader.request(), {"Preview": True})
|
|
self.writer.write(b'\xa9"}\n')
|
|
self.writer.close()
|
|
self.assertEqual(self.reader.read_remaining(), '{"Name": "\u00e9"}\n')
|
|
self.assertEqual(self.reader.read_remaining(), "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|