standalone test case (read regardless of EPOLLIN like portage-2.2.17 does)

test_read_pty.py (text/x-python), 1.52 KB, created by Zac Medico on 2015-02-13 04:53:15 UTC

#!/usr/bin/env python
# Attempt to read for all events (regardless of EPOLLIN),
# like portage-2.2.17 does:
# https://github.com/gentoo/portage/commit/b3bba4dc8f4f93adeaeaf662870bb00a09bb1de7
import errno
import fcntl
import os
import pty
import select

BUFSIZE = 4096

def test_read(open_pipe, data):

	consumer_fd, producer_fd = open_pipe()

	pid = os.fork()
	if pid == 0:
		os.close(consumer_fd)
		while data:
			data = data[os.write(producer_fd, data):]
		os._exit(os.EX_OK)

	os.close(producer_fd)
	epoll_obj = select.epoll()
	fcntl.fcntl(consumer_fd, fcntl.F_SETFL,
		fcntl.fcntl(consumer_fd, fcntl.F_GETFL) | os.O_NONBLOCK)

	try:
		epoll_obj.register(consumer_fd,
			select.EPOLLIN | select.EPOLLHUP | select.EPOLLERR)

		received = []
		eof = None

		while eof is None:
			for fd, event in epoll_obj.poll(-1):

				try:
					while eof is None:
						received.append(os.read(fd, BUFSIZE))
						if not received[-1]:
							eof = "os.read returned empty string"
				except OSError as e:
					if e.errno != errno.EAGAIN:
						eof = e
						break

				if event & select.EPOLLHUP:
					eof = "EPOLLHUP"
					break

				if event & select.EPOLLERR:
					eof = "EPOLLERR"
					break

		received = b"".join(received)

		os.waitpid(pid, 0)

		if len(data) != len(received):
			raise AssertionError("%s != %s eof: %s" %
				(len(data), len(received), eof))
	finally:
		epoll_obj.close()
		os.close(consumer_fd)

def main():
	for x in (1, 2, 5, 6, 7, 8, 2**5, 2**10, 2**12, 2**13, 2**14):
		test_read(pty.openpty, x * b"a")

if __name__ == '__main__':
	main()