-
Notifications
You must be signed in to change notification settings - Fork 214
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
5 changed files
with
348 additions
and
39 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,8 +1,8 @@ | ||
"""pytest-bdd public API.""" | ||
|
||
from pytest_bdd.steps import given, when, then | ||
from pytest_bdd.scenario import scenario, scenarios, async_scenario, async_scenarios | ||
from pytest_bdd.scenario import scenario, scenarios | ||
|
||
__version__ = "4.0.2" | ||
|
||
__all__ = ["given", "when", "then", "scenario", "scenarios", "async_scenario", "async_scenarios"] | ||
__all__ = ["given", "when", "then", "scenario", "scenarios"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,22 +1,340 @@ | ||
__all__ = [ | ||
"async_scenario", | ||
"async_scenarios", | ||
"scenario", | ||
"scenarios", | ||
"find_argumented_step_fixture_name", | ||
"make_python_docstring", | ||
"make_python_name", | ||
"make_string_literal", | ||
"get_python_name_generator", | ||
] | ||
|
||
from ._async.scenario import scenario as async_scenario, scenarios as async_scenarios | ||
from ._sync.scenario import ( | ||
scenario, | ||
scenarios, | ||
find_argumented_step_fixture_name, | ||
make_python_docstring, | ||
make_python_name, | ||
make_string_literal, | ||
get_python_name_generator, | ||
"""Scenario implementation. | ||
The pytest will collect the test case and the steps will be executed | ||
line by line. | ||
Example: | ||
test_publish_article = scenario( | ||
feature_name="publish_article.feature", | ||
scenario_name="Publishing the article", | ||
) | ||
""" | ||
import contextlib | ||
import collections | ||
import os | ||
import re | ||
|
||
import pytest | ||
from _pytest.fixtures import FixtureLookupError | ||
|
||
from . import exceptions | ||
from .feature import get_feature, get_features | ||
from .steps import get_step_fixture_name, inject_fixture | ||
from .utils import CONFIG_STACK, get_args, get_caller_module_locals, get_caller_module_path | ||
|
||
PYTHON_REPLACE_REGEX = re.compile(r"\W") | ||
ALPHA_REGEX = re.compile(r"^\d+_*") | ||
|
||
|
||
def find_argumented_step_fixture_name(name, type_, fixturemanager, request=None): | ||
"""Find argumented step fixture name.""" | ||
# happens to be that _arg2fixturedefs is changed during the iteration so we use a copy | ||
for fixturename, fixturedefs in list(fixturemanager._arg2fixturedefs.items()): | ||
for fixturedef in fixturedefs: | ||
parser = getattr(fixturedef.func, "parser", None) | ||
if parser is None: | ||
continue | ||
match = parser.is_matching(name) | ||
if not match: | ||
continue | ||
|
||
converters = getattr(fixturedef.func, "converters", {}) | ||
for arg, value in parser.parse_arguments(name).items(): | ||
if arg in converters: | ||
value = converters[arg](value) | ||
if request: | ||
inject_fixture(request, arg, value) | ||
parser_name = get_step_fixture_name(parser.name, type_) | ||
if request: | ||
try: | ||
request.getfixturevalue(parser_name) | ||
except FixtureLookupError: | ||
continue | ||
return parser_name | ||
|
||
|
||
def _find_step_function(request, step, scenario): | ||
"""Match the step defined by the regular expression pattern. | ||
:param request: PyTest request object. | ||
:param step: Step. | ||
:param scenario: Scenario. | ||
:return: Function of the step. | ||
:rtype: function | ||
""" | ||
name = step.name | ||
try: | ||
# Simple case where no parser is used for the step | ||
return request.getfixturevalue(get_step_fixture_name(name, step.type)) | ||
except FixtureLookupError: | ||
try: | ||
# Could not find a fixture with the same name, let's see if there is a parser involved | ||
name = find_argumented_step_fixture_name(name, step.type, request._fixturemanager, request) | ||
if name: | ||
return request.getfixturevalue(name) | ||
raise | ||
except FixtureLookupError: | ||
raise exceptions.StepDefinitionNotFoundError( | ||
f"Step definition is not found: {step}. " | ||
f'Line {step.line_number} in scenario "{scenario.name}" in the feature "{scenario.feature.filename}"' | ||
) | ||
|
||
|
||
async def _execute_step_function(request, scenario, step, step_func, sync): | ||
"""Execute step function. | ||
:param request: PyTest request. | ||
:param scenario: Scenario. | ||
:param step: Step. | ||
:param function step_func: Step function. | ||
:param example: Example table. | ||
""" | ||
kw = dict(request=request, feature=scenario.feature, scenario=scenario, step=step, step_func=step_func) | ||
|
||
request.config.hook.pytest_bdd_before_step(**kw) | ||
|
||
kw["step_func_args"] = {} | ||
try: | ||
# Get the step argument values. | ||
kwargs = {arg: request.getfixturevalue(arg) for arg in get_args(step_func)} | ||
kw["step_func_args"] = kwargs | ||
|
||
request.config.hook.pytest_bdd_before_step_call(**kw) | ||
target_fixture = getattr(step_func, "target_fixture", None) | ||
# Execute the step. | ||
if sync: | ||
return_value = step_func(**kwargs) | ||
else: | ||
return_value = await step_func(**kwargs) | ||
if target_fixture: | ||
inject_fixture(request, target_fixture, return_value) | ||
|
||
request.config.hook.pytest_bdd_after_step(**kw) | ||
except Exception as exception: | ||
request.config.hook.pytest_bdd_step_error(exception=exception, **kw) | ||
raise | ||
|
||
|
||
async def _execute_scenario(feature, scenario, request, sync): | ||
"""Execute the scenario. | ||
:param feature: Feature. | ||
:param scenario: Scenario. | ||
:param request: request. | ||
:param encoding: Encoding. | ||
""" | ||
request.config.hook.pytest_bdd_before_scenario(request=request, feature=feature, scenario=scenario) | ||
|
||
try: | ||
# Execute scenario steps | ||
for step in scenario.steps: | ||
try: | ||
step_func = _find_step_function(request, step, scenario) | ||
except exceptions.StepDefinitionNotFoundError as exception: | ||
request.config.hook.pytest_bdd_step_func_lookup_error( | ||
request=request, feature=feature, scenario=scenario, step=step, exception=exception | ||
) | ||
raise | ||
await _execute_step_function(request, scenario, step, step_func, sync) | ||
finally: | ||
request.config.hook.pytest_bdd_after_scenario(request=request, feature=feature, scenario=scenario) | ||
|
||
|
||
FakeRequest = collections.namedtuple("FakeRequest", ["module"]) | ||
|
||
|
||
def await_(fn, *args): | ||
v = fn(*args) | ||
with contextlib.closing(v.__await__()) as gen: | ||
try: | ||
gen.send(None) | ||
except StopIteration as e: | ||
return e.value | ||
else: | ||
raise RuntimeError("coro did not stop") | ||
|
||
|
||
def _get_scenario_decorator(feature, feature_name, scenario, scenario_name, *, sync): | ||
# HACK: Ideally we would use `def decorator(fn)`, but we want to return a custom exception | ||
# when the decorator is misused. | ||
# Pytest inspect the signature to determine the required fixtures, and in that case it would look | ||
# for a fixture called "fn" that doesn't exist (if it exists then it's even worse). | ||
# It will error with a "fixture 'fn' not found" message instead. | ||
# We can avoid this hack by using a pytest hook and check for misuse instead. | ||
def decorator(*args): | ||
if not args: | ||
raise exceptions.ScenarioIsDecoratorOnly( | ||
"scenario function can only be used as a decorator. Refer to the documentation." | ||
) | ||
[fn] = args | ||
args = get_args(fn) | ||
function_args = list(args) | ||
for arg in scenario.get_example_params(): | ||
if arg not in function_args: | ||
function_args.append(arg) | ||
|
||
if sync: | ||
|
||
@pytest.mark.usefixtures(*function_args) | ||
def scenario_wrapper(request): | ||
await_(_execute_scenario, feature, scenario, request, sync) | ||
return fn(*[request.getfixturevalue(arg) for arg in args]) | ||
|
||
else: | ||
|
||
@pytest.mark.usefixtures(*function_args) | ||
async def scenario_wrapper(request): | ||
await _execute_scenario(feature, scenario, request, sync) | ||
return await fn(*[request.getfixturevalue(arg) for arg in args]) | ||
|
||
for param_set in scenario.get_params(): | ||
if param_set: | ||
scenario_wrapper = pytest.mark.parametrize(*param_set)(scenario_wrapper) | ||
for tag in scenario.tags.union(feature.tags): | ||
config = CONFIG_STACK[-1] | ||
config.hook.pytest_bdd_apply_tag(tag=tag, function=scenario_wrapper) | ||
|
||
scenario_wrapper.__doc__ = f"{feature_name}: {scenario_name}" | ||
scenario_wrapper.__scenario__ = scenario | ||
scenario.test_function = scenario_wrapper | ||
return scenario_wrapper | ||
|
||
return decorator | ||
|
||
|
||
def scenario(feature_name, scenario_name, encoding="utf-8", example_converters=None, features_base_dir=None, sync=True): | ||
"""Scenario decorator. | ||
:param str feature_name: Feature file name. Absolute or relative to the configured feature base path. | ||
:param str scenario_name: Scenario name. | ||
:param str encoding: Feature file encoding. | ||
:param dict example_converters: optional `dict` of example converter function, where key is the name of the | ||
example parameter, and value is the converter function. | ||
""" | ||
|
||
scenario_name = str(scenario_name) | ||
caller_module_path = get_caller_module_path() | ||
|
||
# Get the feature | ||
if features_base_dir is None: | ||
features_base_dir = get_features_base_dir(caller_module_path) | ||
feature = get_feature(features_base_dir, feature_name, encoding=encoding) | ||
|
||
# Get the scenario | ||
try: | ||
scenario = feature.scenarios[scenario_name] | ||
except KeyError: | ||
feature_name = feature.name or "[Empty]" | ||
raise exceptions.ScenarioNotFound( | ||
f'Scenario "{scenario_name}" in feature "{feature_name}" in {feature.filename} is not found.' | ||
) | ||
|
||
scenario.example_converters = example_converters | ||
|
||
# Validate the scenario | ||
scenario.validate() | ||
|
||
return _get_scenario_decorator( | ||
feature=feature, feature_name=feature_name, scenario=scenario, scenario_name=scenario_name, sync=sync | ||
) | ||
|
||
|
||
def get_features_base_dir(caller_module_path): | ||
default_base_dir = os.path.dirname(caller_module_path) | ||
return get_from_ini("bdd_features_base_dir", default_base_dir) | ||
|
||
|
||
def get_from_ini(key, default): | ||
"""Get value from ini config. Return default if value has not been set. | ||
Use if the default value is dynamic. Otherwise set default on addini call. | ||
""" | ||
config = CONFIG_STACK[-1] | ||
value = config.getini(key) | ||
return value if value != "" else default | ||
|
||
|
||
def make_python_name(string): | ||
"""Make python attribute name out of a given string.""" | ||
string = re.sub(PYTHON_REPLACE_REGEX, "", string.replace(" ", "_")) | ||
return re.sub(ALPHA_REGEX, "", string).lower() | ||
|
||
|
||
def make_python_docstring(string): | ||
"""Make a python docstring literal out of a given string.""" | ||
return '"""{}."""'.format(string.replace('"""', '\\"\\"\\"')) | ||
|
||
|
||
def make_string_literal(string): | ||
"""Make python string literal out of a given string.""" | ||
return "'{}'".format(string.replace("'", "\\'")) | ||
|
||
|
||
def get_python_name_generator(name): | ||
"""Generate a sequence of suitable python names out of given arbitrary string name.""" | ||
python_name = make_python_name(name) | ||
suffix = "" | ||
index = 0 | ||
|
||
def get_name(): | ||
return f"test_{python_name}{suffix}" | ||
|
||
while True: | ||
yield get_name() | ||
index += 1 | ||
suffix = f"_{index}" | ||
|
||
|
||
def scenarios(*feature_paths, sync=True, **kwargs): | ||
"""Parse features from the paths and put all found scenarios in the caller module. | ||
:param *feature_paths: feature file paths to use for scenarios | ||
""" | ||
caller_locals = get_caller_module_locals() | ||
caller_path = get_caller_module_path() | ||
|
||
features_base_dir = kwargs.get("features_base_dir") | ||
if features_base_dir is None: | ||
features_base_dir = get_features_base_dir(caller_path) | ||
|
||
abs_feature_paths = [] | ||
for path in feature_paths: | ||
if not os.path.isabs(path): | ||
path = os.path.abspath(os.path.join(features_base_dir, path)) | ||
abs_feature_paths.append(path) | ||
found = False | ||
|
||
module_scenarios = frozenset( | ||
(attr.__scenario__.feature.filename, attr.__scenario__.name) | ||
for name, attr in caller_locals.items() | ||
if hasattr(attr, "__scenario__") | ||
) | ||
|
||
for feature in get_features(abs_feature_paths): | ||
for scenario_name, scenario_object in feature.scenarios.items(): | ||
# skip already bound scenarios | ||
if (scenario_object.feature.filename, scenario_name) not in module_scenarios: | ||
|
||
decorator = scenario(feature.filename, scenario_name, sync=sync, **kwargs) | ||
if sync: | ||
|
||
@decorator | ||
def _scenario(): | ||
pass # pragma: no cover | ||
|
||
else: | ||
|
||
@decorator | ||
async def _scenario(): | ||
pass # pragma: no cover | ||
|
||
for test_name in get_python_name_generator(scenario_name): | ||
if test_name not in caller_locals: | ||
# found an unique test name | ||
caller_locals[test_name] = _scenario | ||
break | ||
found = True | ||
if not found: | ||
raise exceptions.NoScenariosFound(abs_feature_paths) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,12 +1,3 @@ | ||
import unasync | ||
from setuptools import setup | ||
|
||
setup( | ||
cmdclass={ | ||
"build_py": unasync.cmdclass_build_py( | ||
rules=[ | ||
unasync.Rule("/pytest_bdd/_async/", "/pytest_bdd/_sync/"), | ||
] | ||
) | ||
} | ||
) | ||
setup() |
Oops, something went wrong.