Why does my Elasticsearch client close the event loop after a few queries in PyTest?

I'm trying to run a few integration tests in PyTest, complete with a real containerized ElasticSearch instance. I have no problem doing local functional tests, but under PyTest, the ES client raises an "event loop closed" exception.

Current versions of the packages in question:

• pytest: 7.1.2
• pytest-asyncio: 0.15.1 (I also tried 0.19)
• elasticsearch[async]: 7.14.0

For some additional insights, I wrapped the AsyncElasticsearch class to get a peek into its innards:

class NoisyAsync(AsyncElasticsearch):
    def __init__(
        self,
        *args,
        **kwargs,
    ):
        super().__init__(
            *args,
            **kwargs,
        )

    def reveal_loop(func):
        @wraps(func)
        async def wrapper(self, *args, **kwargs):
            print({f"{func.__name__}": {"args": args, "kwargs": kwargs}}, "\n")
            print(self.transport.loop.__dict__.items(), "\n")
            if self.transport.loop._closed:
                print(f"\nCLOSED ({func.__name__})\n")
            retval = await func(self, *args, **kwargs)
            print(f" {retval}")
            return retval

        return wrapper

    #The failing test only calls the search() and count() methods

    @reveal_loop
    async def count(self, *args, **kwargs):
        return await super().count(*args, **kwargs)

    @reveal_loop
    async def search(self, *args, **kwargs):
        return await super().search(*args, **kwargs)


def create_async_es(): # Normally does not return a NoisyAsync, of course
    if es_tls:
        return NoisyAsync( # es_* values come from env vars
            f"https://{es_username}:{es_password}@{es_server}:{es_port}",
            use_ssl=True,
            verify_certs=False,
        )
    return NoisyAsync(f"http://{es_username}:{es_password}@{es_server}:{es_port}")


ES = create_async_es() # ES is returned as a dependency in another function...

Firing off a test reveals the following:

{'count': {'args': (), 'kwargs': {'index': 'blah blah... ' }}

dict_items([('_timer_cancelled_count', 2), ('_closed', False), ('_stopping', False), ('_ready', deque([])), 
..snip.. 
('_csock', <socket.socket fd=17, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>), ('_transports', <WeakValueDictionary at 0x7fc7a97a2040>), ('_signal_handlers', {})])

{'count': 2, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}}

{'search': {'args': (), 'kwargs': {'index': 'blah blah blah...' }}}

dict_items([('_timer_cancelled_count', 2), ('_closed', False), ('_stopping', False), ('_ready', deque([])), 
..snip..
('_csock', <socket.socket fd=17, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>), ('_transports', <WeakValueDictionary at 0x7fc7a97a2040>), ('_signal_handlers', {})])

 {'took': 10, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 2, 
..snip..
}

{'count': {'args': (), 'kwargs': {'index': 'blah blah blah...' }}}

dict_items([('_timer_cancelled_count', 1), ('_closed', True), ('_stopping', False), 
..snip..
 ('_csock', None), ('_transports', <WeakValueDictionary at 0x7fc7a97a2040>), ('_signal_handlers', {})])


CLOSED (count)

ConnectionError(Event loop is closed) caused by: RuntimeError(Event loop is closed)

The first call to ES.count() works. The second call to ES.search() works. The next call to ES.count() fails with the RuntimeError.

What could be causing the event loop, as tracked within the AsyncElasticsearch instance, to close? Again, this only happens whenever PyTest is invoked.