mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-24 11:44:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

[3.13] gh-132742: Fix newly added tcflush() tests on Android (GH-133070) (GH-133104)

Browse source 
(cherry picked from commit 25186c2472)

Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>
This commit is contained in:
Miss Islington (bot) 2025-04-28 21:10:42 +02:00 • committed by GitHub
parent ab89f6ec84
commit b71442f44f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 20 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

18
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')
@ -173,20 +175,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')

14
Lib/test/test_termios.py
View file

@ -142,7 +142,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')
@ -187,20 +189,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')