Commit 453608e6 authored by Leo Pound Singer's avatar Leo Pound Singer
Browse files

Simpler and better-documented conftest.py

parent c88b5c35
......@@ -95,6 +95,29 @@ top directory of your local source checkout::
This will save a coverage report that you can view in a web browser as
``htmlcov/index.html``.
.. admonition:: Eager mode
Most of GWCelery's unit tests use :ref:`eager mode <celery:testing>`, which
causes all tasks to execute immediately and synchronously, even if they are
invoked via :meth:`~celery.app.task.Task.apply_async` or
:meth:`~celery.app.task.Task.delay`. This simplifies writing unit tests,
but sacrifices realism: it may mask concurrency bugs that may only occur
when the tasks are executed asynchronously.
It is preferable to write unit tests that use a live worker so that they
are subject to realistic, asynchronous task execution. To opt in to using a
live worker, simply decorate your test with the `live_worker` marker, like
this:
.. code-block:: python
@pytest.mark.live_worker
def test_some_task():
async_result = some_task.delay()
result = async_result.get()
assert result == 'foobar'
# etc.
Code style
----------
......
......@@ -9,104 +9,106 @@ from .. import app
from .process import starter # noqa: F401
def nuke_celery_backend():
"""Clear the cached Celery backend.
The Celery application object caches a lot of instance members that are
affected by the application configuration. Since we are changing the
configuration between tests, we need to make sure that all of the cached
application state is reset.
FIXME: The pytest celery plugin does not seem like it is really designed
to use a pre-existing application object; it seems like it is designed to
create a test application.
"""
app._pool = None
for key, value in app.__class__.__dict__.items():
if isinstance(value, cached_property):
try:
del app.__dict__[key]
except KeyError:
pass
app._local.__dict__.clear()
@pytest.fixture(autouse=True)
def fake_gracedb_client(monkeypatch):
mock_client = mock.MagicMock()
mock_client.url = 'https://gracedb.invalid/api/'
monkeypatch.setattr('gwcelery.tasks.gracedb.client', mock_client)
@pytest.fixture
def reset_celery_backend():
"""Nuke the celery backend before and after the test."""
nuke_celery_backend()
yield
nuke_celery_backend()
@pytest.fixture(autouse=True)
def fake_legacy_gracedb_client(monkeypatch):
mock_client = mock.MagicMock()
mock_client.url = 'https://gracedb.invalid/api/'
monkeypatch.setattr('gwcelery.tasks.legacy_gracedb.client', mock_client)
@pytest.fixture
def update_celery_config():
"""Monkey patch the Celery application configuration."""
tmp = {}
@pytest.fixture(autouse=True)
def no_sockets():
disable_socket()
def update(new_conf):
tmp.update({key: app.conf[key] for key in new_conf.keys()})
app.conf.update(new_conf)
yield update
app.conf.update(tmp)
#
# The following methods override `fixtures provided by the Celery pytest plugin
# <https://docs.celeryproject.org/en/stable/userguide/testing.html#fixtures>`_.
#
@pytest.fixture
def noop_celery_config(reset_celery_backend, update_celery_config):
"""Ensure that the Celery app is disconnected from live services."""
update_celery_config(dict(
broker_url='redis://redis.invalid',
result_backend='redis://redis.invalid',
def celery_config(request):
"""Prepare Celery application configuration for unit tests."""
# If this unit test does not have the `@pytest.mark.live_worker` mark,
# then turn on eager mode.
eager = not request.node.get_closest_marker('live_worker')
return dict(
broker_url='memory://',
result_backend='cache+memory://',
worker_hijack_root_logger=False,
task_always_eager=eager,
task_eager_propagates=eager,
voevent_broadcaster_address='127.0.0.1:53410',
voevent_broadcaster_whitelist=['127.0.0.0/8'],
voevent_receiver_address='gcn.invalid:8099',
task_always_eager=True,
task_eager_propagates=True,
lvalert_host='lvalert.invalid',
gracedb_host='gracedb.invalid',
expose_to_public=True
))
@pytest.fixture
def celery_config():
"""Celery application configuration for tests that need a real worker."""
return dict(
broker_url='memory://',
result_backend='cache+memory://',
task_always_eager=False,
task_eager_propagates=False,
)
@pytest.fixture
def celery_worker_parameters():
"""Prepare Celery worker configuration for unit tests."""
# Disable the ping check on worker startup. The `ping` task is registered
# on the Celery pytest plugin's default test app, but not on our app.
return dict(perform_ping_check=False)
@pytest.fixture
def celery_app(celery_config, celery_enable_logging, reset_celery_backend,
update_celery_config, monkeypatch):
update_celery_config(celery_config)
@pytest.fixture(autouse=True)
def celery_app(celery_config, celery_enable_logging, monkeypatch):
"""Prepare Celery application for unit tests.
The original fixture returns a specially-created test application. This
version substitutes our own (gwcelery's) application.
"""
# Update the Celery application configuration.
for key, value in celery_config.items():
monkeypatch.setitem(app.conf, key, value)
# Configure logging, if requested.
if not celery_enable_logging:
monkeypatch.setattr(app, 'log', UnitLogging(app))
yield app
# Reset all of the cached Celery application properties.
#
# The Celery application object caches a lot of instance members that are
# affected by the application configuration. Since we are changing the
# configuration between tests, we need to make sure that all of the cached
# application state is reset.
#
# First, reset all of the @cached_property members...
for key, value in app.__class__.__dict__.items():
if isinstance(value, cached_property):
try:
del app.__dict__[key]
except KeyError:
pass
# Then, reset the thread-local storage.
app._local.__dict__.clear()
@pytest.fixture(autouse=True)
def fake_gracedb_client(monkeypatch):
mock_client = mock.MagicMock()
mock_client.url = 'https://gracedb.invalid/api/'
monkeypatch.setattr('gwcelery.tasks.gracedb.client', mock_client)
# Now, allow the unit test to run.
yield app
# Finally, reset the worker pool.
app.close()
@pytest.fixture(autouse=True)
def fake_legacy_gracedb_client(monkeypatch):
mock_client = mock.MagicMock()
mock_client.url = 'https://gracedb.invalid/api/'
monkeypatch.setattr('gwcelery.tasks.legacy_gracedb.client', mock_client)
#
# The following `pytest hooks
# <https://docs.pytest.org/en/latest/how-to/writing_hook_functions.html>`_ and
# fixtures implement the `@pytest.mark.live_worker` decorator that indicates
# unit tests that use a live Celery worker (as opposed to eager mode).
#
def pytest_configure(config):
......@@ -124,14 +126,7 @@ def pytest_collection_modifyitems(session, config, items):
key=lambda item: item.get_closest_marker('live_worker') is not None)
def pytest_runtest_setup(item):
disable_socket()
@pytest.fixture(autouse=True)
def maybe_celery_worker(request):
if request.node.get_closest_marker('live_worker') is None:
fixture = 'noop_celery_config'
else:
fixture = 'celery_worker'
request.getfixturevalue(fixture)
if request.node.get_closest_marker('live_worker'):
request.getfixturevalue('celery_worker')
......@@ -21,12 +21,23 @@ def test_nagios_unknown_error(monkeypatch, capsys):
assert 'UNKNOWN: Unexpected error' in out
def test_nagios(capsys, monkeypatch, socket_enabled, starter, tmp_path):
@pytest.fixture
def celery_worker_parameters():
return dict(
perform_ping_check=False,
queues=['celery', 'exttrig', 'openmp', 'superevent', 'voevent']
)
def test_nagios(capsys, monkeypatch, request, socket_enabled, starter,
tmp_path):
mock_lvalert_client = Mock()
monkeypatch.setattr(
'gwcelery.lvalert.client.LVAlertClient', mock_lvalert_client)
unix_socket = str(tmp_path / 'redis.sock')
app.conf['broker_url'] = f'redis+socket://{unix_socket}'
broker_url = f'redis+socket://{unix_socket}'
monkeypatch.setitem(app.conf, 'broker_url', broker_url)
monkeypatch.setitem(app.conf, 'result_backend', broker_url)
# no broker
......@@ -54,10 +65,7 @@ def test_nagios(capsys, monkeypatch, socket_enabled, starter, tmp_path):
# worker, no LVAlert nodes
starter.python_process(
args=(['gwcelery', 'worker', '-l', 'info', '--pool', 'solo',
'-Q', 'celery,exttrig,openmp,superevent,voevent'],),
target=main, timeout=10, magic_words=b'ready.')
request.getfixturevalue('celery_worker')
mock_lvalert_client.configure_mock(**{
'return_value.get_subscriptions.return_value': {}})
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment