/usr/local/lib/python3.6/site-packages/asynctest/__pycache__
Edit: /usr/local/lib/python3.6/site-packages/asynctest/__pycache__/case.cpython-36.pyc (14973B)
3
8%EgD @ s d Z ddlZddlZddlZddlZddlZddlZddlT ddlZ ddl
Z G dd dej Z G dd dej
Z
G dd d e
ejZG d
d d e
Zdd d
ZdS )a
Module ``case``
---------------
Enhance :class:`unittest.TestCase`:
* a new loop is issued and set as the default loop before each test, and
closed and disposed after,
* if the loop uses a selector, it will be wrapped with
:class:`asynctest.TestSelector`,
* a test method in a TestCase identified as a coroutine function or returning
a coroutine will run on the loop,
* :meth:`~TestCase.setUp()` and :meth:`~TestCase.tearDown()` methods can be
coroutine functions,
* cleanup functions registered with :meth:`~TestCase.addCleanup()` can be
coroutine functions,
* a test fails if the loop did not run during the test.
class-level set-up
~~~~~~~~~~~~~~~~~~
Since each test runs in its own loop, it is not possible to run
:meth:`~TestCase.setUpClass()` and :meth:`~TestCase.tearDownClass()` as
coroutines.
If one needs to perform set-up actions at the class level (meaning
once for all tests in the class), it should be done using a loop created for
this sole purpose and that is not shared with the tests. Ideally, the loop
shall be closed in the method which creates it.
If one really needs to share a loop between tests,
:attr:`TestCase.use_default_loop` can be set to ``True`` (as a class
attribute). The test case will use the loop returned by
:meth:`asyncio.get_event_loop()` instead of creating a new loop for each test.
This way, the event loop or event loop policy can be set during class-level
set-up and tear down.
N)*c @ sL e Zd Zdd Zdd Zdd Zdd Zd d
Zd d Zd
d Z dd Z
dS )_Policyc C s || _ || _|| _d | _d S )N)original_policyforbid_get_event_looploopwatcher)selfr r r r 8/usr/local/lib/python3.6/site-packages/asynctest/case.py__init__; s z_Policy.__init__c C s* | j rtdn| jr | jS | jj S d S )NzSTestCase.forbid_get_event_loop is True, asyncio.get_event_loop() must not be called)r AssertionErrorr r get_event_loop)r r r r
r
E s
z_Policy.get_event_loopc C s
| j j S )N)r new_event_loop)r r r r
r N s z_Policy.new_event_loopc C s | j j|}|| _|S )N)r set_event_loopr )r r resultr r r
r Q s z_Policy.set_event_loopc C s t tdstd S )NSafeChildWatcher)hasattrasyncioNotImplementedError)r r r r
_check_unixV s
z_Policy._check_unixc C sD | j | jr6| jd kr0tj | _| jj| j | jS | jj S d S )N)r r r r r attach_loopr get_child_watcher)r r r r
r Z s
z_Policy.get_child_watcherc C s4 | j | jr$| jj|}|| _|S | jj|S d S )N)r r r set_child_watcherr )r r r r r r
r e s z_Policy.set_child_watcherc C s | j r | j j | jjd d S )N)r closer r )r r r r
reset_watchern s
z_Policy.reset_watcherN) __name__
__module__ __qualname__r r
r r r r r r r r r r
r : s
r c s e Zd ZdZdZdZdZdd Zdd Zdd Z d
d Z
d d
Z d ddZ dd Z
dd Zejdd Z fddZejdd Zejdd Zejd d Zejd d Z ZS )!TestCaseaJ
A test which is a coroutine function or which returns a coroutine will run
on the loop.
Once the test returned, one or more assertions are checked. For instance,
a test fails if the loop didn't run. These checks can be enabled or
disabled using the :func:`~asynctest.fail_on` decorator.
By default, a new loop is created and is set as the default loop before
each test. Test authors can retrieve this loop with
:attr:`~asynctest.TestCase.loop`.
If :attr:`~asynctest.TestCase.use_default_loop` is set to ``True``, the
current default event loop is used instead. In this case, it is up to the
test author to deal with the state of the loop in each test: the loop might
be closed, callbacks and tasks may be scheduled by previous tests. It is
also up to the test author to close the loop and dispose the related
resources.
If :attr:`~asynctest.TestCase.forbid_get_event_loop` is set to ``True``,
a call to :func:`asyncio.get_event_loop()` will raise an
:exc:`AssertionError`. Since Python 3.6, calling
:func:`asyncio.get_event_loop()` from a callback or a coroutine will return
the running loop (instead of raising an exception).
These behaviors should be configured when defining the test case class::
class With_Reusable_Loop_TestCase(asynctest.TestCase):
use_default_loop = True
forbid_get_event_loop = False
def test_something(self):
pass
If :meth:`setUp()` and :meth:`tearDown()` are coroutine functions, they
will run on the loop. Note that :meth:`setUpClass()` and
:meth:`tearDownClass()` can not be coroutines.
.. versionadded:: 0.5
attribute :attr:`~asynctest.TestCase.use_default_loop`.
.. versionadded:: 0.7
attribute :attr:`~asynctest.TestCase.forbid_get_event_loop`.
In any case, the default loop is now reset to its original state
outside a test function.
.. versionadded:: 0.8
``ignore_loop`` has been deprecated in favor of the extensible
:func:`~asynctest.fail_on` decorator.
FNc C sR | j rtj | _d }ntj }| _ttj || j}tj| | j | j| _d S )N)
use_default_loopr r
r r r get_event_loop_policyr set_event_loop_policy _patch_loop)r r policyr r r
_init_loop s
zTestCase._init_loopc C sR t j }| js
.wraps..wrapper) functoolswrapstypes
MethodType)r/ r0 )r )r/ r
r2 s z#TestCase._patch_loop..wraps run_foreverr) )r5 r) ) r r, setattrgetattr
isinstancer selector_eventsBaseSelectorEventLoop asynctestselectorZ TestSelector _selector)r r r2 r/ r )r r
r" s
zTestCase._patch_loopc C sr | j t| | j}t|tjjd }|p0tjj | _| jj| tj | j
r^| j j | j
n| j
d| j _
d S )NF)r$ r7 _testMethodNamer; _fail_onZ
_FAIL_ON_ATTR_checkerZ before_testr iscoroutinefunctionZsetUpr r) r, )r testcheckerr r r
_setUp s zTestCase._setUpc C s6 t j| jr | jj| j n| j | jj| d S )N)r rA ZtearDownr r) r@ Z
check_test)r r r r
_tearDown s zTestCase._tearDownc ( C s |}|d kr.| j }t|dd }|d k r.| |j| t| | j}t| jdds^t|ddrz,t| jddpxt|dd}| j|| | W d |j| X d S t|dd}tjj |}z|| _
|j | | j W d Q R X |j
r2||_|j | dd | j| W d Q R X d|_|j | | j W d Q R X | jj| j | j x"|jD ]\}} | j||| qRW | j||j |j
r|r|jr| j||j n
| j| n
|j| |S |j| |d krt|d d }
|
d k r|
|jj d |_d | _
X d S )
N startTestRunZ__unittest_skip__FZ__unittest_skip_why__ Z __unittest_expecting_failure__T)ZisTest stopTestRun) ZdefaultTestResultr7 Z startTestr> __class__Z_addSkipZstopTestunittestZcase_Outcome_outcometestPartExecutorrD successexpecting_failure_run_test_methodrE r r)
doCleanupsr+ skippedZ_feedErrorsToResulterrorsZexpectedFailureZ_addExpectedFailureZ_addUnexpectedSuccessZ
addSuccessclear) r r Z orig_resultrF Z
testMethodZskip_whyrO outcomerB reasonrH r r r
run sd
z TestCase.runc C s | j zyd| jt| | j | j xD| jrj| jjd\}}}tj|r^| j j
||| q(||| q(W W n t k
r Y nX W d | j X d S )N )
rD rP r7 r> rE _cleanupspopr rA r r) Exceptionr+ )r functionr- r. r r r
debugL s
zTestCase.debugc C s | }t j|r | jj| d S )N)r iscoroutiner r) )r r/ r r r r
rP ] s
zTestCase._run_test_methodc
c sn | j ptjj }xV| jrf| jj \}}}|j| , tj|rP|||E dH n
||| W dQ R X qW |j S )zX
Execute all cleanup functions. Normally called for you after tearDown.
N)
rL rJ ZmockrK rZ r[ rM r rA rN )r rU r] r- r. r r r
rQ d s
zTestCase.doCleanupsc s t j|f||S )z
Add a function, with arguments, to be called when the test is
completed. If function is a coroutine function, it will run on the loop
before it's cleaned.
)super
addCleanup)r r] r- r. )rI r r
ra t s zTestCase.addCleanupc c s | j | |E dH S Q R X dS )z
Test that an exception of type ``exception`` is raised when an
exception is raised when awaiting ``awaitable``, a future or coroutine.
:see: :meth:`unittest.TestCase.assertRaises()`
N) assertRaises)r exception awaitabler r r
assertAsyncRaises| s zTestCase.assertAsyncRaisesc c s" | j || |E dH S Q R X dS )z
Like :meth:`assertAsyncRaises()` but also tests that ``regex`` matches
on the string representation of the raised exception.
:see: :meth:`unittest.TestCase.assertRaisesRegex()`
N)assertRaisesRegex)r rc regexrd r r r
assertAsyncRaisesRegex s z TestCase.assertAsyncRaisesRegexc c s | j | |E dH S Q R X dS )z
Test that a warning is triggered when awaiting ``awaitable``, a future
or a coroutine.
:see: :meth:`unittest.TestCase.assertWarns()`
N)Z assertWarns)r warningrd r r r
assertAsyncWarns s zTestCase.assertAsyncWarnsc c s" | j || |E dH S Q R X dS )z
Like :meth:`assertAsyncWarns()` but also tests that ``regex`` matches
on the message of the triggered warning.
:see: :meth:`unittest.TestCase.assertWarnsRegex()`
N)ZassertWarnsRegex)r ri rg rd r r r
assertAsyncWarnsRegex s z TestCase.assertAsyncWarnsRegex)N)r r r __doc__r r r r$ r+ r" rD rE rW r^ rP r coroutinerQ ra re rh rj rk
__classcell__r r )rI r
r w s$ 6
C r c @ s e Zd ZdZdS )FunctionTestCasezv
Enables the same features as :class:`~asynctest.TestCase`, but for
:class:`~asynctest.FunctionTestCase`.
N)r r r rl r r r r
ro s ro c sD e Zd ZdZ fddZejdd Zdd Zejdd Z Z
S )
ClockedTestCasez
Subclass of :class:`~asynctest.TestCase` with a controlled loop clock,
useful for testing timer based behaviour without slowing test run time.
The clock will only advance when :meth:`advance()` is called.
c s2 t j tj jj fdd j_d _d S )Nc s j S )N)_timer )r r r
s z,ClockedTestCase._init_loop..r )r` r$ r1 r2 r timerq )r )rI )r r
r$ s
zClockedTestCase._init_loopc c sz |dk rt dj|| j E dH | j| }x2| j }|dksH||krJP || _| j E dH q0W || _| j E dH dS )a
Fast forward time by a number of ``seconds``.
Callbacks scheduled to run up to the destination clock time will be
executed on time:
>>> self.loop.call_later(1, print_time)
>>> self.loop.call_later(2, self.loop.call_later, 1, print_time)
>>> await self.advance(3)
1
3
In this example, the third callback is scheduled at ``t = 2`` to be
executed at ``t + 1``. Hence, it will run at ``t = 3``. The callback as
been called on time.
r z#Cannot go back in time ({} seconds)N)
ValueErrorformat _drain_looprq _next_scheduled)r secondsZ target_time next_timer r r
advance s
zClockedTestCase.advancec C s( y| j jd jS tk
r" d S X d S )Nr )r
_scheduled_when
IndexError)r r r r
rw s z ClockedTestCase._next_scheduledc c sH xB| j }| jj r(|d ks&|| jkr(P tjdE d H d| j_qW d S )Nr T)rw r _readyrq r sleepZ_TestCase_asynctest_ran)r ry r r r
rv s
zClockedTestCase._drain_loop) r r r rl r$ r rm rz rw rv rn r r )rI r
rp s
$rp c C s0 t jdt tjjddi}| dkr(|S || S )zK
Ignore the error case where the loop did not run during the test.
zBignore_loop() is deprecated in favor of fail_on(unused_loop=False)Z unused_loopFN)warningswarnDeprecationWarningr; r? )funcrC r r r
ignore_loop s r )N)rl r r1 r3 rJ r' r Z
unittest.caseZasynctest.selectorr; Zasynctest._fail_onAbstractEventLoopPolicyr r ro rp r r r r r
+ s = 4B