test_poll: fix array test

Since SpawnProcess no longer uses array, add conditional array support
to PipeReader and use that for tests.

1 files changed, 20 insertions, 62 deletions

diff --git a/pym/portage/tests/process/test_poll.py b/pym/portage/tests/process/test_poll.py
index 9b1f9cb55..f1ddcb3dd 100644
--- a/pym/portage/tests/process/test_poll.py
+++ b/pym/portage/tests/process/test_poll.py
@@ -1,8 +1,6 @@
 # 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
@@ -10,15 +8,10 @@ from _emerge.PollScheduler import PollScheduler
 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
@@ -43,7 +36,7 @@ class PipeReaderTestCase(TestCase):
 
 		consumer = PipeReader(
 			input_files={"producer" : master_file},
-			scheduler=scheduler)
+			scheduler=scheduler, _use_array=self._use_array)
 
 		consumer.start()
 
@@ -58,57 +51,22 @@ class PipeReaderTestCase(TestCase):
 		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