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_epollreactor.py

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

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

from __future__ import division, absolute_import

from twisted.trial.unittest import TestCase
try:
    from twisted.internet.epollreactor import _ContinuousPolling
except ImportError:
    _ContinuousPolling = None
from twisted.internet.task import Clock
from twisted.internet.error import ConnectionDone

class Descriptor(object):
    """
    Records reads and writes, as if it were a C{FileDescriptor}.
    """

def __init__(self):
        self.events = []

def fileno(self):
        return 1

def doRead(self):
        self.events.append("read")

def doWrite(self):
        self.events.append("write")

def connectionLost(self, reason):
        reason.trap(ConnectionDone)
        self.events.append("lost")

class ContinuousPollingTests(TestCase):
    """
    L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
    objects.
    """

def test_addReader(self):
        """
        Adding a reader when there was previously no reader starts up a
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        self.assertIsNone(poller._loop)
        reader = object()
        self.assertFalse(poller.isReading(reader))
        poller.addReader(reader)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertIs(poller._loop.clock, poller._reactor)
        self.assertTrue(poller.isReading(reader))

def test_addWriter(self):
        """
        Adding a writer when there was previously no writer starts up a
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        self.assertIsNone(poller._loop)
        writer = object()
        self.assertFalse(poller.isWriting(writer))
        poller.addWriter(writer)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertIs(poller._loop.clock, poller._reactor)
        self.assertTrue(poller.isWriting(writer))

def test_removeReader(self):
        """
        Removing a reader stops the C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        poller.addReader(reader)
        poller.removeReader(reader)
        self.assertIsNone(poller._loop)
        self.assertEqual(poller._reactor.getDelayedCalls(), [])
        self.assertFalse(poller.isReading(reader))

def test_removeWriter(self):
        """
        Removing a writer stops the C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        poller.removeWriter(writer)
        self.assertIsNone(poller._loop)
        self.assertEqual(poller._reactor.getDelayedCalls(), [])
        self.assertFalse(poller.isWriting(writer))

def test_removeUnknown(self):
        """
        Removing unknown readers and writers silently does nothing.
        """
        poller = _ContinuousPolling(Clock())
        poller.removeWriter(object())
        poller.removeReader(object())

def test_multipleReadersAndWriters(self):
        """
        Adding multiple readers and writers results in a single
        C{LoopingCall}.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        self.assertIsNotNone(poller._loop)
        poller.addWriter(object())
        self.assertIsNotNone(poller._loop)
        poller.addReader(object())
        self.assertIsNotNone(poller._loop)
        poller.addReader(object())
        poller.removeWriter(writer)
        self.assertIsNotNone(poller._loop)
        self.assertTrue(poller._loop.running)
        self.assertEqual(len(poller._reactor.getDelayedCalls()), 1)

def test_readerPolling(self):
        """
        Adding a reader causes its C{doRead} to be called every 1
        milliseconds.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        poller.addReader(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read"])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read", "read"])
        reactor.advance(0.00001)
        self.assertEqual(desc.events, ["read", "read", "read"])

def test_writerPolling(self):
        """
        Adding a writer causes its C{doWrite} to be called every 1
        milliseconds.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        poller.addWriter(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write"])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write", "write"])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["write", "write", "write"])

def test_connectionLostOnRead(self):
        """
        If a C{doRead} returns a value indicating disconnection,
        C{connectionLost} is called on it.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        desc.doRead = lambda: ConnectionDone()
        poller.addReader(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["lost"])

def test_connectionLostOnWrite(self):
        """
        If a C{doWrite} returns a value indicating disconnection,
        C{connectionLost} is called on it.
        """
        reactor = Clock()
        poller = _ContinuousPolling(reactor)
        desc = Descriptor()
        desc.doWrite = lambda: ConnectionDone()
        poller.addWriter(desc)
        self.assertEqual(desc.events, [])
        reactor.advance(0.001)
        self.assertEqual(desc.events, ["lost"])

def test_removeAll(self):
        """
        L{_ContinuousPolling.removeAll} removes all descriptors and returns
        the readers and writers.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        writer = object()
        both = object()
        poller.addReader(reader)
        poller.addReader(both)
        poller.addWriter(writer)
        poller.addWriter(both)
        removed = poller.removeAll()
        self.assertEqual(poller.getReaders(), [])
        self.assertEqual(poller.getWriters(), [])
        self.assertEqual(len(removed), 3)
        self.assertEqual(set(removed), set([reader, writer, both]))

def test_getReaders(self):
        """
        L{_ContinuousPolling.getReaders} returns a list of the read
        descriptors.
        """
        poller = _ContinuousPolling(Clock())
        reader = object()
        poller.addReader(reader)
        self.assertIn(reader, poller.getReaders())

def test_getWriters(self):
        """
        L{_ContinuousPolling.getWriters} returns a list of the write
        descriptors.
        """
        poller = _ContinuousPolling(Clock())
        writer = object()
        poller.addWriter(writer)
        self.assertIn(writer, poller.getWriters())

if _ContinuousPolling is None:
        skip = "epoll not supported in this environment."