Apr 28, 2025 · Apr 28, 2025 · Apr 28, 2025 · Apr 28, 2025 · Apr 28, 2025 · Apr 28, 2025
diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py
 import sys
 import threading
 import unittest
 from test.support importget_attribute
 from test importsupport
 from test.support import threading_helper
 from test.support.import_helper import import_module
 fcntl = import_module('fcntl')
 class IoctlTestsTty(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        TIOCGPGRP = get_attribute(termios, 'TIOCGPGRP')
        TIOCGPGRP =support.get_attribute(termios, 'TIOCGPGRP')
        try:
            tty = open("/dev/tty", "rb")
        except OSError:
    def test_ioctl_clear_input_or_output(self):
        wfd = self.slave_fd
        rfd = self.master_fd
        # The data is buffered in input buffer on Linux, and in
        # output buffer on other platforms.
        # The data is buffered intheinput buffer on Linux, and in
        #theoutput buffer on other platforms.
        inbuf = sys.platform in ('linux', 'android')

        os.write(wfd, b'abcdef')
        os.write(wfd, b'ABCDEF')
        self.assertEqual(os.read(rfd, 1024), b'ABCDEF')

    @support.skip_android_selinux('tcflow')
    @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux')
    @unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC')
    def test_ioctl_suspend_and_resume_output(self):

        def writer():
            os.write(wfd, b'abc')
            write_suspended.wait()
 self.assertTrue(write_suspended.wait(5))
            os.write(wfd, b'def')
            write_finished.set()

        with threading_helper.start_threads([threading.Thread(target=writer)]):
            self.assertEqual(os.read(rfd, 3), b'abc')
            try:
                fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
                write_suspended.set()
                try:
                    fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
                finally:
                    write_suspended.set()
                self.assertFalse(write_finished.wait(0.5),
                                 'output was not suspended')
            finally:
                fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON)
            self.assertTrue(write_finished.wait(0.5),
            self.assertTrue(write_finished.wait(5),
                            'output was not resumed')
            self.assertEqual(os.read(rfd, 1024), b'def')

diff --git a/Lib/test/test_termios.py b/Lib/test/test_termios.py
    def test_tcflush_clear_input_or_output(self):
        wfd = self.fd
        rfd = self.master_fd
        # The data is buffered in input buffer on Linux, and in
        # output buffer on other platforms.
        # The data is buffered intheinput buffer on Linux, and in
        #theoutput buffer on other platforms.
        inbuf = sys.platform in ('linux', 'android')

        os.write(wfd, b'abcdef')

        def writer():
            os.write(wfd, b'abc')
            write_suspended.wait()
 self.assertTrue(write_suspended.wait(5))
            os.write(wfd, b'def')
            write_finished.set()

        with threading_helper.start_threads([threading.Thread(target=writer)]):
            self.assertEqual(os.read(rfd, 3), b'abc')
            try:
                termios.tcflow(wfd, termios.TCOOFF)
                write_suspended.set()
                try:
                    termios.tcflow(wfd, termios.TCOOFF)
                finally:
                    write_suspended.set()
                self.assertFalse(write_finished.wait(0.5),
                                 'output was not suspended')
            finally:
                termios.tcflow(wfd, termios.TCOON)
            self.assertTrue(write_finished.wait(0.5),
            self.assertTrue(write_finished.wait(5),
                            'output was not resumed')
            self.assertEqual(os.read(rfd, 1024), b'def')
Original file line number	Diff line number	Diff line change
Expand Up		@@ -4,7 +4,7 @@
		import sys
		import threading
		import unittest
		from test.support importget_attribute
		from test importsupport
		from test.support import threading_helper
		from test.support.import_helper import import_module
		fcntl = import_module('fcntl')
Expand All		@@ -13,7 +13,7 @@
		class IoctlTestsTty(unittest.TestCase):
		@classmethod
		def setUpClass(cls):
		TIOCGPGRP = get_attribute(termios, 'TIOCGPGRP')
		TIOCGPGRP =support.get_attribute(termios, 'TIOCGPGRP')
		try:
		tty = open("/dev/tty", "rb")
		except OSError:
Expand DownExpand Up		@@ -143,8 +143,8 @@ def setUp(self):
		def test_ioctl_clear_input_or_output(self):
		wfd = self.slave_fd
		rfd = self.master_fd
		# The data is buffered in input buffer on Linux, and in
		# output buffer on other platforms.
		# The data is buffered intheinput buffer on Linux, and in
		#theoutput buffer on other platforms.
		inbuf = sys.platform in ('linux', 'android')

		os.write(wfd, b'abcdef')
Expand All		@@ -165,6 +165,7 @@ def test_ioctl_clear_input_or_output(self):
		os.write(wfd, b'ABCDEF')
		self.assertEqual(os.read(rfd, 1024), b'ABCDEF')

		@support.skip_android_selinux('tcflow')
		@unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux')
		@unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC')
		def test_ioctl_suspend_and_resume_output(self):
Expand All		@@ -175,20 +176,22 @@ def test_ioctl_suspend_and_resume_output(self):

		def writer():
		os.write(wfd, b'abc')
		write_suspended.wait()
		self.assertTrue(write_suspended.wait(5))
Copy link Member vstinnerApr 28, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. I would prefer to use support.SHORT_TIMEOUT rather than hardcoded timeout. Same remark for other 5 seconds timeouts.
		os.write(wfd, b'def')
		write_finished.set()

		with threading_helper.start_threads([threading.Thread(target=writer)]):
		self.assertEqual(os.read(rfd, 3), b'abc')
		try:
		fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
		write_suspended.set()
		try:
		fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
		finally:
		write_suspended.set()
		self.assertFalse(write_finished.wait(0.5),
		'output was not suspended')
		finally:
		fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON)
		self.assertTrue(write_finished.wait(0.5),
		self.assertTrue(write_finished.wait(5),
		'output was not resumed')
		self.assertEqual(os.read(rfd, 1024), b'def')

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -152,8 +152,8 @@ def test_tcflush_errors(self):
		def test_tcflush_clear_input_or_output(self):
		wfd = self.fd
		rfd = self.master_fd
		# The data is buffered in input buffer on Linux, and in
		# output buffer on other platforms.
		# The data is buffered intheinput buffer on Linux, and in
		#theoutput buffer on other platforms.
		inbuf = sys.platform in ('linux', 'android')

		os.write(wfd, b'abcdef')
Expand DownExpand Up		@@ -202,20 +202,22 @@ def test_tcflow_suspend_and_resume_output(self):

		def writer():
		os.write(wfd, b'abc')
		write_suspended.wait()
		self.assertTrue(write_suspended.wait(5))
		os.write(wfd, b'def')
		write_finished.set()

		with threading_helper.start_threads([threading.Thread(target=writer)]):
		self.assertEqual(os.read(rfd, 3), b'abc')
		try:
		termios.tcflow(wfd, termios.TCOOFF)
		write_suspended.set()
		try:
		termios.tcflow(wfd, termios.TCOOFF)
		finally:
		write_suspended.set()
		self.assertFalse(write_finished.wait(0.5),
		'output was not suspended')
		finally:
		termios.tcflow(wfd, termios.TCOON)
		self.assertTrue(write_finished.wait(0.5),
		self.assertTrue(write_finished.wait(5),
		'output was not resumed')
		self.assertEqual(os.read(rfd, 1024), b'def')

Expand Down