pytest fixture celery_worker cause connection already closed error

[Headmaster11]

Very similar to this #824

Here is project example https://github.com/Headmaster11/celery_pytest

When try to launch pytest with celery_worker fixture, got such error

self = <django.db.backends.postgresql.base.DatabaseWrapper object at 0x7ff2aaaa50a0>
name = None

    def _cursor(self, name=None):
        self.ensure_connection()
        with self.wrap_database_errors:
>           return self._prepare_cursor(self.create_cursor(name))

../venv/lib/python3.9/site-packages/django/db/backends/base/base.py:262: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<django.db.backends.postgresql.base.DatabaseWrapper object at 0x7ff2aaaa50a0>, None)
kwargs = {}

    @functools.wraps(func)
    def inner(*args, **kwargs):
        if not os.environ.get("DJANGO_ALLOW_ASYNC_UNSAFE"):
            # Detect a running event loop in this thread.
            try:
                asyncio.get_running_loop()
            except RuntimeError:
                pass
            else:
                raise SynchronousOnlyOperation(message)
        # Pass onward.
>       return func(*args, **kwargs)

../venv/lib/python3.9/site-packages/django/utils/asyncio.py:26: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.backends.postgresql.base.DatabaseWrapper object at 0x7ff2aaaa50a0>
name = None

    @async_unsafe
    def create_cursor(self, name=None):
        if name:
            # In autocommit mode, the cursor will be used outside of a
            # transaction, hence use a holdable cursor.
            cursor = self.connection.cursor(
                name, scrollable=False, withhold=self.connection.autocommit
            )
        else:
>           cursor = self.connection.cursor()
E           psycopg2.InterfaceError: connection already closed

../venv/lib/python3.9/site-packages/django/db/backends/postgresql/base.py:256: InterfaceError

The above exception was the direct cause of the following exception:

self = <fixtures.test_main.TestItem object at 0x7ff2aaf0aeb0>
celery_worker = <Worker: gen52797@MacBook-Pro-Aliaksei.local (running)>

    def test_item(self, celery_worker):
>       ItemFactory.create_batch(10)

test_main.py:10: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
../venv/lib/python3.9/site-packages/factory/base.py:540: in create_batch
    return [cls.create(**kwargs) for _ in range(size)]
../venv/lib/python3.9/site-packages/factory/base.py:540: in <listcomp>
    return [cls.create(**kwargs) for _ in range(size)]
../venv/lib/python3.9/site-packages/factory/base.py:528: in create
    return cls._generate(enums.CREATE_STRATEGY, kwargs)
../venv/lib/python3.9/site-packages/factory/django.py:117: in _generate
    return super()._generate(strategy, params)
../venv/lib/python3.9/site-packages/factory/base.py:465: in _generate
    return step.build()
../venv/lib/python3.9/site-packages/factory/builder.py:262: in build
    instance = self.factory_meta.instantiate(
../venv/lib/python3.9/site-packages/factory/base.py:317: in instantiate
    return self.factory._create(model, *args, **kwargs)
../venv/lib/python3.9/site-packages/factory/django.py:166: in _create
    return manager.create(*args, **kwargs)
../venv/lib/python3.9/site-packages/django/db/models/manager.py:85: in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
../venv/lib/python3.9/site-packages/django/db/models/query.py:514: in create
    obj.save(force_insert=True, using=self.db)
../venv/lib/python3.9/site-packages/django/db/models/base.py:806: in save
    self.save_base(
../venv/lib/python3.9/site-packages/django/db/models/base.py:857: in save_base
    updated = self._save_table(
../venv/lib/python3.9/site-packages/django/db/models/base.py:1000: in _save_table
    results = self._do_insert(
../venv/lib/python3.9/site-packages/django/db/models/base.py:1041: in _do_insert
    return manager._insert(
../venv/lib/python3.9/site-packages/django/db/models/manager.py:85: in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
../venv/lib/python3.9/site-packages/django/db/models/query.py:1434: in _insert
    return query.get_compiler(using=using).execute_sql(returning_fields)
../venv/lib/python3.9/site-packages/django/db/models/sql/compiler.py:1619: in execute_sql
    with self.connection.cursor() as cursor:
../venv/lib/python3.9/site-packages/django/utils/asyncio.py:26: in inner
    return func(*args, **kwargs)
../venv/lib/python3.9/site-packages/django/db/backends/base/base.py:284: in cursor
    return self._cursor()
../venv/lib/python3.9/site-packages/django/db/backends/base/base.py:262: in _cursor
    return self._prepare_cursor(self.create_cursor(name))
../venv/lib/python3.9/site-packages/django/db/utils.py:91: in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
../venv/lib/python3.9/site-packages/django/db/backends/base/base.py:262: in _cursor
    return self._prepare_cursor(self.create_cursor(name))
../venv/lib/python3.9/site-packages/django/utils/asyncio.py:26: in inner
    return func(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <django.db.backends.postgresql.base.DatabaseWrapper object at 0x7ff2aaaa50a0>
name = None

    @async_unsafe
    def create_cursor(self, name=None):
        if name:
            # In autocommit mode, the cursor will be used outside of a
            # transaction, hence use a holdable cursor.
            cursor = self.connection.cursor(
                name, scrollable=False, withhold=self.connection.autocommit
            )
        else:
>           cursor = self.connection.cursor()
E           django.db.utils.InterfaceError: connection already closed

../venv/lib/python3.9/site-packages/django/db/backends/postgresql/base.py:256: InterfaceError

[g-kisenkov]

@Headmaster11

Try this: celery/celery#4511 (comment)
Worked for me.