Accounting for asynchronous service removals

Makefile

@@ -35,11 +35,11 @@ unit-test-py3: build-py3
 
 .PHONY: integration-test
 integration-test: build
-	docker run -t --rm -v /var/run/docker.sock:/var/run/docker.sock docker-sdk-python py.test -v tests/integration/${file}
+	docker run -t --rm -v /var/run/docker.sock:/var/run/docker.sock docker-sdk-python py.test -v tests/integration/${file} ${pytest_options}
 
 .PHONY: integration-test-py3
 integration-test-py3: build-py3
-	docker run -t --rm -v /var/run/docker.sock:/var/run/docker.sock docker-sdk-python3 py.test tests/integration/${file}
+	docker run -t --rm -v /var/run/docker.sock:/var/run/docker.sock docker-sdk-python3 py.test tests/integration/${file} ${pytest_options}
 
 TEST_API_VERSION ?= 1.35
 TEST_ENGINE_VERSION ?= 17.12.0-ce

tests/helpers.py

@@ -77,14 +77,6 @@ def wrapped(self, *args, **kwargs):
     return req_exp
 
 
-def wait_on_condition(condition, delay=0.1, timeout=40):
-    start_time = time.time()
-    while not condition():
-        if time.time() - start_time > timeout:
-            raise AssertionError("Timeout: %s" % condition)
-        time.sleep(delay)
-
-
 def random_name():
     return u'dockerpytest_{0:x}'.format(random.getrandbits(64))
 
@@ -104,6 +96,16 @@ def force_leave_swarm(client):
                 return
 
 
+def wait_until_truthy(f, args=[], attempts=20, interval=0.5):
+    """Runs `f` with `args` until it returns a truthy value, running it up to
+    `attempts` times and sleeping `interval` seconds between attempts."""
+    for _ in range(attempts):
+        result = f(*args)
+        if result:
+            return result
+        time.sleep(interval)
+
+
 def swarm_listen_addr():
     return '0.0.0.0:{0}'.format(random.randrange(10000, 25000))
 

tests/integration/api_healthcheck_test.py

@@ -8,7 +8,9 @@ def wait_on_health_status(client, container, status):
     def condition():
         res = client.inspect_container(container)
         return res['State']['Health']['Status'] == status
-    return helpers.wait_on_condition(condition)
+    if not helpers.wait_until_truthy(condition, attempts=400, interval=0.1):
+        raise AssertionError('Timed out waiting for %s to get to status %s'
+                             % (container, status))
 
 
 class HealthcheckTest(BaseAPIIntegrationTest):

tests/integration/api_service_test.py

@@ -1,14 +1,14 @@
 # -*- coding: utf-8 -*-
 
 import random
-import time
 
 import docker
 import pytest
 import six
 
 from ..helpers import (
-    force_leave_swarm, requires_api_version, requires_experimental
+    force_leave_swarm, requires_api_version,
+    requires_experimental, wait_until_truthy
 )
 from .base import BaseAPIIntegrationTest, BUSYBOX
 
@@ -26,32 +26,36 @@ def teardown_class(cls):
         force_leave_swarm(client)
 
     def tearDown(self):
-        for service in self.client.services(filters={'name': 'dockerpytest_'}):
+        services = self.client.services(filters={'name': 'dockerpytest_'})
+        service_ids = [service['ID'] for service in services]
+
+        for service_id in service_ids:
             try:
-                self.client.remove_service(service['ID'])
+                self.client.remove_service(service_id)
             except docker.errors.APIError:
+                # possible engine issues are not this repo's concern, let's
+                # ignore those
                 pass
+
+        self._wait_for_services_removal(*service_ids)
+
         super(ServiceTest, self).tearDown()
 
     def get_service_name(self):
         return 'dockerpytest_{0:x}'.format(random.getrandbits(64))
 
-    def get_service_container(self, service_name, attempts=20, interval=0.5,
-                              include_stopped=False):
+    def get_service_container(self, service_name, include_stopped=False,
+                              **wait_options):
         # There is some delay between the service's creation and the creation
         # of the service's containers. This method deals with the uncertainty
         # when trying to retrieve the container associated with a service.
-        while True:
-            containers = self.client.containers(
+        containers = wait_until_truthy(
+            lambda: self.client.containers(
                 filters={'name': [service_name]}, quiet=True,
                 all=include_stopped
-            )
-            if len(containers) > 0:
-                return containers[0]
-            attempts -= 1
-            if attempts <= 0:
-                return None
-            time.sleep(interval)
+            ), **wait_options)
+        if containers:
+            return containers[0]
 
     def create_simple_service(self, name=None, labels=None):
         if name:
@@ -114,12 +118,14 @@ def test_inspect_service_insert_defaults(self):
     def test_remove_service_by_id(self):
         svc_name, svc_id = self.create_simple_service()
         assert self.client.remove_service(svc_id)
+        self._wait_for_services_removal(svc_id)
         test_services = self.client.services(filters={'name': 'dockerpytest_'})
         assert len(test_services) == 0
 
     def test_remove_service_by_name(self):
         svc_name, svc_id = self.create_simple_service()
         assert self.client.remove_service(svc_name)
+        self._wait_for_services_removal(svc_name)
         test_services = self.client.services(filters={'name': 'dockerpytest_'})
         assert len(test_services) == 0
 
@@ -135,20 +141,15 @@ def test_create_service_simple(self):
     def test_service_logs(self):
         name, svc_id = self.create_simple_service()
         assert self.get_service_container(name, include_stopped=True)
-        attempts = 20
-        while True:
-            if attempts == 0:
-                self.fail('No service logs produced by endpoint')
-                return
+
+        def fetch_service_logs():
             logs = self.client.service_logs(svc_id, stdout=True, is_tty=False)
             try:
-                log_line = next(logs)
+                return next(logs)
             except StopIteration:
-                attempts -= 1
-                time.sleep(0.1)
-                continue
-            else:
-                break
+                pass
+
+        log_line = wait_until_truthy(fetch_service_logs, interval=0.1)
 
         if six.PY3:
             log_line = log_line.decode('utf-8')
@@ -1288,3 +1289,24 @@ def _update_service(self, svc_id, *args, **kwargs):
                 self.client.update_service(*args, **kwargs)
             else:
                 raise
+
+    # service removal is async, in the sense that services only get
+    # properly deleted once all of their containers have properly shut down
+    # this function polls until the services are actually deleted
+    def _wait_for_services_removal(self, *svc_ids):
+        def service_doesnt_exist(svc_id):
+            try:
+                svc_info = self.client.inspect_service(svc_id)
+                # the service is not removed yet, but the engine should at
+                # least have marked it for removal
+                assert svc_info['PendingDelete']
+
+                return False
+            except docker.errors.NotFound:
+                return True
+            except docker.errors.APIError:
+                # engine error, retry
+                pass
+
+        for svc_id in svc_ids:
+            wait_until_truthy(service_doesnt_exist, args=[svc_id], attempts=40)