python: mock doesn't work inside celery task [duplicate]

@mock.patch('django.core.mail.mail_managers')

@mock.patch('path.to.tasks.mail_managers')

# views.py
from path.to.tasks
import my_task

def my_view(requtest):
   # do stuff
   my_task.delay('foo', 'bar')
return HttpResponse('whatever')

# test_my_task.py
from views
import my_view
from path.to.tasks
import my_task

class MyTest(TestCase):
   @mock.patch('path.to.tasks.my_task')
def test_my_task_is_called(self, mocked_task):
   client = Client()
client.get('/')
my_task.assert_called_with('foo', 'bar')

def test_my_task_works(self):
   my_task('foo', 'bar') # note I don 't use .delay(...), .apply_async(...), etc
assert 'my task did what I expected it to do'

from airflow.utils.log.timezone_aware
import TimezoneAware

# before
class YourCustomFormatter(logging.Formatter):
   ...

   # after
class YourCustomFormatter(TimezoneAware):
   ...

   AIRFLOW_FORMATTER = LOGGING_CONFIG["formatters"]["airflow"]
AIRFLOW_FORMATTER["class"] = "somewhere.your.custom_config.YourCustomFormatter"
# or use TimezoneAware class directly.If you don 't have custom Formatter.
AIRFLOW_FORMATTER["class"] = "airflow.utils.log.timezone_aware.TimezoneAware"

@dag.task(params = {
   "a": {
      1,
      2,
      3
   },
   "b": pendulum.now()
})
def datetime_param(value):
   print(value)

datetime_param("{{ params.a }} | {{ params.b }}")

> raise exc.NoSuchModuleError(
   "Can't load plugin: %s:%s" % (self.group, name)
)
E sqlalchemy.exc.NoSuchModuleError: Can 't load plugin: sqlalchemy.dialects:postgres

INSERT INTO log_template(id, filename, elasticsearch_id, created_at) VALUES(0, '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log', '{dag_id}-{task_id}-{execution_date}-{try_number}', NOW());

airflow connections
export -2 > & 1 > /dev/null | grep 'non-JSON'

Param(None, type = ["null", "string"])

from task
import some_task

def some_helper():
   some_task.delay(123)

@app.task(queue = "abc", bind = True)
def some_task(self, some_number: int):
   print(self.id) # how to mock this attribute access ?

from django.test.testcases
import TestCase
from helpers
import some_helper

class SomeTest(TestCase):

   def test_some_helper(self):
   some_helper()

class MockResult(dict):
   def __getattr__(self, x):
   return self[x]

      ...
      def test_some_task(self):
      cls = MockResult({
         "id": "asdf"
      })
bound_some_task = some_task.__get__(cls, MockResult)
bound_some_task(123)

@my_celery_app.task(bind = True)
def my_task(self):
   if self.request.retries == 1:
   my_method_to_invoke()
# Do work
for first retry
elif self.request.retries == 2:
   # Do work
for second retry
# do work
for
main task

@patch("path.to.task.my_method_to_invoke")
@patch("celery.app.task.Task.request")
def my_test_method(self, mock_task_request, mock_my_method_to_invoke):

   # Set the value of retries directly
mock_task_request.retries = 1

# Call the task and assert the inside method was
# called
my_task()

mock_my_method_to_invoke.assert_called_once()

from django.test.testcases
import TestCase
from helpers
import some_helper

class SomeTest(TestCase):

   def test_some_helper(self):

   from task
import some_task
setattr(some_task, 'id', 'hello-id')

some_helper()

@patch("task.some_task.request.id", return_value = "hello-id")
def test_some_helper(...): ....

class CeleryTaskHelper:
   ""
"Mock task id"
""

class Request:
   @property
def id(self):
   return ''.join(random.choices(string.ascii_uppercase + string.digits, k = 20))

@property
def request(self):
   return self.Request()

def revoke(self):
   return

@patch('apps.orders.tasks.activate_order.apply_async', return_value = CeleryTaskHelper())

task = some_task.s(<args>).apply()

@app.task
def some_task(self, ...args):
   some_method(self.request.id, ...args)