OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 __init__.py	112 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📁 __pycache__	-	03/31/2022 06:22:39 AM	rwxr-xr-x
📄 _awaittests.py.3only	4.37 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 _posixifaces.py	4.58 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _win32ifaces.py	3.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _yieldfromtests.py.3only	4.17 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 connectionmixins.py	19.71 KB	09/08/2017 10:38:36 AM	rw-r--r--
📁 fake_CAs	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 fakeendpoint.py	1.52 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 modulehelpers.py	1.69 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_cli.py	664 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_connectionlost.py	119 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_gireactornocompat.py	721 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_helper.py	1.06 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 reactormixins.py	13.72 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_abstract.py	1.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_address.py	11.45 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_base.py	10.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_baseprocess.py	2.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_core.py	11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_coroutines.py	1.67 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_default.py	3.46 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_endpoints.py	143.15 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_epollreactor.py	7.16 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_fdset.py	13.18 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_filedescriptor.py	2.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_gireactor.py	8.83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_glibbase.py	2.23 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inlinecb.py	4.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inotify.py	16.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_iocp.py	5.05 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_kqueuereactor.py	1.87 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_main.py	1.4 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_newtls.py	6.54 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_pollingfile.py	1.25 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixbase.py	9.31 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixprocess.py	10.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_process.py	31.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_protocol.py	18.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_resolver.py	19.58 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_serialport.py	1.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_sigchld.py	3.84 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_socket.py	9.11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stdio.py	6.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tcp.py	83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threads.py	7.93 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_time.py	3.63 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tls.py	12.32 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp.py	16.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp_internals.py	4.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_unix.py	33.94 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32events.py	6.35 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32serialport.py	5.16 KB	09/08/2017 10:38:35 AM	rw-r--r--

Editing: test_stdio.py

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

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

from __future__ import absolute_import, division

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.stdio import StandardIO

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

def setUp(self):
        path = self.mktemp()
        open(path, "wb").close()
        self.extraFile = open(path, "rb+")
        self.addCleanup(self.extraFile.close)

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 = b""
            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 == b"hello!":
                    reactor.stop()

path = self.mktemp()

with open(path, "wb") as f:
            f.write(b"hello!")

with open(path, "rb") as f:
            # 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, b"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()

with open(path, "wb") as f:
            # 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(b"hello")
            protocol.transport.write(b", world")
            protocol.transport.loseConnection()

self.runReactor(reactor)

with open(path, "rb") as f:
            self.assertEqual(f.read(), b"hello, world")

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()
        open(path, "wb").close()

with open(path, "rb") as f:
            # 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 = open(self.mktemp(), "wb")

# Have the reader added:
        protocol = Protocol()
        stdio = StandardIO(protocol, stdout=f.fileno(),
                           stdin=self.extraFile.fileno(),
                           reactor=reactor)
        protocol.transport.write(b"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()
        open(path, "wb").close()

# Cleanup might fail if file is GCed too soon:
        self.f = f = open(path, "rb")

# 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()
        open(path, "wb").close()

# Cleanup might fail if file is GCed too soon:
        with open(path, "rb") as f:
            # 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 = open(self.mktemp(), "wb")

# 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())