Reputation: 3725
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.
Upvotes: 0
Views: 396
Reputation: 1847
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():
Upvotes: 1
Reputation: 77912
In your task function, refresh_user_library
is a local name, not an attribute of the task. What you want is the real qualified name of the function you want to mock:
@mock.patch("core.tasks.refresh_user_library")
def test_refresh_library():
# you test here
Upvotes: 0