Closed
Description
Thanks for submitting an issue!
Here's a quick checklist in what to include:
- Include a detailed description of the bug or suggestion
I'm unable to view database actions in a Celery task from my pytest test function. I imagine it's related to transactions, but can't nail it down despite settingtransaction=True
in the pytest marker. -
pip list
of the virtual environment you are using
amqp (2.2.2)
apipkg (1.4)
attrs (17.4.0)
billiard (3.5.0.3)
celery (4.1.0)
certifi (2017.11.5)
chardet (3.0.4)
cookies (2.2.1)
coverage (4.4.2)
Django (2.0.1)
django-celery-beat (1.1.0)
django-celery-results (1.0.1)
django-fsm (2.6.0)
django-model-utils (3.1.1)
djangorestframework (3.7.7)
ephem (3.7.6.0)
execnet (1.5.0)
factory-boy (2.8.1)
Faker (0.8.8)
flake8 (3.5.0)
graphene (2.0.1)
graphene-django (2.0.0)
graphql-core (2.0)
graphql-relay (0.4.5)
idna (2.6)
inflection (0.3.1)
iso8601 (0.1.12)
kombu (4.1.0)
mccabe (0.6.1)
pep8 (1.7.1)
pip (9.0.1)
pluggy (0.6.0)
promise (2.1)
psycopg2 (2.7.3.2)
py (1.5.2)
pycodestyle (2.3.1)
pyflakes (1.6.0)
pytest (3.3.2)
pytest-cache (1.0)
pytest-cov (2.5.1)
pytest-django (3.1.2)
pytest-env (0.6.2)
pytest-factoryboy (1.3.1)
pytest-flake8 (0.9.1)
pytest-mock (1.6.3)
pytest-pep8 (1.0.6)
python-dateutil (2.6.1)
pytz (2017.3)
redis (2.10.6)
requests (2.18.4)
responses (0.8.1)
Rx (1.6.0)
setuptools (38.4.0)
singledispatch (3.4.0.3)
six (1.11.0)
slackclient (1.1.0)
text-unidecode (1.1)
typing (3.6.2)
urllib3 (1.22)
vine (1.1.4)
websocket-client (0.46.0)
wheel (0.30.0)
- pytest and operating system versions
- pytest: 3.3.2, OS: Mac OS X
- Minimal example if possible
Sample test
class TestClosingPendingClosedSession:
@pytest.mark.django_db()
def test_does_get_closed_with_no_messages(self, settings, session_factory):
settings.CELERY_TASK_ALWAYS_EAGER = True
settings.CELERY_TASK_EAGER_PROPAGATES = True
original_time = datetime.now(pytz.UTC) - timedelta(minutes=31)
session = session_factory(time_start=original_time)
session.mark_as_stale()
session.save()
session.mark_as_pending_closed()
session.save()
# wait 5 seconds
wait_until(condition=lambda: Session.objects.get(pk=session.id).is_closed,
timeout=5)
assert Session.objects.get(pk=session.id).is_closed
Sample Celery task
@celery_app.task
def close_pending_closed_session(session_id, datetime_of_last_non_bot_message):
"""
Task that closes session that's been set to PENDING CLOSED
if there is no new activity for 5 additional minutes.
"""
session = Session.objects.get(pk=session_id)
if datetime_of_last_non_bot_message == session.datetime_of_last_non_bot_message:
session.mark_as_closed()
session.save()
print(session.is_closed)
session.question.mark_as_solved()
session.question.save()
This is printing True
despite the assertion failing.