exec: fix file handle leak with container.exec_* APIs (https://github.com/docker/docker-py/pull/2320) (#582)

Requests with stream=True MUST be closed or else the connection will
never be returned to the connection pool. Both ContainerApiMixin.attach
and ExecApiMixin.exec_start were leaking in the stream=False case.
exec_start was modified to follow attach for the stream=True case as
that allows the caller to close the stream when done (untested).

Tested with:

    # Test exec_run (stream=False) - observe one less leak
    make integration-test-py3 file=models_containers_test.py' -k test_exec_run_success -vs -W error::ResourceWarning'
    # Test exec_start (stream=True, fully reads from CancellableStream)
    make integration-test-py3 file=api_exec_test.py' -k test_execute_command -vs -W error::ResourceWarning'

After this change, one resource leak is removed, the remaining resource
leaks occur because none of the tests call client.close().

Fixes https://github.com/docker/docker-py/issues/1293
(Regression from https://github.com/docker/docker-py/pull/1130)

Cherry-picked from 34e6829dd4

Co-authored-by: Peter Wu <pwu@cloudflare.com>
Co-authored-by: Milas Bowman <milas.bowman@docker.com>

changelogs/fragments/582-stream-close.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - "docker_api connection plugin, docker_container_exec, docker_container_copy_into - properly close socket to Daemon after executing commands in containers (https://github.com/ansible-collections/community.docker/pull/582)."

plugins/module_utils/_api/api/client.py

@@ -394,6 +394,10 @@ class APIClient(
             yield out
 
     def _read_from_socket(self, response, stream, tty=True, demux=False):
+        """Consume all data from the socket, close the response and return the
+        data. If stream=True, then a generator is returned instead and the
+        caller is responsible for closing the response.
+        """
         socket = self._get_raw_response_socket(response)
 
         gen = frames_iter(socket, tty)
@@ -408,8 +412,11 @@ class APIClient(
         if stream:
             return gen
         else:
-            # Wait for all the frames, concatenate them, and return the result
-            return consume_socket_output(gen, demux=demux)
+            try:
+                # Wait for all the frames, concatenate them, and return the result
+                return consume_socket_output(gen, demux=demux)
+            finally:
+                response.close()
 
     def _disable_socket_timeout(self, socket):
         """ Depending on the combination of python version and whether we're