OXIESEC PANEL
- Current Dir:
/
/
usr
/
lib
/
python3
/
dist-packages
/
twisted
/
internet
/
test
Server IP: 139.59.38.164
Upload:
Create Dir:
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_time.py
Close
# Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for implementations of L{IReactorTime}. """ __metaclass__ = type from twisted.python.log import msg from twisted.python.runtime import platform from twisted.trial.unittest import SkipTest from twisted.internet.test.reactormixins import ReactorBuilder from twisted.internet.interfaces import IReactorTime, IReactorThreads class TimeTestsBuilder(ReactorBuilder): """ Builder for defining tests relating to L{IReactorTime}. """ requiredInterfaces = (IReactorTime,) def test_delayedCallStopsReactor(self): """ The reactor can be stopped by a delayed call. """ reactor = self.buildReactor() reactor.callLater(0, reactor.stop) reactor.run() def test_distantDelayedCall(self): """ Scheduling a delayed call at a point in the extreme future does not prevent normal reactor operation. """ reactor = self.buildReactor() if IReactorThreads.providedBy(reactor): def eventSource(reactor, event): msg(format="Thread-based event-source scheduling %(event)r", event=event) reactor.callFromThread(event) else: raise SkipTest("Do not know how to synthesize non-time event to " "stop the test") # Pick a pretty big delay. delayedCall = reactor.callLater(2 ** 128 + 1, lambda: None) def stop(): msg("Stopping the reactor") reactor.stop() # Use repeated invocation of the event source to set up the call to stop # the reactor. This makes it more likely at least one normal iteration # will take place with the delayed call in place before the slightly # different reactor shutdown logic alters things. eventSource(reactor, lambda: eventSource(reactor, stop)) # Run the reactor directly, without a timeout. A timeout would # interfere with the purpose of this test, which is to have the timeout # passed to the reactor's doIterate implementation (potentially) be # very, very large. Hopefully the event source defined above will work # and cause the reactor to stop. reactor.run() # The reactor almost surely stopped before the delayed call # fired... right? self.assertTrue(delayedCall.active()) self.assertIn(delayedCall, reactor.getDelayedCalls()) class GlibTimeTestsBuilder(ReactorBuilder): """ Builder for defining tests relating to L{IReactorTime} for reactors based off glib. """ requiredInterfaces = (IReactorTime,) if platform.isWindows(): _reactors = ["twisted.internet.gtk2reactor.PortableGtkReactor"] else: _reactors = ["twisted.internet.glib2reactor.Glib2Reactor", "twisted.internet.gtk2reactor.Gtk2Reactor"] def test_timeout_add(self): """ A L{reactor.callLater<twisted.internet.interfaces.IReactorTime.callLater>} call scheduled from a C{gobject.timeout_add} call is run on time. """ import gobject reactor = self.buildReactor() result = [] def gschedule(): reactor.callLater(0, callback) return 0 def callback(): result.append(True) reactor.stop() reactor.callWhenRunning(gobject.timeout_add, 10, gschedule) self.runReactor(reactor, 5) self.assertEqual(result, [True]) globals().update(TimeTestsBuilder.makeTestCaseClasses()) globals().update(GlibTimeTestsBuilder.makeTestCaseClasses())