connector-ci: Use Dagger in CAT

[alafanechere]

Context

Our Connector Acceptance Tests (CAT) project is a python program running tests on our connector docker images. These tests execute commands on the container and perform assertions on the output. We containerize CAT and ran it so far with airbyte-ci using a docker-in-docker pattern:

1. airbyte-ci builds the connector container using Dagger.
2. The image is exported as a tarball file and loaded to a global dockerd service.
3. The CAT container is bound to the global dockerd service so it has access to the build container image.
4. We run CAT from airbyte-ci

Problems

We'd love to not rely on the docker-in-docker pattern for the following reasons:

• Using a global dockerd service makes dockerd a single point of failure of our pipelines
• Exporting built containers to tarballs and loading them to dockerd adds a lot of overhead to our pipelines and makes them slower
• A failure of the global docker daemon leads to cascading CAT run failure in nightly builds
• To allow CAT to mount connector config to the connector under test we have to use an insecure_root_capabilities flag, which is of course not recommended.

Solution

To get rid of the docker-in-docker pattern, we'd love to make CAT use Dagger. Docker interactions with containers in CAT are located in a single class: ConnectorRunner.

Implementation

In CAT I refactored ConnectorRunner class to use Dagger instead of docker commands. In airbyte-ci I refactored our with_connector_acceptance_test function to build the CAT container and mount the connector image tarball to it.

How airbyte-ci shares the connector under test image / container with CAT:

We share the build connector image from airbyte-ci by mounting the tarball file to the CAT container and use dagger_client.container().import_(<tarball_file>) to create the connnector container.

I originally tried to share the connector container by sharing the ContainerID to CAT and instantiate the container in CAT with dagger_client.container(dagger.ContainerId(<container_id>)) but it didn't work I occassionally face error like failed to load cache key: unable to get info about digest: NotFound: rpc error: code = NotFound desc = content sha256:7e9d2045e9891e8eef3c917af3a5246a292c2b2c8c7f8c0d0b76a357d5439ea8: .

Current blockers

It's not working for heavy images

The current implementation works for lightweight images (~50mb). But when I try to run CAT on a heavy images (~500mb) the test execution hangs... My hypothesis is that the import_ is a costly operation, and as we instantiate a new Dagger connection for each exec we want on to run on a container, we pay the cost of the import for each exec. We instantiate a new Dagger connection for each exec because the public methods of ConnectorRunner function are synchronous, so we run anyio.run inside them. I'd prefer to not change the public methods of ConnectorRunner to be async as it would require to change the whole CAT codebase to be async.

[bnchrch]

Well laid out @alafanechere 👏

Some great benefits here are

1. Naturally pushes us to update CAT to 3.10
2. No interface changes so everything should continue to work smoothly
3. Dagger is well suited to coordinate docker based systems

So far im a tentative thumbs up!

One question, Is there any other known challenges to "dagger in dagger"?

[evantahler]

+1 from me too! The benefits are real, and the more airbyte code we can standardize, the better.

I'd suggest time boxing this. If there are dagger-in-dagger problems, let's only spend a few days on this, as it is nice-to-have, not required

[alafanechere]

Thank you for your inputs @bnchrch and @evantahler! I asked a couple of month ago to the dagger team if dagger in dagger is tricky, they told it was not because they follow this paradigm for their tests. I think it's indeed worth timeboxing / spiking this: not more than 1.5 day.

[alafanechere]

Updates

The hypothesis we came up with the Dagger team was the following: The import_ operation is not cached throughout sessions, as a new Dagger session is instantiated we pay the cost of loading the tarball to a container on each ConnectorRunner method call.

I refactored the CAT code to use a single Dagger sessions. But the "hanging test" error remain and finish with timeout issue:

• Example workflow testing source-postgres and source-openweather
• Dagger client errors associated with the workflow above:

  778: > in Test Pipeline - source-postgres
  778: [1253.2s] ============================= test session starts ==============================
  778: [1253.2s] platform linux -- Python 3.10.12, pytest-6.2.5, py-1.11.0, pluggy-1.2.0
  778: [1253.2s] rootdir: /test_input
  778: [1253.2s] plugins: hypothesis-6.82.0, anyio-3.7.1, requests-mock-1.9.3, timeout-1.4.2, sugar-0.9.7, mock-3.6.1, custom-exit-code-0.3.0, cov-3.0.0
  778: [1253.2s] collected 76 items / 1 skipped / 75 selected
  778: [1253.2s] 
  778: [1253.2s] test_core.py ........................................ss................. [ 77%]
  778: [1253.2s] .......ssssFFFF                                                          [ 97%]
  778: [1253.2s] test_full_refresh.py FFE                                                 [100%]
  778: [1253.2s] 
  778: [1253.2s] 
  778: [1253.2s] ==================================== ERRORS ====================================
  778: [1253.2s] _____ ERROR at teardown of TestFullRefresh.test_sequential_reads[inputs1] ______
  778: [1253.2s] 
  778: [1253.2s] map = {<class 'TimeoutError'>: <class 'httpcore.ReadTimeout'>, <class 'anyio.BrokenResourceError'>: <class 'httpcore.ReadError'>, <class 'anyio.ClosedResourceError'>: <class 'httpcore.ReadError'>}
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def map_exceptions(map: ExceptionMapping) -> Iterator[None]:
  778: [1253.2s]         try:
  778: [1253.2s] >           yield
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py:10: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpcore._backends.anyio.AnyIOStream object at 0x7f958496a470>
  778: [1253.2s] max_bytes = 65536, timeout = None
  778: [1253.2s] 
  778: [1253.2s]     async def read(
  778: [1253.2s]         self, max_bytes: int, timeout: typing.Optional[float] = None
  778: [1253.2s]     ) -> bytes:
  778: [1253.2s]         exc_map = {
  778: [1253.2s]             TimeoutError: ReadTimeout,
  778: [1253.2s]             anyio.BrokenResourceError: ReadError,
  778: [1253.2s]             anyio.ClosedResourceError: ReadError,
  778: [1253.2s]         }
  778: [1253.2s]         with map_exceptions(exc_map):
  778: [1253.2s]             with anyio.fail_after(timeout):
  778: [1253.2s]                 try:
  778: [1253.2s] >                   return await self._stream.receive(max_bytes=max_bytes)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_backends/anyio.py:34: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <anyio._backends._asyncio.SocketStream object at 0x7f9584968400>
  778: [1253.2s] max_bytes = 65536
  778: [1253.2s] 
  778: [1253.2s]     async def receive(self, max_bytes: int = 65536) -> bytes:
  778: [1253.2s]         with self._receive_guard:
  778: [1253.2s]             await checkpoint()
  778: [1253.2s]     
  778: [1253.2s]             if (
  778: [1253.2s]                 not self._protocol.read_event.is_set()
  778: [1253.2s]                 and not self._transport.is_closing()
  778: [1253.2s]             ):
  778: [1253.2s]                 self._transport.resume_reading()
  778: [1253.2s]                 await self._protocol.read_event.wait()
  778: [1253.2s]                 self._transport.pause_reading()
  778: [1253.2s]     
  778: [1253.2s]             try:
  778: [1253.2s]                 chunk = self._protocol.read_queue.popleft()
  778: [1253.2s]             except IndexError:
  778: [1253.2s]                 if self._closed:
  778: [1253.2s] >                   raise ClosedResourceError from None
  778: [1253.2s] E                   anyio.ClosedResourceError
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:1210: ClosedResourceError
  778: [1253.2s] 
  778: [1253.2s] The above exception was the direct cause of the following exception:
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def map_httpcore_exceptions() -> typing.Iterator[None]:
  778: [1253.2s]         try:
  778: [1253.2s] >           yield
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_transports/default.py:60: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncHTTPTransport object at 0x7f958496ba00>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(
  778: [1253.2s]         self,
  778: [1253.2s]         request: Request,
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         assert isinstance(request.stream, AsyncByteStream)
  778: [1253.2s]     
  778: [1253.2s]         req = httpcore.Request(
  778: [1253.2s]             method=request.method,
  778: [1253.2s]             url=httpcore.URL(
  778: [1253.2s]                 scheme=request.url.raw_scheme,
  778: [1253.2s]                 host=request.url.raw_host,
  778: [1253.2s]                 port=request.url.port,
  778: [1253.2s]                 target=request.url.raw_path,
  778: [1253.2s]             ),
  778: [1253.2s]             headers=request.headers.raw,
  778: [1253.2s]             content=request.stream,
  778: [1253.2s]             extensions=request.extensions,
  778: [1253.2s]         )
  778: [1253.2s]         with map_httpcore_exceptions():
  778: [1253.2s] >           resp = await self._pool.handle_async_request(req)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_transports/default.py:353: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpcore.AsyncConnectionPool object at 0x7f958496a200>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(self, request: Request) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Send an HTTP request, and return an HTTP response.
  778: [1253.2s]     
  778: [1253.2s]         This is the core implementation that is called into by `.request()` or `.stream()`.
  778: [1253.2s]         """
  778: [1253.2s]         scheme = request.url.scheme.decode()
  778: [1253.2s]         if scheme == "":
  778: [1253.2s]             raise UnsupportedProtocol(
  778: [1253.2s]                 "Request URL is missing an 'http://' or 'https://' protocol."
  778: [1253.2s]             )
  778: [1253.2s]         if scheme not in ("http", "https", "ws", "wss"):
  778: [1253.2s]             raise UnsupportedProtocol(
  778: [1253.2s]                 f"Request URL has an unsupported protocol '{scheme}://'."
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         status = RequestStatus(request)
  778: [1253.2s]     
  778: [1253.2s]         async with self._pool_lock:
  778: [1253.2s]             self._requests.append(status)
  778: [1253.2s]             await self._close_expired_connections()
  778: [1253.2s]             await self._attempt_to_acquire_connection(status)
  778: [1253.2s]     
  778: [1253.2s]         while True:
  778: [1253.2s]             timeouts = request.extensions.get("timeout", {})
  778: [1253.2s]             timeout = timeouts.get("pool", None)
  778: [1253.2s]             try:
  778: [1253.2s]                 connection = await status.wait_for_connection(timeout=timeout)
  778: [1253.2s]             except BaseException as exc:
  778: [1253.2s]                 # If we timeout here, or if the task is cancelled, then make
  778: [1253.2s]                 # sure to remove the request from the queue before bubbling
  778: [1253.2s]                 # up the exception.
  778: [1253.2s]                 async with self._pool_lock:
  778: [1253.2s]                     # Ensure only remove when task exists.
  778: [1253.2s]                     if status in self._requests:
  778: [1253.2s]                         self._requests.remove(status)
  778: [1253.2s]                     raise exc
  778: [1253.2s]     
  778: [1253.2s]             try:
  778: [1253.2s]                 response = await connection.handle_async_request(request)
  778: [1253.2s]             except ConnectionNotAvailable:
  778: [1253.2s]                 # The ConnectionNotAvailable exception is a special case, that
  778: [1253.2s]                 # indicates we need to retry the request on a new connection.
  778: [1253.2s]                 #
  778: [1253.2s]                 # The most common case where this can occur is when multiple
  778: [1253.2s]                 # requests are queued waiting for a single connection, which
  778: [1253.2s]                 # might end up as an HTTP/2 connection, but which actually ends
  778: [1253.2s]                 # up as HTTP/1.1.
  778: [1253.2s]                 async with self._pool_lock:
  778: [1253.2s]                     # Maintain our position in the request queue, but reset the
  778: [1253.2s]                     # status so that the request becomes queued again.
  778: [1253.2s]                     status.unset_connection()
  778: [1253.2s]                     await self._attempt_to_acquire_connection(status)
  778: [1253.2s]             except BaseException as exc:
  778: [1253.2s]                 with AsyncShieldCancellation():
  778: [1253.2s]                     await self.response_closed(status)
  778: [1253.2s] >               raise exc
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:262: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpcore.AsyncConnectionPool object at 0x7f958496a200>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(self, request: Request) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Send an HTTP request, and return an HTTP response.
  778: [1253.2s]     
  778: [1253.2s]         This is the core implementation that is called into by `.request()` or `.stream()`.
  778: [1253.2s]         """
  778: [1253.2s]         scheme = request.url.scheme.decode()
  778: [1253.2s]         if scheme == "":
  778: [1253.2s]             raise UnsupportedProtocol(
  778: [1253.2s]                 "Request URL is missing an 'http://' or 'https://' protocol."
  778: [1253.2s]             )
  778: [1253.2s]         if scheme not in ("http", "https", "ws", "wss"):
  778: [1253.2s]             raise UnsupportedProtocol(
  778: [1253.2s]                 f"Request URL has an unsupported protocol '{scheme}://'."
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         status = RequestStatus(request)
  778: [1253.2s]     
  778: [1253.2s]         async with self._pool_lock:
  778: [1253.2s]             self._requests.append(status)
  778: [1253.2s]             await self._close_expired_connections()
  778: [1253.2s]             await self._attempt_to_acquire_connection(status)
  778: [1253.2s]     
  778: [1253.2s]         while True:
  778: [1253.2s]             timeouts = request.extensions.get("timeout", {})
  778: [1253.2s]             timeout = timeouts.get("pool", None)
  778: [1253.2s]             try:
  778: [1253.2s]                 connection = await status.wait_for_connection(timeout=timeout)
  778: [1253.2s]             except BaseException as exc:
  778: [1253.2s]                 # If we timeout here, or if the task is cancelled, then make
  778: [1253.2s]                 # sure to remove the request from the queue before bubbling
  778: [1253.2s]                 # up the exception.
  778: [1253.2s]                 async with self._pool_lock:
  778: [1253.2s]                     # Ensure only remove when task exists.
  778: [1253.2s]                     if status in self._requests:
  778: [1253.2s]                         self._requests.remove(status)
  778: [1253.2s]                     raise exc
  778: [1253.2s]     
  778: [1253.2s]             try:
  778: [1253.2s] >               response = await connection.handle_async_request(request)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:245: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <AsyncHTTPConnection ['http://127.0.0.1:36711', HTTP/1.1, CLOSED, Request Count: 79]>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(self, request: Request) -> Response:
  778: [1253.2s]         if not self.can_handle_request(request.url.origin):
  778: [1253.2s]             raise RuntimeError(
  778: [1253.2s]                 f"Attempted to send request to {request.url.origin} on connection to {self._origin}"
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         async with self._request_lock:
  778: [1253.2s]             if self._connection is None:
  778: [1253.2s]                 try:
  778: [1253.2s]                     stream = await self._connect(request)
  778: [1253.2s]     
  778: [1253.2s]                     ssl_object = stream.get_extra_info("ssl_object")
  778: [1253.2s]                     http2_negotiated = (
  778: [1253.2s]                         ssl_object is not None
  778: [1253.2s]                         and ssl_object.selected_alpn_protocol() == "h2"
  778: [1253.2s]                     )
  778: [1253.2s]                     if http2_negotiated or (self._http2 and not self._http1):
  778: [1253.2s]                         from .http2 import AsyncHTTP2Connection
  778: [1253.2s]     
  778: [1253.2s]                         self._connection = AsyncHTTP2Connection(
  778: [1253.2s]                             origin=self._origin,
  778: [1253.2s]                             stream=stream,
  778: [1253.2s]                             keepalive_expiry=self._keepalive_expiry,
  778: [1253.2s]                         )
  778: [1253.2s]                     else:
  778: [1253.2s]                         self._connection = AsyncHTTP11Connection(
  778: [1253.2s]                             origin=self._origin,
  778: [1253.2s]                             stream=stream,
  778: [1253.2s]                             keepalive_expiry=self._keepalive_expiry,
  778: [1253.2s]                         )
  778: [1253.2s]                 except Exception as exc:
  778: [1253.2s]                     self._connect_failed = True
  778: [1253.2s]                     raise exc
  778: [1253.2s]             elif not self._connection.is_available():
  778: [1253.2s]                 raise ConnectionNotAvailable()
  778: [1253.2s]     
  778: [1253.2s] >       return await self._connection.handle_async_request(request)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py:96: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <AsyncHTTP11Connection ['http://127.0.0.1:36711', CLOSED, Request Count: 79]>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(self, request: Request) -> Response:
  778: [1253.2s]         if not self.can_handle_request(request.url.origin):
  778: [1253.2s]             raise RuntimeError(
  778: [1253.2s]                 f"Attempted to send request to {request.url.origin} on connection "
  778: [1253.2s]                 f"to {self._origin}"
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         async with self._state_lock:
  778: [1253.2s]             if self._state in (HTTPConnectionState.NEW, HTTPConnectionState.IDLE):
  778: [1253.2s]                 self._request_count += 1
  778: [1253.2s]                 self._state = HTTPConnectionState.ACTIVE
  778: [1253.2s]                 self._expire_at = None
  778: [1253.2s]             else:
  778: [1253.2s]                 raise ConnectionNotAvailable()
  778: [1253.2s]     
  778: [1253.2s]         try:
  778: [1253.2s]             kwargs = {"request": request}
  778: [1253.2s]             async with Trace("send_request_headers", logger, request, kwargs) as trace:
  778: [1253.2s]                 await self._send_request_headers(**kwargs)
  778: [1253.2s]             async with Trace("send_request_body", logger, request, kwargs) as trace:
  778: [1253.2s]                 await self._send_request_body(**kwargs)
  778: [1253.2s]             async with Trace(
  778: [1253.2s]                 "receive_response_headers", logger, request, kwargs
  778: [1253.2s]             ) as trace:
  778: [1253.2s]                 (
  778: [1253.2s]                     http_version,
  778: [1253.2s]                     status,
  778: [1253.2s]                     reason_phrase,
  778: [1253.2s]                     headers,
  778: [1253.2s]                 ) = await self._receive_response_headers(**kwargs)
  778: [1253.2s]                 trace.return_value = (
  778: [1253.2s]                     http_version,
  778: [1253.2s]                     status,
  778: [1253.2s]                     reason_phrase,
  778: [1253.2s]                     headers,
  778: [1253.2s]                 )
  778: [1253.2s]     
  778: [1253.2s]             return Response(
  778: [1253.2s]                 status=status,
  778: [1253.2s]                 headers=headers,
  778: [1253.2s]                 content=HTTP11ConnectionByteStream(self, request),
  778: [1253.2s]                 extensions={
  778: [1253.2s]                     "http_version": http_version,
  778: [1253.2s]                     "reason_phrase": reason_phrase,
  778: [1253.2s]                     "network_stream": self._network_stream,
  778: [1253.2s]                 },
  778: [1253.2s]             )
  778: [1253.2s]         except BaseException as exc:
  778: [1253.2s]             with AsyncShieldCancellation():
  778: [1253.2s]                 async with Trace("response_closed", logger, request) as trace:
  778: [1253.2s]                     await self._response_closed()
  778: [1253.2s] >           raise exc
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/http11.py:121: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <AsyncHTTP11Connection ['http://127.0.0.1:36711', CLOSED, Request Count: 79]>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(self, request: Request) -> Response:
  778: [1253.2s]         if not self.can_handle_request(request.url.origin):
  778: [1253.2s]             raise RuntimeError(
  778: [1253.2s]                 f"Attempted to send request to {request.url.origin} on connection "
  778: [1253.2s]                 f"to {self._origin}"
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         async with self._state_lock:
  778: [1253.2s]             if self._state in (HTTPConnectionState.NEW, HTTPConnectionState.IDLE):
  778: [1253.2s]                 self._request_count += 1
  778: [1253.2s]                 self._state = HTTPConnectionState.ACTIVE
  778: [1253.2s]                 self._expire_at = None
  778: [1253.2s]             else:
  778: [1253.2s]                 raise ConnectionNotAvailable()
  778: [1253.2s]     
  778: [1253.2s]         try:
  778: [1253.2s]             kwargs = {"request": request}
  778: [1253.2s]             async with Trace("send_request_headers", logger, request, kwargs) as trace:
  778: [1253.2s]                 await self._send_request_headers(**kwargs)
  778: [1253.2s]             async with Trace("send_request_body", logger, request, kwargs) as trace:
  778: [1253.2s]                 await self._send_request_body(**kwargs)
  778: [1253.2s]             async with Trace(
  778: [1253.2s]                 "receive_response_headers", logger, request, kwargs
  778: [1253.2s]             ) as trace:
  778: [1253.2s]                 (
  778: [1253.2s]                     http_version,
  778: [1253.2s]                     status,
  778: [1253.2s]                     reason_phrase,
  778: [1253.2s]                     headers,
  778: [1253.2s] >               ) = await self._receive_response_headers(**kwargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/http11.py:99: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <AsyncHTTP11Connection ['http://127.0.0.1:36711', CLOSED, Request Count: 79]>
  778: [1253.2s] request = <Request [b'POST']>
  778: [1253.2s] 
  778: [1253.2s]     async def _receive_response_headers(
  778: [1253.2s]         self, request: Request
  778: [1253.2s]     ) -> Tuple[bytes, int, bytes, List[Tuple[bytes, bytes]]]:
  778: [1253.2s]         timeouts = request.extensions.get("timeout", {})
  778: [1253.2s]         timeout = timeouts.get("read", None)
  778: [1253.2s]     
  778: [1253.2s]         while True:
  778: [1253.2s] >           event = await self._receive_event(timeout=timeout)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/http11.py:164: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <AsyncHTTP11Connection ['http://127.0.0.1:36711', CLOSED, Request Count: 79]>
  778: [1253.2s] timeout = None
  778: [1253.2s] 
  778: [1253.2s]     async def _receive_event(
  778: [1253.2s]         self, timeout: Optional[float] = None
  778: [1253.2s]     ) -> Union[h11.Event, Type[h11.PAUSED]]:
  778: [1253.2s]         while True:
  778: [1253.2s]             with map_exceptions({h11.RemoteProtocolError: RemoteProtocolError}):
  778: [1253.2s]                 event = self._h11_state.next_event()
  778: [1253.2s]     
  778: [1253.2s]             if event is h11.NEED_DATA:
  778: [1253.2s] >               data = await self._network_stream.read(
  778: [1253.2s]                     self.READ_NUM_BYTES, timeout=timeout
  778: [1253.2s]                 )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_async/http11.py:200: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpcore._backends.anyio.AnyIOStream object at 0x7f958496a470>
  778: [1253.2s] max_bytes = 65536, timeout = None
  778: [1253.2s] 
  778: [1253.2s]     async def read(
  778: [1253.2s]         self, max_bytes: int, timeout: typing.Optional[float] = None
  778: [1253.2s]     ) -> bytes:
  778: [1253.2s]         exc_map = {
  778: [1253.2s]             TimeoutError: ReadTimeout,
  778: [1253.2s]             anyio.BrokenResourceError: ReadError,
  778: [1253.2s]             anyio.ClosedResourceError: ReadError,
  778: [1253.2s]         }
  778: [1253.2s] >       with map_exceptions(exc_map):
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_backends/anyio.py:31: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <contextlib._GeneratorContextManager object at 0x7f9582ead990>
  778: [1253.2s] typ = <class 'anyio.ClosedResourceError'>, value = ClosedResourceError()
  778: [1253.2s] traceback = <traceback object at 0x7f9583a33640>
  778: [1253.2s] 
  778: [1253.2s]     def __exit__(self, typ, value, traceback):
  778: [1253.2s]         if typ is None:
  778: [1253.2s]             try:
  778: [1253.2s]                 next(self.gen)
  778: [1253.2s]             except StopIteration:
  778: [1253.2s]                 return False
  778: [1253.2s]             else:
  778: [1253.2s]                 raise RuntimeError("generator didn't stop")
  778: [1253.2s]         else:
  778: [1253.2s]             if value is None:
  778: [1253.2s]                 # Need to force instantiation so we can reliably
  778: [1253.2s]                 # tell if we get the same exception back
  778: [1253.2s]                 value = typ()
  778: [1253.2s]             try:
  778: [1253.2s] >               self.gen.throw(typ, value, traceback)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/contextlib.py:153: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] map = {<class 'TimeoutError'>: <class 'httpcore.ReadTimeout'>, <class 'anyio.BrokenResourceError'>: <class 'httpcore.ReadError'>, <class 'anyio.ClosedResourceError'>: <class 'httpcore.ReadError'>}
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def map_exceptions(map: ExceptionMapping) -> Iterator[None]:
  778: [1253.2s]         try:
  778: [1253.2s]             yield
  778: [1253.2s]         except Exception as exc:  # noqa: PIE786
  778: [1253.2s]             for from_exc, to_exc in map.items():
  778: [1253.2s]                 if isinstance(exc, from_exc):
  778: [1253.2s] >                   raise to_exc(exc) from exc
  778: [1253.2s] E                   httpcore.ReadError
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpcore/_exceptions.py:14: ReadError
  778: [1253.2s] 
  778: [1253.2s] The above exception was the direct cause of the following exception:
  778: [1253.2s] 
  778: [1253.2s] self = Context(session=<gql.client.AsyncClientSession object at 0x7f958496aa70>, schema=<gql.dsl.DSLSchema object at 0x7f9584...ainer', name='stdout', args={}, children={})]), converter=<cattrs.preconf.json.JsonConverter object at 0x7f9582d61590>)
  778: [1253.2s] query = DocumentNode
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def _handle_execute(self, query: graphql.DocumentNode):
  778: [1253.2s]         # Reduces duplication when handling errors, between sync and async.
  778: [1253.2s]         try:
  778: [1253.2s] >           yield
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/base.py:244: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = Context(session=<gql.client.AsyncClientSession object at 0x7f958496aa70>, schema=<gql.dsl.DSLSchema object at 0x7f9584...ainer', name='stdout', args={}, children={})]), converter=<cattrs.preconf.json.JsonConverter object at 0x7f9582d61590>)
  778: [1253.2s] return_type = <class 'str'>
  778: [1253.2s] 
  778: [1253.2s]     async def execute(self, return_type: type[T] | None = None) -> T | None:
  778: [1253.2s]         assert isinstance(self.session, AsyncClientSession)
  778: [1253.2s]         await self.resolve_ids()
  778: [1253.2s]         query = self.query()
  778: [1253.2s]         with self._handle_execute(query):
  778: [1253.2s] >           result = await self.session.execute(query)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/base.py:157: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <gql.client.AsyncClientSession object at 0x7f958496aa70>
  778: [1253.2s] document = DocumentNode, variable_values = None, operation_name = None
  778: [1253.2s] serialize_variables = None, parse_result = None, get_execution_result = False
  778: [1253.2s] kwargs = {}
  778: [1253.2s] 
  778: [1253.2s]     async def execute(
  778: [1253.2s]         self,
  778: [1253.2s]         document: DocumentNode,
  778: [1253.2s]         variable_values: Optional[Dict[str, Any]] = None,
  778: [1253.2s]         operation_name: Optional[str] = None,
  778: [1253.2s]         serialize_variables: Optional[bool] = None,
  778: [1253.2s]         parse_result: Optional[bool] = None,
  778: [1253.2s]         get_execution_result: bool = False,
  778: [1253.2s]         **kwargs,
  778: [1253.2s]     ) -> Union[Dict[str, Any], ExecutionResult]:
  778: [1253.2s]         """Coroutine to execute the provided document AST asynchronously using
  778: [1253.2s]         the async transport.
  778: [1253.2s]     
  778: [1253.2s]         Raises a TransportQueryError if an error has been returned in
  778: [1253.2s]             the ExecutionResult.
  778: [1253.2s]     
  778: [1253.2s]         :param document: GraphQL query as AST Node object.
  778: [1253.2s]         :param variable_values: Dictionary of input parameters.
  778: [1253.2s]         :param operation_name: Name of the operation that shall be executed.
  778: [1253.2s]         :param serialize_variables: whether the variable values should be
  778: [1253.2s]             serialized. Used for custom scalars and/or enums.
  778: [1253.2s]             By default use the serialize_variables argument of the client.
  778: [1253.2s]         :param parse_result: Whether gql will unserialize the result.
  778: [1253.2s]             By default use the parse_results argument of the client.
  778: [1253.2s]         :param get_execution_result: return the full ExecutionResult instance instead of
  778: [1253.2s]             only the "data" field. Necessary if you want to get the "extensions" field.
  778: [1253.2s]     
  778: [1253.2s]         The extra arguments are passed to the transport execute method."""
  778: [1253.2s]     
  778: [1253.2s]         # Validate and execute on the transport
  778: [1253.2s] >       result = await self._execute(
  778: [1253.2s]             document,
  778: [1253.2s]             variable_values=variable_values,
  778: [1253.2s]             operation_name=operation_name,
  778: [1253.2s]             serialize_variables=serialize_variables,
  778: [1253.2s]             parse_result=parse_result,
  778: [1253.2s]             **kwargs,
  778: [1253.2s]         )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/gql/client.py:1220: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <gql.client.AsyncClientSession object at 0x7f958496aa70>
  778: [1253.2s] document = DocumentNode, variable_values = None, operation_name = None
  778: [1253.2s] serialize_variables = None, parse_result = None, kwargs = {}
  778: [1253.2s] 
  778: [1253.2s]     async def _execute(
  778: [1253.2s]         self,
  778: [1253.2s]         document: DocumentNode,
  778: [1253.2s]         variable_values: Optional[Dict[str, Any]] = None,
  778: [1253.2s]         operation_name: Optional[str] = None,
  778: [1253.2s]         serialize_variables: Optional[bool] = None,
  778: [1253.2s]         parse_result: Optional[bool] = None,
  778: [1253.2s]         **kwargs,
  778: [1253.2s]     ) -> ExecutionResult:
  778: [1253.2s]         """Coroutine to execute the provided document AST asynchronously using
  778: [1253.2s]         the async transport, returning an ExecutionResult object.
  778: [1253.2s]     
  778: [1253.2s]         * Validate the query with the schema if provided.
  778: [1253.2s]         * Serialize the variable_values if requested.
  778: [1253.2s]     
  778: [1253.2s]         :param document: GraphQL query as AST Node object.
  778: [1253.2s]         :param variable_values: Dictionary of input parameters.
  778: [1253.2s]         :param operation_name: Name of the operation that shall be executed.
  778: [1253.2s]         :param serialize_variables: whether the variable values should be
  778: [1253.2s]             serialized. Used for custom scalars and/or enums.
  778: [1253.2s]             By default use the serialize_variables argument of the client.
  778: [1253.2s]         :param parse_result: Whether gql will unserialize the result.
  778: [1253.2s]             By default use the parse_results argument of the client.
  778: [1253.2s]     
  778: [1253.2s]         The extra arguments are passed to the transport execute method."""
  778: [1253.2s]     
  778: [1253.2s]         # Validate document
  778: [1253.2s]         if self.client.schema:
  778: [1253.2s]             self.client.validate(document)
  778: [1253.2s]     
  778: [1253.2s]             # Parse variable values for custom scalars if requested
  778: [1253.2s]             if variable_values is not None:
  778: [1253.2s]                 if serialize_variables or (
  778: [1253.2s]                     serialize_variables is None and self.client.serialize_variables
  778: [1253.2s]                 ):
  778: [1253.2s]                     variable_values = serialize_variable_values(
  778: [1253.2s]                         self.client.schema,
  778: [1253.2s]                         document,
  778: [1253.2s]                         variable_values,
  778: [1253.2s]                         operation_name=operation_name,
  778: [1253.2s]                     )
  778: [1253.2s]     
  778: [1253.2s]         # Execute the query with the transport with a timeout
  778: [1253.2s] >       result = await asyncio.wait_for(
  778: [1253.2s]             self.transport.execute(
  778: [1253.2s]                 document,
  778: [1253.2s]                 variable_values=variable_values,
  778: [1253.2s]                 operation_name=operation_name,
  778: [1253.2s]                 **kwargs,
  778: [1253.2s]             ),
  778: [1253.2s]             self.client.execute_timeout,
  778: [1253.2s]         )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/gql/client.py:1126: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] fut = <coroutine object HTTPXAsyncTransport.execute at 0x7f9583a4b300>
  778: [1253.2s] timeout = None
  778: [1253.2s] 
  778: [1253.2s]     async def wait_for(fut, timeout):
  778: [1253.2s]         """Wait for the single Future or coroutine to complete, with timeout.
  778: [1253.2s]     
  778: [1253.2s]         Coroutine will be wrapped in Task.
  778: [1253.2s]     
  778: [1253.2s]         Returns result of the Future or coroutine.  When a timeout occurs,
  778: [1253.2s]         it cancels the task and raises TimeoutError.  To avoid the task
  778: [1253.2s]         cancellation, wrap it in shield().
  778: [1253.2s]     
  778: [1253.2s]         If the wait is cancelled, the task is also cancelled.
  778: [1253.2s]     
  778: [1253.2s]         This function is a coroutine.
  778: [1253.2s]         """
  778: [1253.2s]         loop = events.get_running_loop()
  778: [1253.2s]     
  778: [1253.2s]         if timeout is None:
  778: [1253.2s] >           return await fut
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/tasks.py:408: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <dagger.transport.httpx.HTTPXAsyncTransport object at 0x7f958496b8b0>
  778: [1253.2s] document = DocumentNode, variable_values = None, operation_name = None
  778: [1253.2s] extra_args = None, upload_files = False
  778: [1253.2s] 
  778: [1253.2s]     async def execute(
  778: [1253.2s]         self,
  778: [1253.2s]         document: DocumentNode,
  778: [1253.2s]         variable_values: Optional[Dict[str, Any]] = None,
  778: [1253.2s]         operation_name: Optional[str] = None,
  778: [1253.2s]         extra_args: Optional[Dict[str, Any]] = None,
  778: [1253.2s]         upload_files: bool = False,
  778: [1253.2s]     ) -> ExecutionResult:
  778: [1253.2s]         if not self.client:
  778: [1253.2s]             raise TransportClosed("Transport is not connected")
  778: [1253.2s]     
  778: [1253.2s]         post_args = self._prepare_request(
  778: [1253.2s]             document,
  778: [1253.2s]             variable_values,
  778: [1253.2s]             operation_name,
  778: [1253.2s]             extra_args,
  778: [1253.2s]         )
  778: [1253.2s]     
  778: [1253.2s] >       response = await self.client.post(self.url, **post_args)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/transport/httpx.py:179: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>
  778: [1253.2s] url = URL('http://127.0.0.1:36711/query')
  778: [1253.2s] 
  778: [1253.2s]     async def post(
  778: [1253.2s]         self,
  778: [1253.2s]         url: URLTypes,
  778: [1253.2s]         *,
  778: [1253.2s]         content: typing.Optional[RequestContent] = None,
  778: [1253.2s]         data: typing.Optional[RequestData] = None,
  778: [1253.2s]         files: typing.Optional[RequestFiles] = None,
  778: [1253.2s]         json: typing.Optional[typing.Any] = None,
  778: [1253.2s]         params: typing.Optional[QueryParamTypes] = None,
  778: [1253.2s]         headers: typing.Optional[HeaderTypes] = None,
  778: [1253.2s]         cookies: typing.Optional[CookieTypes] = None,
  778: [1253.2s]         auth: typing.Union[AuthTypes, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         follow_redirects: typing.Union[bool, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         timeout: typing.Union[TimeoutTypes, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         extensions: typing.Optional[RequestExtensions] = None,
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Send a `POST` request.
  778: [1253.2s]     
  778: [1253.2s]         **Parameters**: See `httpx.request`.
  778: [1253.2s]         """
  778: [1253.2s] >       return await self.request(
  778: [1253.2s]             "POST",
  778: [1253.2s]             url,
  778: [1253.2s]             content=content,
  778: [1253.2s]             data=data,
  778: [1253.2s]             files=files,
  778: [1253.2s]             json=json,
  778: [1253.2s]             params=params,
  778: [1253.2s]             headers=headers,
  778: [1253.2s]             cookies=cookies,
  778: [1253.2s]             auth=auth,
  778: [1253.2s]             follow_redirects=follow_redirects,
  778: [1253.2s]             timeout=timeout,
  778: [1253.2s]             extensions=extensions,
  778: [1253.2s]         )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1848: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>, method = 'POST'
  778: [1253.2s] url = URL('http://127.0.0.1:36711/query')
  778: [1253.2s] 
  778: [1253.2s]     async def request(
  778: [1253.2s]         self,
  778: [1253.2s]         method: str,
  778: [1253.2s]         url: URLTypes,
  778: [1253.2s]         *,
  778: [1253.2s]         content: typing.Optional[RequestContent] = None,
  778: [1253.2s]         data: typing.Optional[RequestData] = None,
  778: [1253.2s]         files: typing.Optional[RequestFiles] = None,
  778: [1253.2s]         json: typing.Optional[typing.Any] = None,
  778: [1253.2s]         params: typing.Optional[QueryParamTypes] = None,
  778: [1253.2s]         headers: typing.Optional[HeaderTypes] = None,
  778: [1253.2s]         cookies: typing.Optional[CookieTypes] = None,
  778: [1253.2s]         auth: typing.Union[AuthTypes, UseClientDefault, None] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         follow_redirects: typing.Union[bool, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         timeout: typing.Union[TimeoutTypes, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         extensions: typing.Optional[RequestExtensions] = None,
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Build and send a request.
  778: [1253.2s]     
  778: [1253.2s]         Equivalent to:
  778: [1253.2s]     
  778: [1253.2s]         ```python
  778: [1253.2s]         request = client.build_request(...)
  778: [1253.2s]         response = await client.send(request, ...)
  778: [1253.2s]         ```
  778: [1253.2s]     
  778: [1253.2s]         See `AsyncClient.build_request()`, `AsyncClient.send()`
  778: [1253.2s]         and [Merging of configuration][0] for how the various parameters
  778: [1253.2s]         are merged with client-level configuration.
  778: [1253.2s]     
  778: [1253.2s]         [0]: /advanced/#merging-of-configuration
  778: [1253.2s]         """
  778: [1253.2s]         request = self.build_request(
  778: [1253.2s]             method=method,
  778: [1253.2s]             url=url,
  778: [1253.2s]             content=content,
  778: [1253.2s]             data=data,
  778: [1253.2s]             files=files,
  778: [1253.2s]             json=json,
  778: [1253.2s]             params=params,
  778: [1253.2s]             headers=headers,
  778: [1253.2s]             cookies=cookies,
  778: [1253.2s]             timeout=timeout,
  778: [1253.2s]             extensions=extensions,
  778: [1253.2s]         )
  778: [1253.2s] >       return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1530: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] 
  778: [1253.2s]     async def send(
  778: [1253.2s]         self,
  778: [1253.2s]         request: Request,
  778: [1253.2s]         *,
  778: [1253.2s]         stream: bool = False,
  778: [1253.2s]         auth: typing.Union[AuthTypes, UseClientDefault, None] = USE_CLIENT_DEFAULT,
  778: [1253.2s]         follow_redirects: typing.Union[bool, UseClientDefault] = USE_CLIENT_DEFAULT,
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Send a request.
  778: [1253.2s]     
  778: [1253.2s]         The request is sent as-is, unmodified.
  778: [1253.2s]     
  778: [1253.2s]         Typically you'll want to build one with `AsyncClient.build_request()`
  778: [1253.2s]         so that any client-level configuration is merged into the request,
  778: [1253.2s]         but passing an explicit `httpx.Request()` is supported as well.
  778: [1253.2s]     
  778: [1253.2s]         See also: [Request instances][0]
  778: [1253.2s]     
  778: [1253.2s]         [0]: /advanced/#request-instances
  778: [1253.2s]         """
  778: [1253.2s]         if self._state == ClientState.CLOSED:
  778: [1253.2s]             raise RuntimeError("Cannot send a request, as the client has been closed.")
  778: [1253.2s]     
  778: [1253.2s]         self._state = ClientState.OPENED
  778: [1253.2s]         follow_redirects = (
  778: [1253.2s]             self.follow_redirects
  778: [1253.2s]             if isinstance(follow_redirects, UseClientDefault)
  778: [1253.2s]             else follow_redirects
  778: [1253.2s]         )
  778: [1253.2s]     
  778: [1253.2s]         auth = self._build_request_auth(request, auth)
  778: [1253.2s]     
  778: [1253.2s] >       response = await self._send_handling_auth(
  778: [1253.2s]             request,
  778: [1253.2s]             auth=auth,
  778: [1253.2s]             follow_redirects=follow_redirects,
  778: [1253.2s]             history=[],
  778: [1253.2s]         )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1617: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] auth = <httpx.BasicAuth object at 0x7f958496b730>, follow_redirects = False
  778: [1253.2s] history = []
  778: [1253.2s] 
  778: [1253.2s]     async def _send_handling_auth(
  778: [1253.2s]         self,
  778: [1253.2s]         request: Request,
  778: [1253.2s]         auth: Auth,
  778: [1253.2s]         follow_redirects: bool,
  778: [1253.2s]         history: typing.List[Response],
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         auth_flow = auth.async_auth_flow(request)
  778: [1253.2s]         try:
  778: [1253.2s]             request = await auth_flow.__anext__()
  778: [1253.2s]     
  778: [1253.2s]             while True:
  778: [1253.2s] >               response = await self._send_handling_redirects(
  778: [1253.2s]                     request,
  778: [1253.2s]                     follow_redirects=follow_redirects,
  778: [1253.2s]                     history=history,
  778: [1253.2s]                 )
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1645: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] follow_redirects = False, history = []
  778: [1253.2s] 
  778: [1253.2s]     async def _send_handling_redirects(
  778: [1253.2s]         self,
  778: [1253.2s]         request: Request,
  778: [1253.2s]         follow_redirects: bool,
  778: [1253.2s]         history: typing.List[Response],
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         while True:
  778: [1253.2s]             if len(history) > self.max_redirects:
  778: [1253.2s]                 raise TooManyRedirects(
  778: [1253.2s]                     "Exceeded maximum allowed redirects.", request=request
  778: [1253.2s]                 )
  778: [1253.2s]     
  778: [1253.2s]             for hook in self._event_hooks["request"]:
  778: [1253.2s]                 await hook(request)
  778: [1253.2s]     
  778: [1253.2s] >           response = await self._send_single_request(request)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1682: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncClient object at 0x7f958496a740>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] 
  778: [1253.2s]     async def _send_single_request(self, request: Request) -> Response:
  778: [1253.2s]         """
  778: [1253.2s]         Sends a single request, without handling any redirections.
  778: [1253.2s]         """
  778: [1253.2s]         transport = self._transport_for_url(request.url)
  778: [1253.2s]         timer = Timer()
  778: [1253.2s]         await timer.async_start()
  778: [1253.2s]     
  778: [1253.2s]         if not isinstance(request.stream, AsyncByteStream):
  778: [1253.2s]             raise RuntimeError(
  778: [1253.2s]                 "Attempted to send an sync request with an AsyncClient instance."
  778: [1253.2s]             )
  778: [1253.2s]     
  778: [1253.2s]         with request_context(request=request):
  778: [1253.2s] >           response = await transport.handle_async_request(request)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_client.py:1719: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <httpx.AsyncHTTPTransport object at 0x7f958496ba00>
  778: [1253.2s] request = <Request('POST', 'http://127.0.0.1:36711/query')>
  778: [1253.2s] 
  778: [1253.2s]     async def handle_async_request(
  778: [1253.2s]         self,
  778: [1253.2s]         request: Request,
  778: [1253.2s]     ) -> Response:
  778: [1253.2s]         assert isinstance(request.stream, AsyncByteStream)
  778: [1253.2s]     
  778: [1253.2s]         req = httpcore.Request(
  778: [1253.2s]             method=request.method,
  778: [1253.2s]             url=httpcore.URL(
  778: [1253.2s]                 scheme=request.url.raw_scheme,
  778: [1253.2s]                 host=request.url.raw_host,
  778: [1253.2s]                 port=request.url.port,
  778: [1253.2s]                 target=request.url.raw_path,
  778: [1253.2s]             ),
  778: [1253.2s]             headers=request.headers.raw,
  778: [1253.2s]             content=request.stream,
  778: [1253.2s]             extensions=request.extensions,
  778: [1253.2s]         )
  778: [1253.2s] >       with map_httpcore_exceptions():
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_transports/default.py:352: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <contextlib._GeneratorContextManager object at 0x7f9582d374f0>
  778: [1253.2s] typ = <class 'httpcore.ReadError'>, value = ReadError(ClosedResourceError())
  778: [1253.2s] traceback = <traceback object at 0x7f9583a30800>
  778: [1253.2s] 
  778: [1253.2s]     def __exit__(self, typ, value, traceback):
  778: [1253.2s]         if typ is None:
  778: [1253.2s]             try:
  778: [1253.2s]                 next(self.gen)
  778: [1253.2s]             except StopIteration:
  778: [1253.2s]                 return False
  778: [1253.2s]             else:
  778: [1253.2s]                 raise RuntimeError("generator didn't stop")
  778: [1253.2s]         else:
  778: [1253.2s]             if value is None:
  778: [1253.2s]                 # Need to force instantiation so we can reliably
  778: [1253.2s]                 # tell if we get the same exception back
  778: [1253.2s]                 value = typ()
  778: [1253.2s]             try:
  778: [1253.2s] >               self.gen.throw(typ, value, traceback)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/contextlib.py:153: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def map_httpcore_exceptions() -> typing.Iterator[None]:
  778: [1253.2s]         try:
  778: [1253.2s]             yield
  778: [1253.2s]         except Exception as exc:  # noqa: PIE-786
  778: [1253.2s]             mapped_exc = None
  778: [1253.2s]     
  778: [1253.2s]             for from_exc, to_exc in HTTPCORE_EXC_MAP.items():
  778: [1253.2s]                 if not isinstance(exc, from_exc):
  778: [1253.2s]                     continue
  778: [1253.2s]                 # We want to map to the most specific exception we can find.
  778: [1253.2s]                 # Eg if `exc` is an `httpcore.ReadTimeout`, we want to map to
  778: [1253.2s]                 # `httpx.ReadTimeout`, not just `httpx.TimeoutException`.
  778: [1253.2s]                 if mapped_exc is None or issubclass(to_exc, mapped_exc):
  778: [1253.2s]                     mapped_exc = to_exc
  778: [1253.2s]     
  778: [1253.2s]             if mapped_exc is None:  # pragma: no cover
  778: [1253.2s]                 raise
  778: [1253.2s]     
  778: [1253.2s]             message = str(exc)
  778: [1253.2s] >           raise mapped_exc(message) from exc
  778: [1253.2s] E           httpx.ReadError
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/httpx/_transports/default.py:77: ReadError
  778: [1253.2s] 
  778: [1253.2s] The above exception was the direct cause of the following exception:
  778: [1253.2s] 
  778: [1253.2s] anyio_backend = 'asyncio', args = (), kwargs = {'anyio_backend': 'asyncio'}
  778: [1253.2s] backend_name = 'asyncio', backend_options = {}
  778: [1253.2s] runner = <anyio._backends._asyncio.TestRunner object at 0x7f958713aa70>
  778: [1253.2s] 
  778: [1253.2s]     def wrapper(*args, anyio_backend, **kwargs):  # type: ignore[no-untyped-def]
  778: [1253.2s]         backend_name, backend_options = extract_backend_and_options(anyio_backend)
  778: [1253.2s]         if has_backend_arg:
  778: [1253.2s]             kwargs["anyio_backend"] = anyio_backend
  778: [1253.2s]     
  778: [1253.2s]         with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s]             if isasyncgenfunction(func):
  778: [1253.2s] >               yield from runner.run_asyncgen_fixture(func, kwargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/pytest_plugin.py:68: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2098: in run_asyncgen_fixture
  778: [1253.2s]     self._raise_async_exceptions()
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2054: in _raise_async_exceptions
  778: [1253.2s]     raise exceptions[0]
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/connector_acceptance_test/tests/test_core.py:968: in test_read
  778: [1253.2s]     output = await docker_runner.call_read(connector_config, configured_catalog)
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/connector_acceptance_test/utils/connector_runner.py:47: in call_read
  778: [1253.2s]     return await self._run(
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/connector_acceptance_test/utils/connector_runner.py:104: in _run
  778: [1253.2s]     output = await container.stdout()
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/base.py:486: in async_wrapper
  778: [1253.2s]     return await bear(*args, **kwargs)
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/gen.py:878: in stdout
  778: [1253.2s]     return await _ctx.execute(str)
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/base.py:156: in execute
  778: [1253.2s]     with self._handle_execute(query):
  778: [1253.2s] /usr/local/lib/python3.10/contextlib.py:153: in __exit__
  778: [1253.2s]     self.gen.throw(typ, value, traceback)
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = Context(session=<gql.client.AsyncClientSession object at 0x7f958496aa70>, schema=<gql.dsl.DSLSchema object at 0x7f9584...ainer', name='stdout', args={}, children={})]), converter=<cattrs.preconf.json.JsonConverter object at 0x7f9582d61590>)
  778: [1253.2s] query = DocumentNode
  778: [1253.2s] 
  778: [1253.2s]     @contextlib.contextmanager
  778: [1253.2s]     def _handle_execute(self, query: graphql.DocumentNode):
  778: [1253.2s]         # Reduces duplication when handling errors, between sync and async.
  778: [1253.2s]         try:
  778: [1253.2s]             yield
  778: [1253.2s]     
  778: [1253.2s]         except httpx.TimeoutException as e:
  778: [1253.2s]             msg = (
  778: [1253.2s]                 "Request timed out. Try setting a higher value in 'execute_timeout' "
  778: [1253.2s]                 "config for this `dagger.Connection()`."
  778: [1253.2s]             )
  778: [1253.2s]             raise ExecuteTimeoutError(msg) from e
  778: [1253.2s]     
  778: [1253.2s]         except httpx.RequestError as e:
  778: [1253.2s]             msg = f"Failed to make request: {e}"
  778: [1253.2s] >           raise TransportError(msg) from e
  778: [1253.2s] E           dagger.exceptions.TransportError: Failed to make request:
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/dagger/api/base.py:255: TransportError
  778: [1253.2s] =================================== FAILURES ===================================
  778: [1253.2s] _______________________ TestBasicRead.test_read[inputs0] _______________________
  778: [1253.2s] 
  778: [1253.2s] pyfuncitem = <Function test_read[inputs0]>
  778: [1253.2s] 
  778: [1253.2s]     @pytest.hookimpl(tryfirst=True)
  778: [1253.2s]     def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
  778: [1253.2s]         def run_with_hypothesis(**kwargs: Any) -> None:
  778: [1253.2s]             with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s]                 runner.run_test(original_func, kwargs)
  778: [1253.2s]     
  778: [1253.2s]         backend = pyfuncitem.funcargs.get("anyio_backend")
  778: [1253.2s]         if backend:
  778: [1253.2s]             backend_name, backend_options = extract_backend_and_options(backend)
  778: [1253.2s]     
  778: [1253.2s]             if hasattr(pyfuncitem.obj, "hypothesis"):
  778: [1253.2s]                 # Wrap the inner test function unless it's already wrapped
  778: [1253.2s]                 original_func = pyfuncitem.obj.hypothesis.inner_test
  778: [1253.2s]                 if original_func.__qualname__ != run_with_hypothesis.__qualname__:
  778: [1253.2s]                     if iscoroutinefunction(original_func):
  778: [1253.2s]                         pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
  778: [1253.2s]     
  778: [1253.2s]                 return None
  778: [1253.2s]     
  778: [1253.2s]             if iscoroutinefunction(pyfuncitem.obj):
  778: [1253.2s]                 funcargs = pyfuncitem.funcargs
  778: [1253.2s]                 testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
  778: [1253.2s]                 with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s] >                   runner.run_test(pyfuncitem.obj, testargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/pytest_plugin.py:117: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2113: in run_test
  778: [1253.2s]     self._loop.run_until_complete(test_func(**kwargs))
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:636: in run_until_complete
  778: [1253.2s]     self.run_forever()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:603: in run_forever
  778: [1253.2s]     self._run_once()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:1871: in _run_once
  778: [1253.2s]     event_list = self._selector.select(timeout)
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <selectors.EpollSelector object at 0x7f95841e07f0>, timeout = -1
  778: [1253.2s] 
  778: [1253.2s]     def select(self, timeout=None):
  778: [1253.2s]         if timeout is None:
  778: [1253.2s]             timeout = -1
  778: [1253.2s]         elif timeout <= 0:
  778: [1253.2s]             timeout = 0
  778: [1253.2s]         else:
  778: [1253.2s]             # epoll_wait() has a resolution of 1 millisecond, round away
  778: [1253.2s]             # from zero to wait *at least* timeout seconds.
  778: [1253.2s]             timeout = math.ceil(timeout * 1e3) * 1e-3
  778: [1253.2s]     
  778: [1253.2s]         # epoll_wait() expects `maxevents` to be greater than zero;
  778: [1253.2s]         # we want to make sure that `select()` can be called when no
  778: [1253.2s]         # FD is registered.
  778: [1253.2s]         max_ev = max(len(self._fd_to_key), 1)
  778: [1253.2s]     
  778: [1253.2s]         ready = []
  778: [1253.2s]         try:
  778: [1253.2s] >           fd_event_list = self._selector.poll(timeout, max_ev)
  778: [1253.2s] E           Failed: Timeout >300.0s
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/selectors.py:469: Failed
  778: [1253.2s] ----------------------------- Captured stderr call -----------------------------
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] 
  778: [1253.2s] ~~~~~~~~~~~~~~~~ Stack of AnyIO worker thread (140280134059712) ~~~~~~~~~~~~~~~~
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
  778: [1253.2s]     self._bootstrap_inner()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
  778: [1253.2s]     self.run()
  778: [1253.2s]   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 797, in run
  778: [1253.2s]     item = self.queue.get()
  778: [1253.2s]   File "/usr/local/lib/python3.10/queue.py", line 171, in get
  778: [1253.2s]     self.not_empty.wait()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 320, in wait
  778: [1253.2s]     waiter.acquire()
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] _______________________ TestBasicRead.test_read[inputs1] _______________________
  778: [1253.2s] 
  778: [1253.2s] pyfuncitem = <Function test_read[inputs1]>
  778: [1253.2s] 
  778: [1253.2s]     @pytest.hookimpl(tryfirst=True)
  778: [1253.2s]     def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
  778: [1253.2s]         def run_with_hypothesis(**kwargs: Any) -> None:
  778: [1253.2s]             with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s]                 runner.run_test(original_func, kwargs)
  778: [1253.2s]     
  778: [1253.2s]         backend = pyfuncitem.funcargs.get("anyio_backend")
  778: [1253.2s]         if backend:
  778: [1253.2s]             backend_name, backend_options = extract_backend_and_options(backend)
  778: [1253.2s]     
  778: [1253.2s]             if hasattr(pyfuncitem.obj, "hypothesis"):
  778: [1253.2s]                 # Wrap the inner test function unless it's already wrapped
  778: [1253.2s]                 original_func = pyfuncitem.obj.hypothesis.inner_test
  778: [1253.2s]                 if original_func.__qualname__ != run_with_hypothesis.__qualname__:
  778: [1253.2s]                     if iscoroutinefunction(original_func):
  778: [1253.2s]                         pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
  778: [1253.2s]     
  778: [1253.2s]                 return None
  778: [1253.2s]     
  778: [1253.2s]             if iscoroutinefunction(pyfuncitem.obj):
  778: [1253.2s]                 funcargs = pyfuncitem.funcargs
  778: [1253.2s]                 testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
  778: [1253.2s]                 with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s] >                   runner.run_test(pyfuncitem.obj, testargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/pytest_plugin.py:117: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2113: in run_test
  778: [1253.2s]     self._loop.run_until_complete(test_func(**kwargs))
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:636: in run_until_complete
  778: [1253.2s]     self.run_forever()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:603: in run_forever
  778: [1253.2s]     self._run_once()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:1871: in _run_once
  778: [1253.2s]     event_list = self._selector.select(timeout)
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <selectors.EpollSelector object at 0x7f95841e07f0>, timeout = -1
  778: [1253.2s] 
  778: [1253.2s]     def select(self, timeout=None):
  778: [1253.2s]         if timeout is None:
  778: [1253.2s]             timeout = -1
  778: [1253.2s]         elif timeout <= 0:
  778: [1253.2s]             timeout = 0
  778: [1253.2s]         else:
  778: [1253.2s]             # epoll_wait() has a resolution of 1 millisecond, round away
  778: [1253.2s]             # from zero to wait *at least* timeout seconds.
  778: [1253.2s]             timeout = math.ceil(timeout * 1e3) * 1e-3
  778: [1253.2s]     
  778: [1253.2s]         # epoll_wait() expects `maxevents` to be greater than zero;
  778: [1253.2s]         # we want to make sure that `select()` can be called when no
  778: [1253.2s]         # FD is registered.
  778: [1253.2s]         max_ev = max(len(self._fd_to_key), 1)
  778: [1253.2s]     
  778: [1253.2s]         ready = []
  778: [1253.2s]         try:
  778: [1253.2s] >           fd_event_list = self._selector.poll(timeout, max_ev)
  778: [1253.2s] E           Failed: Timeout >300.0s
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/selectors.py:469: Failed
  778: [1253.2s] ----------------------------- Captured stderr call -----------------------------
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] 
  778: [1253.2s] ~~~~~~~~~~~~~~~~ Stack of AnyIO worker thread (140280134059712) ~~~~~~~~~~~~~~~~
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
  778: [1253.2s]     self._bootstrap_inner()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
  778: [1253.2s]     self.run()
  778: [1253.2s]   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 797, in run
  778: [1253.2s]     item = self.queue.get()
  778: [1253.2s]   File "/usr/local/lib/python3.10/queue.py", line 171, in get
  778: [1253.2s]     self.not_empty.wait()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 320, in wait
  778: [1253.2s]     waiter.acquire()
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] _________ TestBasicRead.test_airbyte_trace_message_on_failure[inputs0] _________
  778: [1253.2s] 
  778: [1253.2s] pyfuncitem = <Function test_airbyte_trace_message_on_failure[inputs0]>
  778: [1253.2s] 
  778: [1253.2s]     @pytest.hookimpl(tryfirst=True)
  778: [1253.2s]     def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
  778: [1253.2s]         def run_with_hypothesis(**kwargs: Any) -> None:
  778: [1253.2s]             with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s]                 runner.run_test(original_func, kwargs)
  778: [1253.2s]     
  778: [1253.2s]         backend = pyfuncitem.funcargs.get("anyio_backend")
  778: [1253.2s]         if backend:
  778: [1253.2s]             backend_name, backend_options = extract_backend_and_options(backend)
  778: [1253.2s]     
  778: [1253.2s]             if hasattr(pyfuncitem.obj, "hypothesis"):
  778: [1253.2s]                 # Wrap the inner test function unless it's already wrapped
  778: [1253.2s]                 original_func = pyfuncitem.obj.hypothesis.inner_test
  778: [1253.2s]                 if original_func.__qualname__ != run_with_hypothesis.__qualname__:
  778: [1253.2s]                     if iscoroutinefunction(original_func):
  778: [1253.2s]                         pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
  778: [1253.2s]     
  778: [1253.2s]                 return None
  778: [1253.2s]     
  778: [1253.2s]             if iscoroutinefunction(pyfuncitem.obj):
  778: [1253.2s]                 funcargs = pyfuncitem.funcargs
  778: [1253.2s]                 testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
  778: [1253.2s]                 with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s] >                   runner.run_test(pyfuncitem.obj, testargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/pytest_plugin.py:117: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2113: in run_test
  778: [1253.2s]     self._loop.run_until_complete(test_func(**kwargs))
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:636: in run_until_complete
  778: [1253.2s]     self.run_forever()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:603: in run_forever
  778: [1253.2s]     self._run_once()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:1871: in _run_once
  778: [1253.2s]     event_list = self._selector.select(timeout)
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <selectors.EpollSelector object at 0x7f95841e07f0>, timeout = -1
  778: [1253.2s] 
  778: [1253.2s]     def select(self, timeout=None):
  778: [1253.2s]         if timeout is None:
  778: [1253.2s]             timeout = -1
  778: [1253.2s]         elif timeout <= 0:
  778: [1253.2s]             timeout = 0
  778: [1253.2s]         else:
  778: [1253.2s]             # epoll_wait() has a resolution of 1 millisecond, round away
  778: [1253.2s]             # from zero to wait *at least* timeout seconds.
  778: [1253.2s]             timeout = math.ceil(timeout * 1e3) * 1e-3
  778: [1253.2s]     
  778: [1253.2s]         # epoll_wait() expects `maxevents` to be greater than zero;
  778: [1253.2s]         # we want to make sure that `select()` can be called when no
  778: [1253.2s]         # FD is registered.
  778: [1253.2s]         max_ev = max(len(self._fd_to_key), 1)
  778: [1253.2s]     
  778: [1253.2s]         ready = []
  778: [1253.2s]         try:
  778: [1253.2s] >           fd_event_list = self._selector.poll(timeout, max_ev)
  778: [1253.2s] E           Failed: Timeout >300.0s
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/selectors.py:469: Failed
  778: [1253.2s] ----------------------------- Captured stderr call -----------------------------
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] 
  778: [1253.2s] ~~~~~~~~~~~~~~~~ Stack of AnyIO worker thread (140280134059712) ~~~~~~~~~~~~~~~~
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
  778: [1253.2s]     self._bootstrap_inner()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
  778: [1253.2s]     self.run()
  778: [1253.2s]   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 797, in run
  778: [1253.2s]     item = self.queue.get()
  778: [1253.2s]   File "/usr/local/lib/python3.10/queue.py", line 171, in get
  778: [1253.2s]     self.not_empty.wait()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 320, in wait
  778: [1253.2s]     waiter.acquire()
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] _________ TestBasicRead.test_airbyte_trace_message_on_failure[inputs1] _________
  778: [1253.2s] 
  778: [1253.2s] pyfuncitem = <Function test_airbyte_trace_message_on_failure[inputs1]>
  778: [1253.2s] 
  778: [1253.2s]     @pytest.hookimpl(tryfirst=True)
  778: [1253.2s]     def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
  778: [1253.2s]         def run_with_hypothesis(**kwargs: Any) -> None:
  778: [1253.2s]             with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s]                 runner.run_test(original_func, kwargs)
  778: [1253.2s]     
  778: [1253.2s]         backend = pyfuncitem.funcargs.get("anyio_backend")
  778: [1253.2s]         if backend:
  778: [1253.2s]             backend_name, backend_options = extract_backend_and_options(backend)
  778: [1253.2s]     
  778: [1253.2s]             if hasattr(pyfuncitem.obj, "hypothesis"):
  778: [1253.2s]                 # Wrap the inner test function unless it's already wrapped
  778: [1253.2s]                 original_func = pyfuncitem.obj.hypothesis.inner_test
  778: [1253.2s]                 if original_func.__qualname__ != run_with_hypothesis.__qualname__:
  778: [1253.2s]                     if iscoroutinefunction(original_func):
  778: [1253.2s]                         pyfuncitem.obj.hypothesis.inner_test = run_with_hypothesis
  778: [1253.2s]     
  778: [1253.2s]                 return None
  778: [1253.2s]     
  778: [1253.2s]             if iscoroutinefunction(pyfuncitem.obj):
  778: [1253.2s]                 funcargs = pyfuncitem.funcargs
  778: [1253.2s]                 testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
  778: [1253.2s]                 with get_runner(backend_name, backend_options) as runner:
  778: [1253.2s] >                   runner.run_test(pyfuncitem.obj, testargs)
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/pytest_plugin.py:117: 
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:2113: in run_test
  778: [1253.2s]     self._loop.run_until_complete(test_func(**kwargs))
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:636: in run_until_complete
  778: [1253.2s]     self.run_forever()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:603: in run_forever
  778: [1253.2s]     self._run_once()
  778: [1253.2s] /usr/local/lib/python3.10/asyncio/base_events.py:1871: in _run_once
  778: [1253.2s]     event_list = self._selector.select(timeout)
  778: [1253.2s] _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  778: [1253.2s] 
  778: [1253.2s] self = <selectors.EpollSelector object at 0x7f95841e07f0>, timeout = -1
  778: [1253.2s] 
  778: [1253.2s]     def select(self, timeout=None):
  778: [1253.2s]         if timeout is None:
  778: [1253.2s]             timeout = -1
  778: [1253.2s]         elif timeout <= 0:
  778: [1253.2s]             timeout = 0
  778: [1253.2s]         else:
  778: [1253.2s]             # epoll_wait() has a resolution of 1 millisecond, round away
  778: [1253.2s]             # from zero to wait *at least* timeout seconds.
  778: [1253.2s]             timeout = math.ceil(timeout * 1e3) * 1e-3
  778: [1253.2s]     
  778: [1253.2s]         # epoll_wait() expects `maxevents` to be greater than zero;
  778: [1253.2s]         # we want to make sure that `select()` can be called when no
  778: [1253.2s]         # FD is registered.
  778: [1253.2s]         max_ev = max(len(self._fd_to_key), 1)
  778: [1253.2s]     
  778: [1253.2s]         ready = []
  778: [1253.2s]         try:
  778: [1253.2s] >           fd_event_list = self._selector.poll(timeout, max_ev)
  778: [1253.2s] E           Failed: Timeout >300.0s
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/selectors.py:469: Failed
  778: [1253.2s] ----------------------------- Captured stderr call -----------------------------
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] 
  778: [1253.2s] ~~~~~~~~~~~~~~~~ Stack of AnyIO worker thread (140280134059712) ~~~~~~~~~~~~~~~~
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 973, in _bootstrap
  778: [1253.2s]     self._bootstrap_inner()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
  778: [1253.2s]     self.run()
  778: [1253.2s]   File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 797, in run
  778: [1253.2s]     item = self.queue.get()
  778: [1253.2s]   File "/usr/local/lib/python3.10/queue.py", line 171, in get
  778: [1253.2s]     self.not_empty.wait()
  778: [1253.2s]   File "/usr/local/lib/python3.10/threading.py", line 320, in wait
  778: [1253.2s]     waiter.acquire()
  778: [1253.2s] 
  778: [1253.2s] +++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
  778: [1253.2s] ________________ TestFullRefresh.test_sequential_reads[inputs0] ________________
  778: [1253.2s] 
  778: [1253.2s] self = <connector_acceptance_test.tests.test_full_refresh.TestFullRefresh object at 0x7f9582b25c30>
  778: [1253.2s] connector_config = SecretDict(******)
  778: [1253.2s] configured_catalog = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='id_and_name', json_schema={'type'...'full_refresh'>, cursor_field=[], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=[['id']])])
  778: [1253.2s] ignored_fields = {}
  778: [1253.2s] docker_runner = <connector_acceptance_test.utils.connector_runner.ConnectorRunner object at 0x7f9582b25ff0>
  778: [1253.2s] detailed_logger = <Logger detailed_logger /test_input/acceptance_tests_logs/test_full_refresh.py__TestFullRefresh__test_sequential_reads[inputs0].txt (DEBUG)>
  778: [1253.2s] 
  778: [1253.2s]     async def test_sequential_reads(
  778: [1253.2s]         self,
  778: [1253.2s]         connector_config: SecretDict,
  778: [1253.2s]         configured_catalog: ConfiguredAirbyteCatalog,
  778: [1253.2s]         ignored_fields: Optional[Mapping[str, List[IgnoredFieldsConfiguration]]],
  778: [1253.2s]         docker_runner: ConnectorRunner,
  778: [1253.2s]         detailed_logger: Logger,
  778: [1253.2s]     ):
  778: [1253.2s] >       configured_catalog = await full_refresh_only_catalog(configured_catalog)
  778: [1253.2s] E       TypeError: object ConfiguredAirbyteCatalog can't be used in 'await' expression
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/connector_acceptance_test/tests/test_full_refresh.py:95: TypeError
  778: [1253.2s] ________________ TestFullRefresh.test_sequential_reads[inputs1] ________________
  778: [1253.2s] 
  778: [1253.2s] self = <connector_acceptance_test.tests.test_full_refresh.TestFullRefresh object at 0x7f9583a19600>
  778: [1253.2s] connector_config = SecretDict(******)
  778: [1253.2s] configured_catalog = ConfiguredAirbyteCatalog(streams=[ConfiguredAirbyteStream(stream=AirbyteStream(name='id_and_name', json_schema={'type'...'>, cursor_field=['_ab_cdc_lsn'], destination_sync_mode=<DestinationSyncMode.append: 'append'>, primary_key=[['id']])])
  778: [1253.2s] ignored_fields = {}
  778: [1253.2s] docker_runner = <connector_acceptance_test.utils.connector_runner.ConnectorRunner object at 0x7f9583a18d60>
  778: [1253.2s] detailed_logger = <Logger detailed_logger /test_input/acceptance_tests_logs/test_full_refresh.py__TestFullRefresh__test_sequential_reads[inputs1].txt (DEBUG)>
  778: [1253.2s] 
  778: [1253.2s]     async def test_sequential_reads(
  778: [1253.2s]         self,
  778: [1253.2s]         connector_config: SecretDict,
  778: [1253.2s]         configured_catalog: ConfiguredAirbyteCatalog,
  778: [1253.2s]         ignored_fields: Optional[Mapping[str, List[IgnoredFieldsConfiguration]]],
  778: [1253.2s]         docker_runner: ConnectorRunner,
  778: [1253.2s]         detailed_logger: Logger,
  778: [1253.2s]     ):
  778: [1253.2s] >       configured_catalog = await full_refresh_only_catalog(configured_catalog)
  778: [1253.2s] E       TypeError: object ConfiguredAirbyteCatalog can't be used in 'await' expression
  778: [1253.2s] 
  778: [1253.2s] /usr/local/lib/python3.10/site-packages/connector_acceptance_test/tests/test_full_refresh.py:95: TypeError
  778: [1253.3s] =============================== warnings summary ===============================
  778: [1253.3s] ../usr/local/lib/python3.10/site-packages/hypothesis_jsonschema/_canonicalise.py:116
  778: [1253.3s]   /usr/local/lib/python3.10/site-packages/hypothesis_jsonschema/_canonicalise.py:116: DeprecationWarning: jsonschema.exceptions.RefResolutionError is deprecated as of version 4.18.0. If you wish to catch potential reference resolution errors, directly catch referencing.exceptions.Unresolvable.
  778: [1253.3s]     class HypothesisRefResolutionError(jsonschema.exceptions.RefResolutionError):
  778: [1253.3s] 
  778: [1253.3s] ../usr/local/lib/python3.10/site-packages/hypothesis_jsonschema/_resolve.py:31
  778: [1253.3s]   /usr/local/lib/python3.10/site-packages/hypothesis_jsonschema/_resolve.py:31: DeprecationWarning: jsonschema.RefResolver is deprecated as of v4.18.0, in favor of the https://github.com/python-jsonschema/referencing library, which provides more compliant referencing behavior as well as more flexible APIs for customization. A future release will remove RefResolver. Please file a feature request (on referencing) if you are missing an API for the kind of customization you need.
  778: [1253.3s]     class LocalResolver(jsonschema.RefResolver):
  778: [1253.3s] 
  778: [1253.3s] -- Docs: https://docs.pytest.org/en/stable/warnings.html
  778: [1253.3s] =========================== short test summary info ============================
  778: [1253.3s] FAILED test_core.py::TestBasicRead::test_read[inputs0] - Failed: Timeout >300.0s
  778: [1253.3s] FAILED test_core.py::TestBasicRead::test_read[inputs1] - Failed: Timeout >300.0s
  778: [1253.3s] FAILED test_core.py::TestBasicRead::test_airbyte_trace_message_on_failure[inputs0]
  778: [1253.3s] FAILED test_core.py::TestBasicRead::test_airbyte_trace_message_on_failure[inputs1]
  778: [1253.3s] FAILED test_full_refresh.py::TestFullRefresh::test_sequential_reads[inputs0]
  778: [1253.3s] FAILED test_full_refresh.py::TestFullRefresh::test_sequential_reads[inputs1]
  778: [1253.3s] ERROR test_full_refresh.py::TestFullRefresh::test_sequential_reads[inputs1]
  778: [1253.3s] == 6 failed, 64 passed, 7 skipped, 2 warnings, 1 error in 1252.06s (0:20:52) ===
  778: exec python -m pytest -p connector_acceptance_test.plugin --suppress-tests-failed-exit-code --acceptance-test-config /test_input -p integration_tests.acceptance DONE

Running locally, the source-openweather test run correctly. ( The TypeError: object ConfiguredAirbyteCatalog can't be used in 'await' expression was a easily fixed)

[alafanechere]

I provided the Dagger team with access to source-openweather and source-postgres secrets for repro.

[evantahler]

Currently testing with nightly builds... PR review will be big... so... thanks :D

[alafanechere]

Moving this back to the sprint backlog as I want to focus on other task force issue and https://github.com/airbytehq/airbyte/pull/28656 fixed a lot of transient CAT failure.