Index: twisted/internet/_pollingfile.py =================================================================== --- twisted/internet/_pollingfile.py (revision 27422) +++ twisted/internet/_pollingfile.py (working copy) @@ -3,9 +3,9 @@ Implements a simple polling interface for file descriptors that don't work with select() - this is pretty much only useful on Windows. - """ +import sys from zope.interface import implements from twisted.internet.interfaces import IConsumer, IPushProducer @@ -13,7 +13,8 @@ MIN_TIMEOUT = 0.000000001 MAX_TIMEOUT = 0.1 -class _PollableResource: + +class _PollableResource(object): active = True def activate(self): @@ -22,7 +23,8 @@ def deactivate(self): self.active = False -class _PollingTimer: + +class _PollingTimer(object): # Everything is private here because it is really an implementation detail. def __init__(self, reactor): @@ -91,51 +93,31 @@ # If we ever (let's hope not) need the above functionality on UNIX, this could # be factored into a different module. -import win32pipe import win32file import win32api import pywintypes -class _PollableReadPipe(_PollableResource): +class _PollableReader(_PollableResource): + implements(IPushProducer) - def __init__(self, pipe, receivedCallback, lostCallback): - # security attributes for pipes - self.pipe = pipe + def __init__(self, handle, receivedCallback, lostCallback): + self.handle = handle self.receivedCallback = receivedCallback self.lostCallback = lostCallback def checkWork(self): - finished = 0 - fullDataRead = [] + raise NotImplementedError() - while 1: - try: - buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.pipe, 1) - # finished = (result == -1) - if not bytesToRead: - break - hr, data = win32file.ReadFile(self.pipe, bytesToRead, None) - fullDataRead.append(data) - except win32api.error: - finished = 1 - break - - dataBuf = ''.join(fullDataRead) - if dataBuf: - self.receivedCallback(dataBuf) - if finished: - self.cleanup() - return len(dataBuf) - def cleanup(self): self.deactivate() self.lostCallback() def close(self): + # XXX why not cleanup? try: - win32api.CloseHandle(self.pipe) + win32api.CloseHandle(self.handle) except pywintypes.error: # You can't close std handles...? pass @@ -150,28 +132,20 @@ self.activate() -FULL_BUFFER_SIZE = 64 * 1024 -class _PollableWritePipe(_PollableResource): +class _PollableWriter(_PollableResource): + FULL_BUFFER_SIZE = 64 * 1024 implements(IConsumer) - - def __init__(self, writePipe, lostCallback): + + def __init__(self, handle, lostCallback): self.disconnecting = False self.producer = None self.producerPaused = 0 self.streamingProducer = 0 self.outQueue = [] - self.writePipe = writePipe + self.handle = handle self.lostCallback = lostCallback - try: - win32pipe.SetNamedPipeHandleState(writePipe, - win32pipe.PIPE_NOWAIT, - None, - None) - except pywintypes.error: - # Maybe it's an invalid handle. Who knows. - pass def close(self): self.disconnecting = True @@ -219,7 +193,7 @@ def writeConnectionLost(self): self.deactivate() try: - win32api.CloseHandle(self.writePipe) + win32api.CloseHandle(self.handle) except pywintypes.error: # OMG what pass @@ -232,17 +206,70 @@ if self.disconnecting: return self.outQueue.append(data) - if sum(map(len, self.outQueue)) > FULL_BUFFER_SIZE: + if sum(map(len, self.outQueue)) > self.FULL_BUFFER_SIZE: self.bufferFull() def checkWork(self): + raise NotImplementedError() + + + +# +# Pipe support +# +import win32pipe + + +class _PollableReadPipe(_PollableReader): + def __init__(self, pipe, receivedCallback, lostCallback): + _PollableReader.__init__(self, pipe, receivedCallback, lostCallback) + # security attributes for pipes + + def checkWork(self): + finished = 0 + fullDataRead = [] + + while 1: + try: + buffer, bytesToRead, result = win32pipe.PeekNamedPipe(self.handle, 1) + # finished = (result == -1) + if not bytesToRead: + break + hr, data = win32file.ReadFile(self.handle, bytesToRead, None) + fullDataRead.append(data) + except win32api.error: + finished = 1 + break + + dataBuf = ''.join(fullDataRead) + if dataBuf: + self.receivedCallback(dataBuf) + if finished: + self.cleanup() + return len(dataBuf) + + +class _PollableWritePipe(_PollableWriter): + def __init__(self, writePipe, lostCallback): + _PollableWriter.__init__(self, writePipe, lostCallback) + + try: + win32pipe.SetNamedPipeHandleState(writePipe, + win32pipe.PIPE_NOWAIT, + None, + None) + except pywintypes.error: + # Maybe it's an invalid handle. Who knows. + pass + + def checkWork(self): numBytesWritten = 0 if not self.outQueue: if self.disconnecting: self.writeConnectionLost() return 0 try: - win32file.WriteFile(self.writePipe, '', None) + win32file.WriteFile(self.handle, '', None) except pywintypes.error: self.writeConnectionLost() return numBytesWritten @@ -250,7 +277,7 @@ data = self.outQueue.pop(0) errCode = 0 try: - errCode, nBytesWritten = win32file.WriteFile(self.writePipe, + errCode, nBytesWritten = win32file.WriteFile(self.handle, data, None) except win32api.error: self.writeConnectionLost() @@ -266,5 +293,3 @@ if not resumed and self.disconnecting: self.writeConnectionLost() return numBytesWritten - - Index: twisted/internet/_win32stdio.py =================================================================== --- twisted/internet/_win32stdio.py (revision 27422) +++ twisted/internet/_win32stdio.py (working copy) @@ -1,18 +1,149 @@ -# -*- test-case-name: twisted.test.test_process.ProcessTestCase.testStdio -*- +# -*- test-case-name: twisted.test.test_process.ProcessTestCase.testStdio,twisted.test.test_conio.StdIOTestCase -*- +import os +import errno +import msvcrt + +import pywintypes import win32api -import os, msvcrt +import win32console from zope.interface import implements from twisted.internet.interfaces import IHalfCloseableProtocol, ITransport, IAddress from twisted.internet.interfaces import IConsumer, IPushProducer +from twisted.internet import main, abstract -from twisted.internet import _pollingfile, main +import _pollingfile + + +# _pollingfile support +# XXX check me +class _PollableReadConsole(_pollingfile._PollableReader): + def __init__(self, channel, receivedCallback, lostCallback): + _pollingfile._PollableReader.__init__(self, channel.handle, + receivedCallback, lostCallback) + self.channel = channel + + def checkWork(self): + try: + data = self.channel.read(2000) # 2000 = 80 x 25 + except IOError, ioe: + assert ioe.args[0] == errno.EAGAIN + return 0 + except win32console.error: + # stdin or stdout closed? + self.cleanup() + return 0 + + self.receivedCallback(data) + return len(data) + + +class _PollableWriteConsole(_pollingfile._PollableWriter): + def __init__(self, channel, lostCallback): + _pollingfile._PollableWriter.__init__(self, channel.handle, lostCallback) + + self.channel = channel + + def checkWork(self): + numBytesWritten = 0 + if not self.outQueue: + if self.disconnecting: + self.writeConnectionLost() + return 0 + try: + self.channel.write('') + except pywintypes.error: + self.writeConnectionLost() + return numBytesWritten + + while self.outQueue: + data = self.outQueue.pop(0) + try: + # XXX as far as I know, + # nBytesWritten is always equal to len(data) + nBytesWritten = self.channel.write(data) + except win32console.error: + self.writeConnectionLost() + break + else: + numBytesWritten += nBytesWritten + if len(data) > nBytesWritten: + self.outQueue.insert(0, data[nBytesWritten:]) + break + else: + resumed = self.bufferEmpty() + if not resumed and self.disconnecting: + self.writeConnectionLost() + + return numBytesWritten + + + +# support for win32eventreactor +# XXX check me +class ConsoleReader(abstract.FileDescriptor): + def __init__(self, channel, receivedCallback, lostCallback, reactor=None): + """win32eventreactor is assumed. + """ + + if not reactor: + from twisted.internet import reactor + + self.channel = channel + self.receivedCallback = receivedCallback + self.lostCallback = lostCallback + self.reactor = reactor + + self.reactor.addEvent(self.channel.handle, self, "doRead") + + + def doRead(self): + try: + data = self.channel.read(2000) # 2000 = 80 x 25 + except IOError, ioe: + assert ioe.args[0] == errno.EAGAIN + return 0 + except win32console.error: + # stdin or stdout closed? + self.cleanup() + return 0 + + self.receivedCallback(data) + return len(data) + + def cleanup(self): + self.reactor.removeEvent(self.channel.handle) + self.lostCallback() + + def close(self): + try: + win32api.CloseHandle(self.channel.handle) + except pywintypes.error: + # You can't close std handles...? + pass + + self.cleanup() + + def stopProducing(self): + self.close() + + def pauseProducing(self): + self.reactor.removeEvent(self.channel.handle) + + def resumeProducing(self): + self.reactor.addEvent(self.channel.handle, self, "doRead") + + class Win32PipeAddress(object): implements(IAddress) +class Win32ConsoleAddress(object): + implements(IAddress) + + class StandardIO(_pollingfile._PollingTimer): implements(ITransport, @@ -28,24 +159,54 @@ Also, put it stdin/stdout/stderr into binary mode. """ + from twisted.internet import reactor - + from twisted.internet.win32eventreactor import Win32Reactor + for stdfd in range(0, 1, 2): msvcrt.setmode(stdfd, os.O_BINARY) _pollingfile._PollingTimer.__init__(self, reactor) self.proto = proto + + # Check for win32eventreactor + win32Enabled = reactor.__class__ is Win32Reactor - hstdin = win32api.GetStdHandle(win32api.STD_INPUT_HANDLE) - hstdout = win32api.GetStdHandle(win32api.STD_OUTPUT_HANDLE) + # Check if we are connected to a console. + # If this is the case, connect to the console, else connect to + # anonymous pipes. + if os.isatty(0): + import conio + + if win32Enabled: + self.stdin = ConsoleReader( + conio.stdin, self.dataReceived, self.readConnectionLost + ) + else: + self.stdin = _PollableReadConsole( + conio.stdin, self.dataReceived, self.readConnectionLost + ) + else: + hstdin = win32api.GetStdHandle(win32api.STD_INPUT_HANDLE) + self.stdin = _pollingfile._PollableReadPipe( + hstdin, self.dataReceived, self.readConnectionLost + ) - self.stdin = _pollingfile._PollableReadPipe( - hstdin, self.dataReceived, self.readConnectionLost) + if os.isatty(1): + import conio + + self.stdout = _PollableWriteConsole( + conio.stdout, self.writeConnectionLost + ) + else: + hstdout = win32api.GetStdHandle(win32api.STD_OUTPUT_HANDLE) + self.stdout = _pollingfile._PollableWritePipe( + hstdout, self.writeConnectionLost + ) + + if not (os.isatty(0) and win32Enabled): + self._addPollableResource(self.stdin) - self.stdout = _pollingfile._PollableWritePipe( - hstdout, self.writeConnectionLost) - - self._addPollableResource(self.stdin) self._addPollableResource(self.stdout) self.proto.makeConnection(self) @@ -86,11 +247,18 @@ self.stdout.close() def getPeer(self): - return Win32PipeAddress() + if os.isatty(0) and os.isatty(1): + return Win32ConsoleAddress() + else: + return Win32PipeAddress() def getHost(self): - return Win32PipeAddress() + if os.isatty(0) and os.isatty(1): + return Win32ConsoleAddress() + else: + return Win32PipeAddress() + # IConsumer def registerProducer(self, producer, streaming): @@ -113,4 +281,3 @@ def resumeProducing(self): self.stdin.resumeProducing() - Index: twisted/internet/conio.py =================================================================== --- twisted/internet/conio.py (revision 0) +++ twisted/internet/conio.py (revision 0) @@ -0,0 +1,328 @@ +# -*- test-case-name: twisted.test.test_conio -*- +"""This module implements POSIX replacements for stdin/stdout/stderr +that support asyncronous read/write to a Windows console. + +Some details about Windows console: +- a process can have attached only one console +- there can be only one input buffer +- there can be more then one output buffer + +Moreover this module tries to offer an higher level and convenient +interface for termios commands. +""" + +import errno +import pywintypes +import win32api +import win32console + + + +_ENABLE_NORMAL_MODE = win32console.ENABLE_ECHO_INPUT | win32console.ENABLE_LINE_INPUT +_ENABLE_WINDOW_INPUT = win32console.ENABLE_WINDOW_INPUT + + +class ConIn(object): + """I implement a file like object that supports asyncronous reading + from a console. + + This class should be considered a singleton, don't instantiate new + objects and instead use the global stdin object. + """ + + def __init__(self, handle): + # handle should be std handle for STD_INPUT_HANDLE + self.handle = handle + + # The code page in use + # I assume that this does not change + self.cp = "cp%d" % win32console.GetConsoleCP() + + # The temporary (accumulation) buffer used to store the data as + # it arrives from the console + self._buf = [] + + # The buffer used to store data ready to be read + self.buffer = '' + + # Enable the receiving of input records when the console + # window (buffer) is changed + defaultMode = handle.GetConsoleMode() + handle.SetConsoleMode(defaultMode | _ENABLE_WINDOW_INPUT) + + # The callback to be called upon the receiving of a windows + # change record + self._windowChangeCallback = None + + # To optimize the code we use different functions for normal + # and raw mode + self.read = self._read + self.readline = self._readline + + # + # termios interface + # + def enableRawMode(self, enabled=True): + """Enable raw mode. + + XXX check me + """ + + # Flush buffer + self._buf = [] + self.buffer = '' + + # Flush the console buffer, too + self.handle.FlushConsoleInputBuffer() + + mode = self.handle.GetConsoleMode() + + if enabled: + self.read = self._read_raw + self.readline = self._readline_raw + + # Set mode on the console, too + # XXX check me (this seems not to work) + self.handle.SetConsoleMode(mode & ~_ENABLE_NORMAL_MODE) + else: + self.read = self._read + self.readline = self._readline + + # Set mode on the console, too + self.handle.SetConsoleMode(mode | _ENABLE_NORMAL_MODE) + + def setWindowChangeCallback(self, callback): + """callback is called when the console window buffer is + changed. + + Note: WINDOW_BUFFER_SIZE_EVENT is only raised when changing + the window *buffer* size from the console menu + """ + + self._windowChangeCallback = callback + + + # + # File object interface + # + def close(self): + win32api.CloseHandle(self.handle) + + def flush(self): + # Flush both internal buffer and system console buffer + self.buffer = '' + self._buf = [] + + self.handle.FlushConsoleInputBuffer() + + def fileno(self): + return self.handle + + def isatty(self): + return True + + def next(self): + raise NotImplementedError("Not yet implemented") + + def _read(self, size=None): + """Read size bytes from the console. + An exception is raised when the operation would block. + + XXX Just return the empty string instead of raising an exception? + """ + + # This can fail if stdout has been closed + info = stdout.handle.GetConsoleScreenBufferInfo() + rowSize = info["MaximumWindowSize"].X + + # Initialize the current cursor position + if not self._buf: + self.pos = info["CursorPosition"] + + while 1: + n = self.handle.GetNumberOfConsoleInputEvents() + if n == 0: + break + + records = self.handle.ReadConsoleInput(n) + + # Process input + for record in records: + if record.EventType == win32console.WINDOW_BUFFER_SIZE_EVENT: + rowSize = record.Size.X + if self._windowChangeCallback: + self._windowChangeCallback() + if record.EventType != win32console.KEY_EVENT \ + or not record.KeyDown: + continue + + char = record.Char + n = record.RepeatCount + if char == '\b': + pos = stdout.handle.GetConsoleScreenBufferInfo()["CursorPosition"] + + # Move the cursor + x = pos.X - n + if x >= 0: + pos.X = x + # XXX assuming |x| < rowSize (I'm lazy) + elif pos.Y > self.pos.Y: + pos.X = rowSize - 1 + pos.Y -= 1 + + stdout.handle.SetConsoleCursorPosition(pos) + stdout.handle.WriteConsoleOutputCharacter(' ' * n, pos) + + # Delete the characters from accumulation buffer + self._buf = self._buf[:-n] + continue + elif char == '\0': + vCode = record.VirtualKeyCode + # XXX TODO handle keyboard navigation + continue + elif char == '\r': + char = '\n' * n + + self._buf.append(char) + stdout.handle.WriteConsole(char) # do echo + + # We have some data ready to be read + self.buffer = ''.join(self._buf) + self._buf = [] + self.pos = info["CursorPosition"] + + if size is None: + size = len(self.buffer) + + data = self.buffer[:size] + self.buffer = self.buffer[size:] + return data + + char = char * n + data = char.encode(self.cp) + stdout.handle.WriteConsole(data) # do echo + + self._buf.append(data) + + if self.buffer: + data = self.buffer[:size] + self.buffer = self.buffer[size:] + return data + else: + raise IOError(errno.EAGAIN) + + def _read_raw(self, size=None): + """Read size bytes from the console, in raw mode. + + XXX check me. + """ + + while 1: # XXX is this loop really needed? + n = self.handle.GetNumberOfConsoleInputEvents() + if n == 0: + break + + records = self.handle.ReadConsoleInput(n) + + # Process input + for record in records: + if record.EventType == win32console.WINDOW_BUFFER_SIZE_EVENT: + if self._windowChangeCallback: + self._windowChangeCallback() + if record.EventType != win32console.KEY_EVENT \ + or not record.KeyDown: + continue + + char = record.Char + n = record.RepeatCount + if char == '\0': + vCode = record.VirtualKeyCode + # XXX TODO handle keyboard navigation + continue + elif char == '\r': + char = '\n' * n + + char = char * n + data = char.encode(self.cp) + + self._buf.append(data) + + + buffer = ''.join(self._buf) + if buffer: + if size is None: + size = len(buffer) + + data = buffer[:size] + # Keep the remaining data in the accumulation buffer + self._buf = [buffer[size:]] + return data + else: + return '' + + def _readline(self, size=None): + # XXX check me + return self._read(size) + + def _readline_raw(self, size=None): + raise NotImplementedError("Not yet implemented") + + + +class ConOut(object): + """I implement a file like object that supports asyncronous writing + to a console. + + This class should be considered private, don't instantiate new + objects and instead use the global stdout and stderr objects. + + Note that there is no option to make WriteConsole non blocking, + but is seems that this function does not block at all. + When a blocking operation like text selection is in action, the + process is halted. + """ + + def __init__(self, handle): + # handle should be std handle for STD_OUTOUT_HANDLE or STD_ERROR_HANDLE + self.handle = handle + + + # + # File object interface + # + def close(self): + win32api.CloseHandle(self.handle) + + def flush(self): + # There is no buffering + pass + + def fileno(self): + return self.handle + + def isatty(self): + return True + + def write(self, s): + """Write a string to the console. + """ + + return self.handle.WriteConsole(s) + + def writelines(self, seq): + """Write a sequence of strings to the console. + """ + + s = ''.join(seq) + return self.handle.WriteConsole(s) + + + +# The public interface of this module +# XXX TODO replace sys.stdin, sys.stdout and sys.stderr? +stdin = ConIn(win32console.GetStdHandle(win32console.STD_INPUT_HANDLE)) +stdout = ConOut(win32console.GetStdHandle(win32console.STD_OUTPUT_HANDLE)) +stderr = ConOut(win32console.GetStdHandle(win32console.STD_ERROR_HANDLE)) + + +__all__ = [stdin, stdout, stderr] Index: twisted/test/test_conio.py =================================================================== --- twisted/test/test_conio.py (revision 0) +++ twisted/test/test_conio.py (revision 0) @@ -0,0 +1,299 @@ +"""Test suite for asyncronous I/O support for Windows Console. + +For testing I use the low level WriteConsoleInput function that allows +to write directly in the console input queue. +""" + +import os, sys +import win32console + +from twisted.trial import unittest +from twisted.python import filepath +from twisted.internet import error, defer, protocol, reactor + +from twisted.internet import conio, _win32stdio as stdio + + + +def createKeyEvent(char, repeat=1): + """Create a low level record structure with the given character. + """ + + evt = win32console.PyINPUT_RECORDType(win32console.KEY_EVENT) + evt.KeyDown = True + evt.Char = char + evt.RepeatCount = repeat + + return evt + + +stdin = win32console.GetStdHandle(win32console.STD_INPUT_HANDLE) + + +class ConInTestCase(unittest.TestCase): + """Test case for console stdin. + """ + + def tearDown(self): + conio.stdin.flush() + + def testRead(self): + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hello\n") + + def testRead2(self): + """Test two consecutives read. + """ + + def read(): + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hello\n") + + read() + read() + + def testReadMultiple(self): + """Test if repeated characters are handled correctly. + """ + + data = u"hello\r" + records = [createKeyEvent(c, 3) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hhheeellllllooo\n\n\n") + + def testReadWithDelete(self): + """Test if deletion is handled correctly. + """ + + data = u"hello" + u"\b" * 5 + u"world\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "world\n") + + def testDeleteBoundary(self): + """Test if deletion is handled correctly. + """ + + data = u"h" + "\b\b" + u"w\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "w\n") + + def testDeleteFullBoundary(self): + """Test if deletion is handled correctly. + """ + + data = u"h" * 500 + "\b" * 600 + u"w\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "w\n") + + def testReadWithBuffer(self): + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read(3) + self.failUnlessEqual(result, "hel") + + result = conio.stdin.read(3) + self.failUnlessEqual(result, "lo\n") + + def testReadWouldBlock(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + self.failUnlessRaises(IOError, conio.stdin.read) + + def testReadWouldBlockBuffer(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + self.failUnlessRaises(IOError, conio.stdin.read, 3) + + def testIsatty(self): + self.failUnless(conio.stdin.isatty()) + + def testBuffer(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + try: + # This will put the data in the accumulation buffer + conio.stdin.read() + except IOError: + pass + + self.failUnlessEqual(conio.stdin._buf, list("hello")) + + def testFlush(self): + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read(3) + conio.stdin.flush() + + self.failIf(conio.stdin.buffer) + self.failUnlessRaises(IOError, conio.stdin.read, 3) + + def testFlushBuffer(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + try: + # This will put the data in the accumulation buffer + conio.stdin.read() + except IOError: + pass + + conio.stdin.flush() + + self.failIf(conio.stdin.buffer) + self.failIf(conio.stdin._buf) + self.failUnlessRaises(IOError, conio.stdin.read, 3) + + +class ConInRawTestCase(unittest.TestCase): + """Test case for console stdin in raw mode. + """ + + def setUp(self): + conio.stdin.enableRawMode() + + def tearDown(self): + conio.stdin.flush() + conio.stdin.enableRawMode(False) + + def testRead(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hello") + + + def testReadMultiple(self): + data = u"hello" + records = [createKeyEvent(c, 3) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hhheeellllllooo") + + + def testReadWithDelete(self): + data = u"hello" + u'\b' * 5 + u"world" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read() + self.failUnlessEqual(result, "hello" + '\b' * 5 + "world") + + def testReadWithBuffer(self): + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read(3) + self.failUnlessEqual(result, "hel") + + result = conio.stdin.read(3) + self.failUnlessEqual(result, "lo\n") + + def testFlush(self): + data = u"hello" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + result = conio.stdin.read(3) + conio.stdin.flush() + + self.failIf(conio.stdin.buffer) + self.failIf(conio.stdin.read()) + + +class ConOutTestCase(unittest.TestCase): + """Test case for console stdout. + Not very much to test, yet. + """ + + def testWrite(self): + data = "hello" + n = conio.stdout.write(data) + + self.failUnlessEqual(n, 5) + + def testWriteUnicode(self): + data = u"hello" + n = conio.stdout.write(data) + + self.failUnlessEqual(n, 5) + + def testWritelines(self): + data = ["hello", "world"] + n = conio.stdout.writelines(data) + + self.failUnlessEqual(n, 10) + + def testIsatty(self): + self.failUnless(conio.stdout.isatty()) + + + +class StdIOTestProtocol(protocol.Protocol): + def __init__(self): + self.onData = defer.Deferred() + + def dataReceived(self, data): + self.onData.callback(data) + + +class StdIOTestCase(unittest.TestCase): + """Test twisted.internet.stdio support for consoles. + """ + + def setUp(self): + p = StdIOTestProtocol() + self.stdio = stdio.StandardIO(p) + self.onData = p.onData + + def tearDown(self): + self.stdio._pause() + try: + self.stdio._stopPolling() + except error.AlreadyCalled: + pass + + conio.stdin.flush() + + def testRead(self): + def cb(result): + self.failUnlessEqual(result, "hello\n") + + data = u"hello\r" + records = [createKeyEvent(c) for c in data] + stdin.WriteConsoleInput(records) + + return self.onData.addCallback(cb)