Due to circular-import issues which are common with Celery tasks in Django, I'm often importing Celery tasks inside of my methods, like so:
# some code omitted for brevity
# accounts/models.py
def refresh_library(self, queue_type="regular"):
from core.tasks import refresh_user_library
refresh_user_library.apply_async(
kwargs={"user_id": self.user.id}, queue=queue_type
)
return 0
In my pytest
test for refresh_library
, I'd only like to test that refresh_user_library
(the Celery task) is called with the correct args
and kwargs
. But this isn't working:
# tests/test_accounts_models.py
@mock.patch("accounts.models.UserProfile.refresh_library.refresh_user_library")
def test_refresh_library():
Error is about refresh_library
not having an attribute refresh_user_library
.
I suspect this is due to the fact that the task(refresh_user_library
) is imported inside the function itself, but I'm not too experienced with mocking so this might be completely wrong.
Even though apply_async
is your own-created function in your core.tasks
, if you do not want to test it but only make sure you are giving correct arguments, you need to mock it. In your question you're mocking wrong package. You should do:
# tests/test_accounts_models.py
@mock.patch("core.tasks.rehresh_user_library.apply_sync")
def test_refresh_library():