# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.internet.stdio}.
"""

from twisted.python.runtime import platform
from twisted.internet.test.reactormixins import ReactorBuilder
from twisted.internet.protocol import Protocol
if not platform.isWindows():
from twisted.internet._posixstdio import StandardIO



class StdioFilesTests(ReactorBuilder):
"""
L{StandardIO} supports reading and writing to filesystem files.
"""

def setUp(self):
path = self.mktemp()
file(path, "w").close()
self.extraFile = file(path, "r+")


def test_addReader(self):
"""
Adding a filesystem file reader to a reactor will make sure it is
polled.
"""
reactor = self.buildReactor()

class DataProtocol(Protocol):
data = ""
def dataReceived(self, data):
self.data += data
# It'd be better to stop reactor on connectionLost, but that
# fails on FreeBSD, probably due to
# http://bugs.python.org/issue9591:
if self.data == "hello!":
reactor.stop()

path = self.mktemp()
f = file(path, "w")
f.write("hello!")
f.close()
f = file(path, "r")

# Read bytes from a file, deliver them to a protocol instance:
protocol = DataProtocol()
StandardIO(protocol, stdin=f.fileno(),
stdout=self.extraFile.fileno(),
reactor=reactor)

self.runReactor(reactor)
self.assertEqual(protocol.data, "hello!")


def test_addWriter(self):
"""
Adding a filesystem file writer to a reactor will make sure it is
polled.
"""
reactor = self.buildReactor()

class DisconnectProtocol(Protocol):
def connectionLost(self, reason):
reactor.stop()

path = self.mktemp()
f = file(path, "w")

# Write bytes to a transport, hopefully have them written to a file:
protocol = DisconnectProtocol()
StandardIO(protocol, stdout=f.fileno(),
stdin=self.extraFile.fileno(), reactor=reactor)
protocol.transport.write("hello")
protocol.transport.write(", world")
protocol.transport.loseConnection()

self.runReactor(reactor)
f.close()
f = file(path, "r")
self.assertEqual(f.read(), "hello, world")
f.close()


def test_removeReader(self):
"""
Removing a filesystem file reader from a reactor will make sure it is
no longer polled.
"""
reactor = self.buildReactor()
self.addCleanup(self.unbuildReactor, reactor)

path = self.mktemp()
file(path, "w").close()
# Cleanup might fail if file is GCed too soon:
self.f = f = file(path, "r")

# Have the reader added:
stdio = StandardIO(Protocol(), stdin=f.fileno(),
stdout=self.extraFile.fileno(),
reactor=reactor)
self.assertIn(stdio._reader, reactor.getReaders())
stdio._reader.stopReading()
self.assertNotIn(stdio._reader, reactor.getReaders())


def test_removeWriter(self):
"""
Removing a filesystem file writer from a reactor will make sure it is
no longer polled.
"""
reactor = self.buildReactor()
self.addCleanup(self.unbuildReactor, reactor)

# Cleanup might fail if file is GCed too soon:
self.f = f = file(self.mktemp(), "w")

# Have the reader added:
protocol = Protocol()
stdio = StandardIO(protocol, stdout=f.fileno(),
stdin=self.extraFile.fileno(),
reactor=reactor)
protocol.transport.write("hello")
self.assertIn(stdio._writer, reactor.getWriters())
stdio._writer.stopWriting()
self.assertNotIn(stdio._writer, reactor.getWriters())


def test_removeAll(self):
"""
Calling C{removeAll} on a reactor includes descriptors that are
filesystem files.
"""
reactor = self.buildReactor()
self.addCleanup(self.unbuildReactor, reactor)

path = self.mktemp()
file(path, "w").close()
# Cleanup might fail if file is GCed too soon:
self.f = f = file(path, "r")

# Have the reader added:
stdio = StandardIO(Protocol(), stdin=f.fileno(),
stdout=self.extraFile.fileno(), reactor=reactor)
# And then removed:
removed = reactor.removeAll()
self.assertIn(stdio._reader, removed)
self.assertNotIn(stdio._reader, reactor.getReaders())


def test_getReaders(self):
"""
C{reactor.getReaders} includes descriptors that are filesystem files.
"""
reactor = self.buildReactor()
self.addCleanup(self.unbuildReactor, reactor)

path = self.mktemp()
file(path, "w").close()
# Cleanup might fail if file is GCed too soon:
self.f = f = file(path, "r")

# Have the reader added:
stdio = StandardIO(Protocol(), stdin=f.fileno(),
stdout=self.extraFile.fileno(), reactor=reactor)
self.assertIn(stdio._reader, reactor.getReaders())


def test_getWriters(self):
"""
C{reactor.getWriters} includes descriptors that are filesystem files.
"""
reactor = self.buildReactor()
self.addCleanup(self.unbuildReactor, reactor)

# Cleanup might fail if file is GCed too soon:
self.f = f = file(self.mktemp(), "w")

# Have the reader added:
stdio = StandardIO(Protocol(), stdout=f.fileno(),
stdin=self.extraFile.fileno(), reactor=reactor)
self.assertNotIn(stdio._writer, reactor.getWriters())
stdio._writer.startWriting()
self.assertIn(stdio._writer, reactor.getWriters())

if platform.isWindows():
skip = ("StandardIO does not accept stdout as an argument to Windows. "
"Testing redirection to a file is therefore harder.")


globals().update(StdioFilesTests.makeTestCaseClasses())