Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1"""Fixtures and testing utilities for :pypi:`py.test <pytest>`.""" 

2from __future__ import absolute_import, unicode_literals 

3 

4import os 

5from contextlib import contextmanager 

6 

7import pytest 

8 

9from .testing import worker 

10from .testing.app import TestApp, setup_default_app 

11 

12NO_WORKER = os.environ.get('NO_WORKER') 

13 

14# pylint: disable=redefined-outer-name 

15# Well, they're called fixtures.... 

16 

17 

18def pytest_configure(config): 

19 """Register additional pytest configuration.""" 

20 # add the pytest.mark.celery() marker registration to the pytest.ini [markers] section 

21 # this prevents pytest 4.5 and newer from issueing a warning about an unknown marker 

22 # and shows helpful marker documentation when running pytest --markers. 

23 config.addinivalue_line( 

24 "markers", "celery(**overrides): override celery configuration for a test case" 

25 ) 

26 

27 

28@contextmanager 

29def _create_app(enable_logging=False, 

30 use_trap=False, 

31 parameters=None, 

32 **config): 

33 # type: (Any, Any, Any, **Any) -> Celery 

34 """Utility context used to setup Celery app for pytest fixtures.""" 

35 parameters = {} if not parameters else parameters 

36 test_app = TestApp( 

37 set_as_current=False, 

38 enable_logging=enable_logging, 

39 config=config, 

40 **parameters 

41 ) 

42 with setup_default_app(test_app, use_trap=use_trap): 

43 yield test_app 

44 

45 

46@pytest.fixture(scope='session') 

47def use_celery_app_trap(): 

48 # type: () -> bool 

49 """You can override this fixture to enable the app trap. 

50 

51 The app trap raises an exception whenever something attempts 

52 to use the current or default apps. 

53 """ 

54 return False 

55 

56 

57@pytest.fixture(scope='session') 

58def celery_session_app(request, 

59 celery_config, 

60 celery_parameters, 

61 celery_enable_logging, 

62 use_celery_app_trap): 

63 # type: (Any, Any, Any, Any, Any) -> Celery 

64 """Session Fixture: Return app for session fixtures.""" 

65 mark = request.node.get_closest_marker('celery') 

66 config = dict(celery_config, **mark.kwargs if mark else {}) 

67 with _create_app(enable_logging=celery_enable_logging, 

68 use_trap=use_celery_app_trap, 

69 parameters=celery_parameters, 

70 **config) as app: 

71 if not use_celery_app_trap: 

72 app.set_default() 

73 app.set_current() 

74 yield app 

75 

76 

77@pytest.fixture(scope='session') 

78def celery_session_worker(request, 

79 celery_session_app, 

80 celery_includes, 

81 celery_worker_pool, 

82 celery_worker_parameters): 

83 # type: (Any, Celery, Sequence[str], str, Any) -> WorkController 

84 """Session Fixture: Start worker that lives throughout test suite.""" 

85 if not NO_WORKER: 

86 for module in celery_includes: 

87 celery_session_app.loader.import_task_module(module) 

88 with worker.start_worker(celery_session_app, 

89 pool=celery_worker_pool, 

90 **celery_worker_parameters) as w: 

91 yield w 

92 

93 

94@pytest.fixture(scope='session') 

95def celery_enable_logging(): 

96 # type: () -> bool 

97 """You can override this fixture to enable logging.""" 

98 return False 

99 

100 

101@pytest.fixture(scope='session') 

102def celery_includes(): 

103 # type: () -> Sequence[str] 

104 """You can override this include modules when a worker start. 

105 

106 You can have this return a list of module names to import, 

107 these can be task modules, modules registering signals, and so on. 

108 """ 

109 return () 

110 

111 

112@pytest.fixture(scope='session') 

113def celery_worker_pool(): 

114 # type: () -> Union[str, Any] 

115 """You can override this fixture to set the worker pool. 

116 

117 The "solo" pool is used by default, but you can set this to 

118 return e.g. "prefork". 

119 """ 

120 return 'solo' 

121 

122 

123@pytest.fixture(scope='session') 

124def celery_config(): 

125 # type: () -> Mapping[str, Any] 

126 """Redefine this fixture to configure the test Celery app. 

127 

128 The config returned by your fixture will then be used 

129 to configure the :func:`celery_app` fixture. 

130 """ 

131 return {} 

132 

133 

134@pytest.fixture(scope='session') 

135def celery_parameters(): 

136 # type: () -> Mapping[str, Any] 

137 """Redefine this fixture to change the init parameters of test Celery app. 

138 

139 The dict returned by your fixture will then be used 

140 as parameters when instantiating :class:`~celery.Celery`. 

141 """ 

142 return {} 

143 

144 

145@pytest.fixture(scope='session') 

146def celery_worker_parameters(): 

147 # type: () -> Mapping[str, Any] 

148 """Redefine this fixture to change the init parameters of Celery workers. 

149 

150 This can be used e. g. to define queues the worker will consume tasks from. 

151 

152 The dict returned by your fixture will then be used 

153 as parameters when instantiating :class:`~celery.worker.WorkController`. 

154 """ 

155 return {} 

156 

157 

158@pytest.fixture() 

159def celery_app(request, 

160 celery_config, 

161 celery_parameters, 

162 celery_enable_logging, 

163 use_celery_app_trap): 

164 """Fixture creating a Celery application instance.""" 

165 mark = request.node.get_closest_marker('celery') 

166 config = dict(celery_config, **mark.kwargs if mark else {}) 

167 with _create_app(enable_logging=celery_enable_logging, 

168 use_trap=use_celery_app_trap, 

169 parameters=celery_parameters, 

170 **config) as app: 

171 yield app 

172 

173 

174@pytest.fixture() 

175def celery_worker(request, 

176 celery_app, 

177 celery_includes, 

178 celery_worker_pool, 

179 celery_worker_parameters): 

180 # type: (Any, Celery, Sequence[str], str, Any) -> WorkController 

181 """Fixture: Start worker in a thread, stop it when the test returns.""" 

182 if not NO_WORKER: 

183 for module in celery_includes: 

184 celery_app.loader.import_task_module(module) 

185 with worker.start_worker(celery_app, 

186 pool=celery_worker_pool, 

187 **celery_worker_parameters) as w: 

188 yield w 

189 

190 

191@pytest.fixture() 

192def depends_on_current_app(celery_app): 

193 """Fixture that sets app as current.""" 

194 celery_app.set_current()