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: _awaittests.py.3only
Close
# Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for C{await} support in Deferreds. These tests can only work and be imported on Python 3.5+! """ import types from twisted.internet.defer import Deferred, ensureDeferred, fail from twisted.trial.unittest import TestCase from twisted.test.proto_helpers import Clock class AwaitTests(TestCase): """ Tests for using Deferreds in conjunction with PEP-492. """ def test_awaitReturnsIterable(self): """ C{Deferred.__await__} returns an iterable. """ d = Deferred() awaitedDeferred = d.__await__() self.assertEqual(awaitedDeferred, iter(awaitedDeferred)) def test_ensureDeferred(self): """ L{ensureDeferred} will turn a coroutine into a L{Deferred}. """ async def run(): d = Deferred() d.callback("bar") await d res = await run2() return res async def run2(): d = Deferred() d.callback("foo") res = await d return res # It's a coroutine... r = run() self.assertIsInstance(r, types.CoroutineType) # Now it's a Deferred. d = ensureDeferred(r) self.assertIsInstance(d, Deferred) # The Deferred has the result we want. res = self.successResultOf(d) self.assertEqual(res, "foo") def test_basic(self): """ L{ensureDeferred} allows a function to C{await} on a L{Deferred}. """ async def run(): d = Deferred() d.callback("foo") res = await d return res d = ensureDeferred(run()) res = self.successResultOf(d) self.assertEqual(res, "foo") def test_exception(self): """ An exception in a coroutine wrapped with L{ensureDeferred} will cause the returned L{Deferred} to fire with a failure. """ async def run(): d = Deferred() d.callback("foo") await d raise ValueError("Oh no!") d = ensureDeferred(run()) res = self.failureResultOf(d) self.assertEqual(type(res.value), ValueError) self.assertEqual(res.value.args, ("Oh no!",)) def test_twoDeep(self): """ A coroutine wrapped with L{ensureDeferred} that awaits a L{Deferred} suspends its execution until the inner L{Deferred} fires. """ reactor = Clock() sections = [] async def runone(): sections.append(2) d = Deferred() reactor.callLater(1, d.callback, 2) await d sections.append(3) return "Yay!" async def run(): sections.append(1) result = await runone() sections.append(4) d = Deferred() reactor.callLater(1, d.callback, 1) await d sections.append(5) return result d = ensureDeferred(run()) reactor.advance(0.9) self.assertEqual(sections, [1, 2]) reactor.advance(0.1) self.assertEqual(sections, [1, 2, 3, 4]) reactor.advance(0.9) self.assertEqual(sections, [1, 2, 3, 4]) reactor.advance(0.1) self.assertEqual(sections, [1, 2, 3, 4, 5]) res = self.successResultOf(d) self.assertEqual(res, "Yay!") def test_reraise(self): """ Awaiting an already failed Deferred will raise the exception. """ async def test(): try: await fail(ValueError("Boom")) except ValueError as e: self.assertEqual(e.args, ("Boom",)) return 1 return 0 res = self.successResultOf(ensureDeferred(test())) self.assertEqual(res, 1) def test_chained(self): """ Awaiting a paused & chained Deferred will give the result when it has one. """ reactor = Clock() async def test(): d = Deferred() d2 = Deferred() d.addCallback(lambda ignored: d2) d.callback(None) reactor.callLater(0, d2.callback, "bye") return await d d = ensureDeferred(test()) reactor.advance(0.1) res = self.successResultOf(d) self.assertEqual(res, "bye")