1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
# -*- coding: UTF-8 -*-
"""
Unit tests for :mod:`behave.api.async_test`.
"""
# -- IMPORTS:
from __future__ import absolute_import, print_function
from behave.api.async_step import AsyncContext, use_or_create_async_context
from behave._stepimport import use_step_import_modules
from behave.runner import Context, Runner
import sys
from mock import Mock
import pytest
from .testing_support import StopWatch, SimpleStepContainer
from .testing_support_async import AsyncStepTheory
# -----------------------------------------------------------------------------
# ASYNC STEP EXAMPLES:
# -----------------------------------------------------------------------------
# if python_version >= 3.5:
# @step('an async coroutine step waits "{duration:f}" seconds')
# @async_run_until_complete
# async def step_async_step_waits_seconds(context, duration):
# print("async_step: Should sleep for %.3f seconds" % duration)
# await asyncio.sleep(duration)
#
# if python_version >= 3.4:
# @step('a tagged-coroutine async step waits "{duration:f}" seconds')
# @async_run_until_complete
# @asyncio.coroutine
# def step_async_step_waits_seconds2(context, duration):
# print("async_step2: Should sleep for %.3f seconds" % duration)
# yield from asyncio.sleep(duration)
#
# -----------------------------------------------------------------------------
# TEST MARKERS:
# -----------------------------------------------------------------------------
# DEPRECATED: @asyncio.coroutine decorator (since: Python >= 3.8)
_python_version = float("%s.%s" % sys.version_info[:2])
requires_py34_to_py37 = pytest.mark.skipif(not (3.4 <= _python_version < 3.8),
reason="Supported only for python.versions: 3.4 .. 3.7 (inclusive)")
# -----------------------------------------------------------------------------
# TESTSUITE:
# -----------------------------------------------------------------------------
@requires_py34_to_py37
class TestAsyncStepDecorator34(object):
def test_step_decorator_async_run_until_complete2(self):
step_container = SimpleStepContainer()
with use_step_import_modules(step_container):
# -- STEP-DEFINITIONS EXAMPLE (as MODULE SNIPPET):
# VARIANT 2: Use @asyncio.coroutine def step_impl()
from behave import step
from behave.api.async_step import async_run_until_complete
import asyncio
@step('a tagged-coroutine async step waits "{duration:f}" seconds')
@async_run_until_complete
@asyncio.coroutine
def step_async_step_waits_seconds2(context, duration):
yield from asyncio.sleep(duration)
# -- USES: async def step_impl(...) as async-step (coroutine)
AsyncStepTheory.validate(step_async_step_waits_seconds2)
# -- RUN ASYNC-STEP: Verify that it is behaving correctly.
# ENSURE: Execution of async-step matches expected duration.
context = Context(runner=Runner(config={}))
with StopWatch() as stop_watch:
step_async_step_waits_seconds2(context, duration=0.2)
assert abs(stop_watch.duration - 0.2) <= 0.05
class TestAsyncContext(object):
@staticmethod
def make_context():
return Context(runner=Runner(config={}))
def test_use_or_create_async_context__when_missing(self):
# -- CASE: AsynContext attribute is created with default_name
context = self.make_context()
context._push()
async_context = use_or_create_async_context(context)
assert isinstance(async_context, AsyncContext)
assert async_context.name == "async_context"
assert getattr(context, "async_context", None) is async_context
context._pop()
assert getattr(context, "async_context", None) is None
def test_use_or_create_async_context__when_exists(self):
# -- CASE: AsynContext attribute exists with default_name
context = self.make_context()
async_context0 = context.async_context = AsyncContext()
assert context.async_context.name == "async_context"
assert hasattr(context, AsyncContext.default_name)
async_context = use_or_create_async_context(context)
assert isinstance(async_context, AsyncContext)
assert async_context.name == "async_context"
assert getattr(context, "async_context", None) is async_context
assert async_context is async_context0
def test_use_or_create_async_context__when_missing_with_name(self):
# -- CASE: AsynContext attribute is created with own name
loop0 = Mock()
context = self.make_context()
async_context = use_or_create_async_context(context, "acontext", loop=loop0)
assert isinstance(async_context, AsyncContext)
assert async_context.name == "acontext"
assert getattr(context, "acontext", None) is async_context
assert async_context.loop is loop0
def test_use_or_create_async_context__when_exists_with_name(self):
# -- CASE: AsynContext attribute exists with own name
loop0 = Mock()
context = self.make_context()
async_context0 = context.acontext = AsyncContext(name="acontext", loop=loop0)
assert context.acontext.name == "acontext"
loop1 = Mock()
async_context = use_or_create_async_context(context, "acontext", loop=loop1)
assert isinstance(async_context, AsyncContext)
assert async_context is async_context0
assert getattr(context, "acontext", None) is async_context
assert async_context.loop is loop0
|
