mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-23 09:19:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

gh-132742: Fix newly added tcflush() tests on Android (GH-133070)

Browse source
This commit is contained in:
Serhiy Storchaka 2025-04-28 21:41:09 +03:00 • committed by GitHub
parent 3940e1f873
commit 25186c2472
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 24 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

21
Lib/test/test_ioctl.py
View file

@ -4,7 +4,7 @@ import struct
import sys
import threading
import unittest
from test.support import get_attribute
from test import support
from test.support import threading_helper
from test.support.import_helper import import_module
fcntl = import_module('fcntl')
@ -13,7 +13,7 @@ termios = import_module('termios')
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:
@ -143,7 +143,9 @@ class IoctlTestsPty(unittest.TestCase):
def test_ioctl_clear_input_or_output(self):
wfd = self.slave_fd
rfd = self.master_fd
inbuf = sys.platform == 'linux'
# The data is buffered in the input buffer on Linux, and in
# the output buffer on other platforms.
inbuf = sys.platform in ('linux', 'android')
os.write(wfd, b'abcdef')
self.assertEqual(os.read(rfd, 2), b'ab')
@ -163,7 +165,8 @@ class IoctlTestsPty(unittest.TestCase):
os.write(wfd, b'ABCDEF')
self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
@unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
@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):
wfd = self.slave_fd
@ -173,20 +176,22 @@ class IoctlTestsPty(unittest.TestCase):
def writer():
os.write(wfd, b'abc')
write_suspended.wait()
self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT))
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(support.SHORT_TIMEOUT),
'output was not resumed')
self.assertEqual(os.read(rfd, 1024), b'def')

17
Lib/test/test_termios.py
View file

@ -152,7 +152,9 @@ class TestFunctions(unittest.TestCase):
def test_tcflush_clear_input_or_output(self):
wfd = self.fd
rfd = self.master_fd
inbuf = sys.platform == 'linux'
# The data is buffered in the input buffer on Linux, and in
# the output buffer on other platforms.
inbuf = sys.platform in ('linux', 'android')
os.write(wfd, b'abcdef')
self.assertEqual(os.read(rfd, 2), b'ab')
@ -190,7 +192,8 @@ class TestFunctions(unittest.TestCase):
self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON)
self.assertRaises(TypeError, termios.tcflow, self.fd)
@unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
@support.skip_android_selinux('tcflow')
@unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux')
def test_tcflow_suspend_and_resume_output(self):
wfd = self.fd
rfd = self.master_fd
@ -199,20 +202,22 @@ class TestFunctions(unittest.TestCase):
def writer():
os.write(wfd, b'abc')
write_suspended.wait()
self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT))
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(support.SHORT_TIMEOUT),
'output was not resumed')
self.assertEqual(os.read(rfd, 1024), b'def')