pytestplugin.py 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194
  1. """NOTE: copied/adapted from SQLAlchemy master for backwards compatibility;
  2. this should be removable when Alembic targets SQLAlchemy 1.0.0.
  3. """
  4. try:
  5. # installed by bootstrap.py
  6. import alembic_plugin_base as plugin_base
  7. except ImportError:
  8. # assume we're a package, use traditional import
  9. from . import plugin_base
  10. import sys
  11. py3k = sys.version_info >= (3, 0)
  12. import pytest
  13. import argparse
  14. import inspect
  15. import collections
  16. import os
  17. try:
  18. import xdist # noqa
  19. has_xdist = True
  20. except ImportError:
  21. has_xdist = False
  22. def pytest_addoption(parser):
  23. group = parser.getgroup("sqlalchemy")
  24. def make_option(name, **kw):
  25. callback_ = kw.pop("callback", None)
  26. if callback_:
  27. class CallableAction(argparse.Action):
  28. def __call__(self, parser, namespace,
  29. values, option_string=None):
  30. callback_(option_string, values, parser)
  31. kw["action"] = CallableAction
  32. group.addoption(name, **kw)
  33. plugin_base.setup_options(make_option)
  34. plugin_base.read_config()
  35. def pytest_configure(config):
  36. if hasattr(config, "slaveinput"):
  37. plugin_base.restore_important_follower_config(config.slaveinput)
  38. plugin_base.configure_follower(
  39. config.slaveinput["follower_ident"]
  40. )
  41. if config.option.write_idents:
  42. with open(config.option.write_idents, "a") as file_:
  43. file_.write(config.slaveinput["follower_ident"] + "\n")
  44. else:
  45. if config.option.write_idents and \
  46. os.path.exists(config.option.write_idents):
  47. os.remove(config.option.write_idents)
  48. plugin_base.pre_begin(config.option)
  49. coverage = bool(getattr(config.option, "cov_source", False))
  50. plugin_base.set_coverage_flag(coverage)
  51. def pytest_sessionstart(session):
  52. plugin_base.post_begin()
  53. if has_xdist:
  54. import uuid
  55. def pytest_configure_node(node):
  56. # the master for each node fills slaveinput dictionary
  57. # which pytest-xdist will transfer to the subprocess
  58. plugin_base.memoize_important_follower_config(node.slaveinput)
  59. node.slaveinput["follower_ident"] = "test_%s" % uuid.uuid4().hex[0:12]
  60. from alembic.testing import provision
  61. provision.create_follower_db(node.slaveinput["follower_ident"])
  62. def pytest_testnodedown(node, error):
  63. from alembic.testing import provision
  64. provision.drop_follower_db(node.slaveinput["follower_ident"])
  65. def pytest_collection_modifyitems(session, config, items):
  66. # look for all those classes that specify __backend__ and
  67. # expand them out into per-database test cases.
  68. # this is much easier to do within pytest_pycollect_makeitem, however
  69. # pytest is iterating through cls.__dict__ as makeitem is
  70. # called which causes a "dictionary changed size" error on py3k.
  71. # I'd submit a pullreq for them to turn it into a list first, but
  72. # it's to suit the rather odd use case here which is that we are adding
  73. # new classes to a module on the fly.
  74. rebuilt_items = collections.defaultdict(list)
  75. items[:] = [
  76. item for item in
  77. items if isinstance(item.parent, pytest.Instance)]
  78. test_classes = set(item.parent for item in items)
  79. for test_class in test_classes:
  80. for sub_cls in plugin_base.generate_sub_tests(
  81. test_class.cls, test_class.parent.module):
  82. if sub_cls is not test_class.cls:
  83. list_ = rebuilt_items[test_class.cls]
  84. for inst in pytest.Class(
  85. sub_cls.__name__,
  86. parent=test_class.parent.parent).collect():
  87. list_.extend(inst.collect())
  88. newitems = []
  89. for item in items:
  90. if item.parent.cls in rebuilt_items:
  91. newitems.extend(rebuilt_items[item.parent.cls])
  92. rebuilt_items[item.parent.cls][:] = []
  93. else:
  94. newitems.append(item)
  95. # seems like the functions attached to a test class aren't sorted already?
  96. # is that true and why's that? (when using unittest, they're sorted)
  97. items[:] = sorted(newitems, key=lambda item: (
  98. item.parent.parent.parent.name,
  99. item.parent.parent.name,
  100. item.name
  101. ))
  102. def pytest_pycollect_makeitem(collector, name, obj):
  103. if inspect.isclass(obj) and plugin_base.want_class(obj):
  104. return pytest.Class(name, parent=collector)
  105. elif inspect.isfunction(obj) and \
  106. isinstance(collector, pytest.Instance) and \
  107. plugin_base.want_method(collector.cls, obj):
  108. return pytest.Function(name, parent=collector)
  109. else:
  110. return []
  111. _current_class = None
  112. def pytest_runtest_setup(item):
  113. # here we seem to get called only based on what we collected
  114. # in pytest_collection_modifyitems. So to do class-based stuff
  115. # we have to tear that out.
  116. global _current_class
  117. if not isinstance(item, pytest.Function):
  118. return
  119. # ... so we're doing a little dance here to figure it out...
  120. if _current_class is None:
  121. class_setup(item.parent.parent)
  122. _current_class = item.parent.parent
  123. # this is needed for the class-level, to ensure that the
  124. # teardown runs after the class is completed with its own
  125. # class-level teardown...
  126. def finalize():
  127. global _current_class
  128. class_teardown(item.parent.parent)
  129. _current_class = None
  130. item.parent.parent.addfinalizer(finalize)
  131. test_setup(item)
  132. def pytest_runtest_teardown(item):
  133. # ...but this works better as the hook here rather than
  134. # using a finalizer, as the finalizer seems to get in the way
  135. # of the test reporting failures correctly (you get a bunch of
  136. # py.test assertion stuff instead)
  137. test_teardown(item)
  138. def test_setup(item):
  139. plugin_base.before_test(item, item.parent.module.__name__,
  140. item.parent.cls, item.name)
  141. def test_teardown(item):
  142. plugin_base.after_test(item)
  143. def class_setup(item):
  144. plugin_base.start_test_class(item.cls)
  145. def class_teardown(item):
  146. plugin_base.stop_test_class(item.cls)