codekingpro/portable-devtools
114k
1# testing/plugin/pytestplugin.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9from __future__ import annotations
10
11import argparse
12import collections
13from functools import update_wrapper
14import inspect
15import itertools
16import operator
17import os
18import re
19import sys
20from typing import TYPE_CHECKING
21import uuid
22
23import pytest
24
25try:
26 # installed by bootstrap.py
27 if not TYPE_CHECKING:
28 import sqla_plugin_base as plugin_base
29except ImportError:
30 # assume we're a package, use traditional import
31 from . import plugin_base
32
33
34def pytest_addoption(parser):
35 group = parser.getgroup("sqlalchemy")
36
37 def make_option(name, **kw):
38 callback_ = kw.pop("callback", None)
39 if callback_:
40
41 class CallableAction(argparse.Action):
42 def __call__(
43 self, parser, namespace, values, option_string=None
44 ):
45 callback_(option_string, values, parser)
46
47 kw["action"] = CallableAction
48
49 zeroarg_callback = kw.pop("zeroarg_callback", None)
50 if zeroarg_callback:
51
52 class CallableAction(argparse.Action):
53 def __init__(
54 self,
55 option_strings,
56 dest,
57 default=False,
58 required=False,
59 help=None, # noqa
60 ):
61 super().__init__(
62 option_strings=option_strings,
63 dest=dest,
64 nargs=0,
65 const=True,
66 default=default,
67 required=required,
68 help=help,
69 )
70
71 def __call__(
72 self, parser, namespace, values, option_string=None
73 ):
74 zeroarg_callback(option_string, values, parser)
75
76 kw["action"] = CallableAction
77
78 group.addoption(name, **kw)
79
80 plugin_base.setup_options(make_option)
81
82
83def pytest_configure(config: pytest.Config):
84 plugin_base.read_config(config.rootpath)
85 if plugin_base.exclude_tags or plugin_base.include_tags:
86 new_expr = " and ".join(
87 list(plugin_base.include_tags)
88 + [f"not {tag}" for tag in plugin_base.exclude_tags]
89 )
90
91 if config.option.markexpr:
92 config.option.markexpr += f" and {new_expr}"
93 else:
94 config.option.markexpr = new_expr
95
96 if config.pluginmanager.hasplugin("xdist"):
97 config.pluginmanager.register(XDistHooks())
98
99 if hasattr(config, "workerinput"):
100 plugin_base.restore_important_follower_config(config.workerinput)
101 plugin_base.configure_follower(config.workerinput["follower_ident"])
102 else:
103 if config.option.write_idents and os.path.exists(
104 config.option.write_idents
105 ):
106 os.remove(config.option.write_idents)
107
108 plugin_base.pre_begin(config.option)
109
110 plugin_base.set_coverage_flag(
111 bool(getattr(config.option, "cov_source", False))
112 )
113
114 plugin_base.set_fixture_functions(PytestFixtureFunctions)
115
116 if config.option.dump_pyannotate:
117 global DUMP_PYANNOTATE
118 DUMP_PYANNOTATE = True
119
120
121DUMP_PYANNOTATE = False
122
123
124@pytest.fixture(autouse=True)
125def collect_types_fixture():
126 if DUMP_PYANNOTATE:
127 from pyannotate_runtime import collect_types
128
129 collect_types.start()
130 yield
131 if DUMP_PYANNOTATE:
132 collect_types.stop()
133
134
135def _log_sqlalchemy_info(session):
136 import sqlalchemy
137 from sqlalchemy import __version__
138 from sqlalchemy.util import has_compiled_ext
139 from sqlalchemy.util._has_cy import _CYEXTENSION_MSG
140
141 greet = "sqlalchemy installation"
142 site = "no user site" if sys.flags.no_user_site else "user site loaded"
143 msgs = [
144 f"SQLAlchemy {__version__} ({site})",
145 f"Path: {sqlalchemy.__file__}",
146 ]
147
148 if has_compiled_ext():
149 from sqlalchemy.cyextension import util
150
151 msgs.append(f"compiled extension enabled, e.g. {util.__file__} ")
152 else:
153 msgs.append(f"compiled extension not enabled; {_CYEXTENSION_MSG}")
154
155 pm = session.config.pluginmanager.get_plugin("terminalreporter")
156 if pm:
157 pm.write_sep("=", greet)
158 for m in msgs:
159 pm.write_line(m)
160 else:
161 # fancy pants reporter not found, fallback to plain print
162 print("=" * 25, greet, "=" * 25)
163 for m in msgs:
164 print(m)
165
166
167def pytest_sessionstart(session):
168 from sqlalchemy.testing import asyncio
169
170 _log_sqlalchemy_info(session)
171 asyncio._assume_async(plugin_base.post_begin)
172
173
174def pytest_sessionfinish(session):
175 from sqlalchemy.testing import asyncio
176
177 asyncio._maybe_async_provisioning(plugin_base.final_process_cleanup)
178
179 if session.config.option.dump_pyannotate:
180 from pyannotate_runtime import collect_types
181
182 collect_types.dump_stats(session.config.option.dump_pyannotate)
183
184
185def pytest_unconfigure(config):
186 from sqlalchemy.testing import asyncio
187
188 asyncio._shutdown()
189
190
191def pytest_collection_finish(session):
192 if session.config.option.dump_pyannotate:
193 from pyannotate_runtime import collect_types
194
195 lib_sqlalchemy = os.path.abspath("lib/sqlalchemy")
196
197 def _filter(filename):
198 filename = os.path.normpath(os.path.abspath(filename))
199 if "lib/sqlalchemy" not in os.path.commonpath(
200 [filename, lib_sqlalchemy]
201 ):
202 return None
203 if "testing" in filename:
204 return None
205
206 return filename
207
208 collect_types.init_types_collection(filter_filename=_filter)
209
210
211class XDistHooks:
212 def pytest_configure_node(self, node):
213 from sqlalchemy.testing import provision
214 from sqlalchemy.testing import asyncio
215
216 # the master for each node fills workerinput dictionary
217 # which pytest-xdist will transfer to the subprocess
218
219 plugin_base.memoize_important_follower_config(node.workerinput)
220
221 node.workerinput["follower_ident"] = "test_%s" % uuid.uuid4().hex[0:12]
222
223 asyncio._maybe_async_provisioning(
224 provision.create_follower_db, node.workerinput["follower_ident"]
225 )
226
227 def pytest_testnodedown(self, node, error):
228 from sqlalchemy.testing import provision
229 from sqlalchemy.testing import asyncio
230
231 asyncio._maybe_async_provisioning(
232 provision.drop_follower_db, node.workerinput["follower_ident"]
233 )
234
235
236def pytest_collection_modifyitems(session, config, items):
237 # look for all those classes that specify __backend__ and
238 # expand them out into per-database test cases.
239
240 # this is much easier to do within pytest_pycollect_makeitem, however
241 # pytest is iterating through cls.__dict__ as makeitem is
242 # called which causes a "dictionary changed size" error on py3k.
243 # I'd submit a pullreq for them to turn it into a list first, but
244 # it's to suit the rather odd use case here which is that we are adding
245 # new classes to a module on the fly.
246
247 from sqlalchemy.testing import asyncio
248
249 rebuilt_items = collections.defaultdict(
250 lambda: collections.defaultdict(list)
251 )
252
253 items[:] = [
254 item
255 for item in items
256 if item.getparent(pytest.Class) is not None
257 and not item.getparent(pytest.Class).name.startswith("_")
258 ]
259
260 test_classes = {item.getparent(pytest.Class) for item in items}
261
262 def collect(element):
263 for inst_or_fn in element.collect():
264 if isinstance(inst_or_fn, pytest.Collector):
265 yield from collect(inst_or_fn)
266 else:
267 yield inst_or_fn
268
269 def setup_test_classes():
270 for test_class in test_classes:
271 # transfer legacy __backend__ and __sparse_backend__ symbols
272 # to be markers
273 add_markers = set()
274 if getattr(test_class.cls, "__backend__", False) or getattr(
275 test_class.cls, "__only_on__", False
276 ):
277 add_markers = {"backend"}
278 elif getattr(test_class.cls, "__sparse_backend__", False):
279 add_markers = {"sparse_backend"}
280 else:
281 add_markers = frozenset()
282
283 existing_markers = {
284 mark.name for mark in test_class.iter_markers()
285 }
286 add_markers = add_markers - existing_markers
287 all_markers = existing_markers.union(add_markers)
288
289 for marker in add_markers:
290 test_class.add_marker(marker)
291
292 for sub_cls in plugin_base.generate_sub_tests(
293 test_class.cls, test_class.module, all_markers
294 ):
295 if sub_cls is not test_class.cls:
296 per_cls_dict = rebuilt_items[test_class.cls]
297
298 module = test_class.getparent(pytest.Module)
299
300 new_cls = pytest.Class.from_parent(
301 name=sub_cls.__name__, parent=module
302 )
303 for marker in add_markers:
304 new_cls.add_marker(marker)
305
306 for fn in collect(new_cls):
307 per_cls_dict[fn.name].append(fn)
308
309 # class requirements will sometimes need to access the DB to check
310 # capabilities, so need to do this for async
311 asyncio._maybe_async_provisioning(setup_test_classes)
312
313 newitems = []
314 for item in items:
315 cls_ = item.cls
316 if cls_ in rebuilt_items:
317 newitems.extend(rebuilt_items[cls_][item.name])
318 else:
319 newitems.append(item)
320
321 # seems like the functions attached to a test class aren't sorted already?
322 # is that true and why's that? (when using unittest, they're sorted)
323 items[:] = sorted(
324 newitems,
325 key=lambda item: (
326 item.getparent(pytest.Module).name,
327 item.getparent(pytest.Class).name,
328 item.name,
329 ),
330 )
331
332
333def pytest_pycollect_makeitem(collector, name, obj):
334 if inspect.isclass(obj) and plugin_base.want_class(name, obj):
335 from sqlalchemy.testing import config
336
337 if config.any_async:
338 obj = _apply_maybe_async(obj)
339
340 return [
341 pytest.Class.from_parent(
342 name=parametrize_cls.__name__, parent=collector
343 )
344 for parametrize_cls in _parametrize_cls(collector.module, obj)
345 ]
346 elif (
347 inspect.isfunction(obj)
348 and collector.cls is not None
349 and plugin_base.want_method(collector.cls, obj)
350 ):
351 # None means, fall back to default logic, which includes
352 # method-level parametrize
353 return None
354 else:
355 # empty list means skip this item
356 return []
357
358
359def _is_wrapped_coroutine_function(fn):
360 while hasattr(fn, "__wrapped__"):
361 fn = fn.__wrapped__
362
363 return inspect.iscoroutinefunction(fn)
364
365
366def _apply_maybe_async(obj, recurse=True):
367 from sqlalchemy.testing import asyncio
368
369 for name, value in vars(obj).items():
370 if (
371 (callable(value) or isinstance(value, classmethod))
372 and not getattr(value, "_maybe_async_applied", False)
373 and (name.startswith("test_"))
374 and not _is_wrapped_coroutine_function(value)
375 ):
376 is_classmethod = False
377 if isinstance(value, classmethod):
378 value = value.__func__
379 is_classmethod = True
380
381 @_pytest_fn_decorator
382 def make_async(fn, *args, **kwargs):
383 return asyncio._maybe_async(fn, *args, **kwargs)
384
385 do_async = make_async(value)
386 if is_classmethod:
387 do_async = classmethod(do_async)
388 do_async._maybe_async_applied = True
389
390 setattr(obj, name, do_async)
391 if recurse:
392 for cls in obj.mro()[1:]:
393 if cls != object:
394 _apply_maybe_async(cls, False)
395 return obj
396
397
398def _parametrize_cls(module, cls):
399 """implement a class-based version of pytest parametrize."""
400
401 if "_sa_parametrize" not in cls.__dict__:
402 return [cls]
403
404 _sa_parametrize = cls._sa_parametrize
405 classes = []
406 for full_param_set in itertools.product(
407 *[params for argname, params in _sa_parametrize]
408 ):
409 cls_variables = {}
410
411 for argname, param in zip(
412 [_sa_param[0] for _sa_param in _sa_parametrize], full_param_set
413 ):
414 if not argname:
415 raise TypeError("need argnames for class-based combinations")
416 argname_split = re.split(r",\s*", argname)
417 for arg, val in zip(argname_split, param.values):
418 cls_variables[arg] = val
419 parametrized_name = "_".join(
420 re.sub(r"\W", "", token)
421 for param in full_param_set
422 for token in param.id.split("-")
423 )
424 name = "%s_%s" % (cls.__name__, parametrized_name)
425 newcls = type.__new__(type, name, (cls,), cls_variables)
426 setattr(module, name, newcls)
427 classes.append(newcls)
428 return classes
429
430
431_current_class = None
432
433
434def pytest_runtest_setup(item):
435 from sqlalchemy.testing import asyncio
436
437 # pytest_runtest_setup runs *before* pytest fixtures with scope="class".
438 # plugin_base.start_test_class_outside_fixtures may opt to raise SkipTest
439 # for the whole class and has to run things that are across all current
440 # databases, so we run this outside of the pytest fixture system altogether
441 # and ensure asyncio greenlet if any engines are async
442
443 global _current_class
444
445 if isinstance(item, pytest.Function) and _current_class is None:
446 asyncio._maybe_async_provisioning(
447 plugin_base.start_test_class_outside_fixtures,
448 item.cls,
449 )
450 _current_class = item.getparent(pytest.Class)
451
452
453@pytest.hookimpl(hookwrapper=True)
454def pytest_runtest_teardown(item, nextitem):
455 # runs inside of pytest function fixture scope
456 # after test function runs
457
458 from sqlalchemy.testing import asyncio
459
460 asyncio._maybe_async(plugin_base.after_test, item)
461
462 yield
463 # this is now after all the fixture teardown have run, the class can be
464 # finalized. Since pytest v7 this finalizer can no longer be added in
465 # pytest_runtest_setup since the class has not yet been setup at that
466 # time.
467 # See https://github.com/pytest-dev/pytest/issues/9343
468 global _current_class, _current_report
469
470 if _current_class is not None and (
471 # last test or a new class
472 nextitem is None
473 or nextitem.getparent(pytest.Class) is not _current_class
474 ):
475 _current_class = None
476
477 try:
478 asyncio._maybe_async_provisioning(
479 plugin_base.stop_test_class_outside_fixtures, item.cls
480 )
481 except Exception as e:
482 # in case of an exception during teardown attach the original
483 # error to the exception message, otherwise it will get lost
484 if _current_report.failed:
485 if not e.args:
486 e.args = (
487 "__Original test failure__:\n"
488 + _current_report.longreprtext,
489 )
490 elif e.args[-1] and isinstance(e.args[-1], str):
491 args = list(e.args)
492 args[-1] += (
493 "\n__Original test failure__:\n"
494 + _current_report.longreprtext
495 )
496 e.args = tuple(args)
497 else:
498 e.args += (
499 "__Original test failure__",
500 _current_report.longreprtext,
501 )
502 raise
503 finally:
504 _current_report = None
505
506
507def pytest_runtest_call(item):
508 # runs inside of pytest function fixture scope
509 # before test function runs
510
511 from sqlalchemy.testing import asyncio
512
513 asyncio._maybe_async(
514 plugin_base.before_test,
515 item,
516 item.module.__name__,
517 item.cls,
518 item.name,
519 )
520
521
522_current_report = None
523
524
525def pytest_runtest_logreport(report):
526 global _current_report
527 if report.when == "call":
528 _current_report = report
529
530
531@pytest.fixture(scope="class")
532def setup_class_methods(request):
533 from sqlalchemy.testing import asyncio
534
535 cls = request.cls
536
537 if hasattr(cls, "setup_test_class"):
538 asyncio._maybe_async(cls.setup_test_class)
539
540 yield
541
542 if hasattr(cls, "teardown_test_class"):
543 asyncio._maybe_async(cls.teardown_test_class)
544
545 asyncio._maybe_async(plugin_base.stop_test_class, cls)
546
547
548@pytest.fixture(scope="function")
549def setup_test_methods(request):
550 from sqlalchemy.testing import asyncio
551
552 # called for each test
553
554 self = request.instance
555
556 # before this fixture runs:
557
558 # 1. function level "autouse" fixtures under py3k (examples: TablesTest
559 # define tables / data, MappedTest define tables / mappers / data)
560
561 # 2. was for p2k. no longer applies
562
563 # 3. run outer xdist-style setup
564 if hasattr(self, "setup_test"):
565 asyncio._maybe_async(self.setup_test)
566
567 # alembic test suite is using setUp and tearDown
568 # xdist methods; support these in the test suite
569 # for the near term
570 if hasattr(self, "setUp"):
571 asyncio._maybe_async(self.setUp)
572
573 # inside the yield:
574 # 4. function level fixtures defined on test functions themselves,
575 # e.g. "connection", "metadata" run next
576
577 # 5. pytest hook pytest_runtest_call then runs
578
579 # 6. test itself runs
580
581 yield
582
583 # yield finishes:
584
585 # 7. function level fixtures defined on test functions
586 # themselves, e.g. "connection" rolls back the transaction, "metadata"
587 # emits drop all
588
589 # 8. pytest hook pytest_runtest_teardown hook runs, this is associated
590 # with fixtures close all sessions, provisioning.stop_test_class(),
591 # engines.testing_reaper -> ensure all connection pool connections
592 # are returned, engines created by testing_engine that aren't the
593 # config engine are disposed
594
595 asyncio._maybe_async(plugin_base.after_test_fixtures, self)
596
597 # 10. run xdist-style teardown
598 if hasattr(self, "tearDown"):
599 asyncio._maybe_async(self.tearDown)
600
601 if hasattr(self, "teardown_test"):
602 asyncio._maybe_async(self.teardown_test)
603
604 # 11. was for p2k. no longer applies
605
606 # 12. function level "autouse" fixtures under py3k (examples: TablesTest /
607 # MappedTest delete table data, possibly drop tables and clear mappers
608 # depending on the flags defined by the test class)
609
610
611def _pytest_fn_decorator(target):
612 """Port of langhelpers.decorator with pytest-specific tricks."""
613
614 from sqlalchemy.util.langhelpers import format_argspec_plus
615 from sqlalchemy.util.compat import inspect_getfullargspec
616
617 def _exec_code_in_env(code, env, fn_name):
618 # note this is affected by "from __future__ import annotations" at
619 # the top; exec'ed code will use non-evaluated annotations
620 # which allows us to be more flexible with code rendering
621 # in format_argpsec_plus()
622 exec(code, env)
623 return env[fn_name]
624
625 def decorate(fn, add_positional_parameters=()):
626 spec = inspect_getfullargspec(fn)
627 if add_positional_parameters:
628 spec.args.extend(add_positional_parameters)
629
630 metadata = dict(
631 __target_fn="__target_fn", __orig_fn="__orig_fn", name=fn.__name__
632 )
633 metadata.update(format_argspec_plus(spec, grouped=False))
634 code = (
635 """\
636def %(name)s%(grouped_args)s:
637 return %(__target_fn)s(%(__orig_fn)s, %(apply_kw)s)
638"""
639 % metadata
640 )
641 decorated = _exec_code_in_env(
642 code, {"__target_fn": target, "__orig_fn": fn}, fn.__name__
643 )
644 if not add_positional_parameters:
645 decorated.__defaults__ = getattr(fn, "__func__", fn).__defaults__
646 decorated.__wrapped__ = fn
647 return update_wrapper(decorated, fn)
648 else:
649 # this is the pytest hacky part. don't do a full update wrapper
650 # because pytest is really being sneaky about finding the args
651 # for the wrapped function
652 decorated.__module__ = fn.__module__
653 decorated.__name__ = fn.__name__
654 if hasattr(fn, "pytestmark"):
655 decorated.pytestmark = fn.pytestmark
656 return decorated
657
658 return decorate
659
660
661class PytestFixtureFunctions(plugin_base.FixtureFunctions):
662 def skip_test_exception(self, *arg, **kw):
663 return pytest.skip.Exception(*arg, **kw)
664
665 @property
666 def add_to_marker(self):
667 return pytest.mark
668
669 def mark_base_test_class(self):
670 return pytest.mark.usefixtures(
671 "setup_class_methods", "setup_test_methods"
672 )
673
674 _combination_id_fns = {
675 "i": lambda obj: obj,
676 "r": repr,
677 "s": str,
678 "n": lambda obj: (
679 obj.__name__ if hasattr(obj, "__name__") else type(obj).__name__
680 ),
681 }
682
683 def combinations(self, *arg_sets, **kw):
684 """Facade for pytest.mark.parametrize.
685
686 Automatically derives argument names from the callable which in our
687 case is always a method on a class with positional arguments.
688
689 ids for parameter sets are derived using an optional template.
690
691 """
692 from sqlalchemy.testing import exclusions
693
694 if len(arg_sets) == 1 and hasattr(arg_sets[0], "__next__"):
695 arg_sets = list(arg_sets[0])
696
697 argnames = kw.pop("argnames", None)
698
699 def _filter_exclusions(args):
700 result = []
701 gathered_exclusions = []
702 for a in args:
703 if isinstance(a, exclusions.compound):
704 gathered_exclusions.append(a)
705 else:
706 result.append(a)
707
708 return result, gathered_exclusions
709
710 id_ = kw.pop("id_", None)
711
712 tobuild_pytest_params = []
713 has_exclusions = False
714 if id_:
715 _combination_id_fns = self._combination_id_fns
716
717 # because itemgetter is not consistent for one argument vs.
718 # multiple, make it multiple in all cases and use a slice
719 # to omit the first argument
720 _arg_getter = operator.itemgetter(
721 0,
722 *[
723 idx
724 for idx, char in enumerate(id_)
725 if char in ("n", "r", "s", "a")
726 ],
727 )
728 fns = [
729 (operator.itemgetter(idx), _combination_id_fns[char])
730 for idx, char in enumerate(id_)
731 if char in _combination_id_fns
732 ]
733
734 for arg in arg_sets:
735 if not isinstance(arg, tuple):
736 arg = (arg,)
737
738 fn_params, param_exclusions = _filter_exclusions(arg)
739
740 parameters = _arg_getter(fn_params)[1:]
741
742 if param_exclusions:
743 has_exclusions = True
744
745 tobuild_pytest_params.append(
746 (
747 parameters,
748 param_exclusions,
749 "-".join(
750 comb_fn(getter(arg)) for getter, comb_fn in fns
751 ),
752 )
753 )
754
755 else:
756 for arg in arg_sets:
757 if not isinstance(arg, tuple):
758 arg = (arg,)
759
760 fn_params, param_exclusions = _filter_exclusions(arg)
761
762 if param_exclusions:
763 has_exclusions = True
764
765 tobuild_pytest_params.append(
766 (fn_params, param_exclusions, None)
767 )
768
769 pytest_params = []
770 for parameters, param_exclusions, id_ in tobuild_pytest_params:
771 if has_exclusions:
772 parameters += (param_exclusions,)
773
774 param = pytest.param(*parameters, id=id_)
775 pytest_params.append(param)
776
777 def decorate(fn):
778 if inspect.isclass(fn):
779 if has_exclusions:
780 raise NotImplementedError(
781 "exclusions not supported for class level combinations"
782 )
783 if "_sa_parametrize" not in fn.__dict__:
784 fn._sa_parametrize = []
785 fn._sa_parametrize.append((argnames, pytest_params))
786 return fn
787 else:
788 _fn_argnames = inspect.getfullargspec(fn).args[1:]
789 if argnames is None:
790 _argnames = _fn_argnames
791 else:
792 _argnames = re.split(r", *", argnames)
793
794 if has_exclusions:
795 existing_exl = sum(
796 1 for n in _fn_argnames if n.startswith("_exclusions")
797 )
798 current_exclusion_name = f"_exclusions_{existing_exl}"
799 _argnames += [current_exclusion_name]
800
801 @_pytest_fn_decorator
802 def check_exclusions(fn, *args, **kw):
803 _exclusions = args[-1]
804 if _exclusions:
805 exlu = exclusions.compound().add(*_exclusions)
806 fn = exlu(fn)
807 return fn(*args[:-1], **kw)
808
809 fn = check_exclusions(
810 fn, add_positional_parameters=(current_exclusion_name,)
811 )
812
813 return pytest.mark.parametrize(_argnames, pytest_params)(fn)
814
815 return decorate
816
817 def param_ident(self, *parameters):
818 ident = parameters[0]
819 return pytest.param(*parameters[1:], id=ident)
820
821 def fixture(self, *arg, **kw):
822 from sqlalchemy.testing import config
823 from sqlalchemy.testing import asyncio
824
825 # wrapping pytest.fixture function. determine if
826 # decorator was called as @fixture or @fixture().
827 if len(arg) > 0 and callable(arg[0]):
828 # was called as @fixture(), we have the function to wrap.
829 fn = arg[0]
830 arg = arg[1:]
831 else:
832 # was called as @fixture, don't have the function yet.
833 fn = None
834
835 # create a pytest.fixture marker. because the fn is not being
836 # passed, this is always a pytest.FixtureFunctionMarker()
837 # object (or whatever pytest is calling it when you read this)
838 # that is waiting for a function.
839 fixture = pytest.fixture(*arg, **kw)
840
841 # now apply wrappers to the function, including fixture itself
842
843 def wrap(fn):
844 if config.any_async:
845 fn = asyncio._maybe_async_wrapper(fn)
846 # other wrappers may be added here
847
848 # now apply FixtureFunctionMarker
849 fn = fixture(fn)
850
851 return fn
852
853 if fn:
854 return wrap(fn)
855 else:
856 return wrap
857
858 def get_current_test_name(self):
859 return os.environ.get("PYTEST_CURRENT_TEST")
860
861 def async_test(self, fn):
862 from sqlalchemy.testing import asyncio
863
864 @_pytest_fn_decorator
865 def decorate(fn, *args, **kwargs):
866 asyncio._run_coroutine_function(fn, *args, **kwargs)
867
868 return decorate(fn)
869