Add support for force disconnect

Signed-off-by: Joffrey F <[email protected]>

Author: shin-

docker/api/network.py

@@ -93,8 +93,15 @@ def connect_container_to_network(self, container, net_id,
 
     @check_resource
     @minimum_version('1.21')
-    def disconnect_container_from_network(self, container, net_id):
-        data = {"container": container}
+    def disconnect_container_from_network(self, container, net_id,
+                                          force=False):
+        data = {"Container": container}
+        if force:
+            if version_lt(self._version, '1.22'):
+                raise InvalidVersion(
+                    'Forced disconnect was introduced in API 1.22'
+                )
+            data['Force'] = force
         url = self._url("/networks/{0}/disconnect", net_id)
         res = self._post_json(url, data=data)
         self._raise_for_status(res)

docs/api.md

@@ -355,6 +355,8 @@ Inspect changes on a container's filesystem.
 
 * container (str): container-id/name to be disconnected from a network
 * net_id (str): network id
+* force (bool): Force the container to disconnect from a network.
+  Default: `False`
 
 ## events
 

tests/integration/network_test.py

@@ -115,7 +115,8 @@ def test_connect_and_disconnect_container(self):
         network_data = self.client.inspect_network(net_id)
         self.assertEqual(
             list(network_data['Containers'].keys()),
-            [container['Id']])
+            [container['Id']]
+        )
 
         with pytest.raises(docker.errors.APIError):
             self.client.connect_container_to_network(container, net_id)
@@ -127,6 +128,33 @@ def test_connect_and_disconnect_container(self):
         with pytest.raises(docker.errors.APIError):
             self.client.disconnect_container_from_network(container, net_id)
 
+    @requires_api_version('1.22')
+    def test_connect_and_force_disconnect_container(self):
+        net_name, net_id = self.create_network()
+
+        container = self.client.create_container('busybox', 'top')
+        self.tmp_containers.append(container)
+        self.client.start(container)
+
+        network_data = self.client.inspect_network(net_id)
+        self.assertFalse(network_data.get('Containers'))
+
+        self.client.connect_container_to_network(container, net_id)
+        network_data = self.client.inspect_network(net_id)
+        self.assertEqual(
+            list(network_data['Containers'].keys()),
+            [container['Id']]
+        )
+
+        self.client.disconnect_container_from_network(container, net_id, True)
+        network_data = self.client.inspect_network(net_id)
+        self.assertFalse(network_data.get('Containers'))
+
+        with pytest.raises(docker.errors.APIError):
+            self.client.disconnect_container_from_network(
+                container, net_id, force=True
+            )
+
     @requires_api_version('1.22')
     def test_connect_with_aliases(self):
         net_name, net_id = self.create_network()

tests/unit/network_test.py

@@ -184,4 +184,4 @@ def test_disconnect_container_from_network(self):
 
         self.assertEqual(
             json.loads(post.call_args[1]['data']),
-            {'container': container_id})
+            {'Container': container_id})