# Copyright 1999-2011 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2
+import array
import errno
import logging
import os
def isAlive(self):
return bool(self._registered)
+ def _read_array(self, f, event):
+ """
+ NOTE: array.fromfile() is used here only for testing purposes,
+ because it has bugs in all known versions of Python (including
+ Python 2.7 and Python 3.2).
+
+ | POLLIN | RETURN
+ | BIT | VALUE
+ | ---------------------------------------------------
+ | 1 | Read self._bufsize into an instance of
+ | | array.array('B') and return it, ignoring
+ | | EOFError and IOError. An empty array
+ | | indicates EOF.
+ | ---------------------------------------------------
+ | 0 | None
+ """
+ buf = None
+ if event & PollConstants.POLLIN:
+ buf = array.array('B')
+ try:
+ buf.fromfile(f, self._bufsize)
+ except EOFError:
+ pass
+ except TypeError:
+ # Python 3.2:
+ # TypeError: read() didn't return bytes
+ pass
+ except IOError as e:
+ # EIO happens with pty on Linux after the
+ # slave end of the pty has been closed.
+ if e.errno == errno.EIO:
+ # EOF: return empty string of bytes
+ pass
+ elif e.errno == errno.EAGAIN:
+ # EAGAIN: return None
+ buf = None
+ else:
+ raise
+
+ if buf is not None:
+ try:
+ # Python >=3.2
+ buf = buf.tobytes()
+ except AttributeError:
+ buf = buf.tostring()
+
+ return buf
+
def _read_buf(self, fd, event):
"""
| POLLIN | RETURN
"""
__slots__ = ("input_files",) + \
- ("_read_data", "_reg_ids")
+ ("_read_data", "_reg_ids", "_use_array")
def _start(self):
self._reg_ids = set()
self._read_data = []
+
+ if self._use_array:
+ output_handler = self._array_output_handler
+ else:
+ output_handler = self._output_handler
+
for k, f in self.input_files.items():
fcntl.fcntl(f.fileno(), fcntl.F_SETFL,
fcntl.fcntl(f.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK)
self._reg_ids.add(self.scheduler.register(f.fileno(),
- self._registered_events, self._output_handler))
+ self._registered_events, output_handler))
self._registered = True
def isAlive(self):
self._unregister_if_appropriate(event)
+ def _array_output_handler(self, fd, event):
+
+ for f in self.input_files.values():
+ if f.fileno() == fd:
+ break
+
+ while True:
+ data = self._read_array(f, event)
+ if data is None:
+ break
+ if data:
+ self._read_data.append(data)
+ else:
+ self._unregister()
+ self.wait()
+ break
+
+ self._unregister_if_appropriate(event)
+
def _unregister(self):
"""
Unregister from the scheduler and close open files.
# Copyright 1998-2011 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2
-import tempfile
-
from portage import os
from portage.tests import TestCase
from portage.util._pty import _create_pty_or_pipe
from _emerge.PipeReader import PipeReader
from _emerge.SpawnProcess import SpawnProcess
-class _SpawnProcessPty(SpawnProcess):
- __slots__ = ("got_pty",)
- def _pipe(self, fd_pipes):
- got_pty, master_fd, slave_fd = _create_pty_or_pipe()
- self.got_pty = got_pty
- return (master_fd, slave_fd)
-
class PipeReaderTestCase(TestCase):
+ _use_array = False
+
def _testPipeReader(self, test_string, use_pty):
"""
Use a poll loop to read data from a pipe and assert that
consumer = PipeReader(
input_files={"producer" : master_file},
- scheduler=scheduler)
+ scheduler=scheduler, _use_array=self._use_array)
consumer.start()
output = consumer.getvalue().decode('ascii', 'replace')
return (output, got_pty)
- def _testPipeReaderArray(self, test_string, use_pty):
- """
- Use a poll loop to read data from a pipe and assert that
- the data written to the pipe is identical to the data
- read from the pipe.
- """
-
- scheduler = PollScheduler().sched_iface
- if use_pty:
- spawn_process = _SpawnProcessPty
- else:
- spawn_process = SpawnProcess
-
- fd, logfile = tempfile.mkstemp()
- os.close(fd)
- producer = spawn_process(
- background=True,
- args=["bash", "-c", "echo -n '%s'" % test_string],
- env=os.environ,
- scheduler=scheduler, logfile=logfile)
-
- try:
- producer.start()
- scheduler.schedule()
- self.assertEqual(producer.returncode, os.EX_OK)
-
- if use_pty:
- got_pty = producer.got_pty
- else:
- got_pty = False
-
- with open(logfile, 'rb') as f:
- output = f.read().decode('ascii')
- return (output, got_pty)
- finally:
- try:
- os.unlink(logfile)
- except OSError:
- pass
-
def testPipeReader(self):
for use_pty in (False, True):
- for use_array in (False, True):
- for x in (1, 2, 5, 6, 7, 8, 2**5, 2**10, 2**12, 2**13, 2**14):
- test_string = x * "a"
- if use_array:
- method = self._testPipeReaderArray
- else:
- method = self._testPipeReader
- output, got_pty = method(test_string, use_pty)
- self.assertEqual(test_string, output,
- "x = %s, len(output) = %s, use_array = %s, "
- "use_pty = %s, got_pty = %s" %
- (x, len(output), use_array, use_pty, got_pty))
+ for x in (1, 2, 5, 6, 7, 8, 2**5, 2**10, 2**12, 2**13, 2**14):
+ test_string = x * "a"
+ output, got_pty = self._testPipeReader(test_string, use_pty)
+ self.assertEqual(test_string, output,
+ "x = %s, len(output) = %s, "
+ "use_pty = %s, got_pty = %s" %
+ (x, len(output), use_pty, got_pty))
+
+class PipeReaderArrayTestCase(PipeReaderTestCase):
+
+ _use_array = True
+
+ def __init__(self, *args, **kwargs):
+ super(PipeReaderArrayTestCase, self).__init__(*args, **kwargs)
+ # http://bugs.python.org/issue5380
+ # https://bugs.pypy.org/issue956
+ self.todo = True
projects / portage.git / commitdiff