Implement methods for managing networks

Signed-off-by: Aanand Prasad <[email protected]>

Author: aanand

docker/api/__init__.py

@@ -5,3 +5,4 @@
 from .exec_api import ExecApiMixin
 from .image import ImageApiMixin
 from .volume import VolumeApiMixin
+from .network import NetworkApiMixin

docker/api/network.py

@@ -0,0 +1,49 @@
+import json
+
+from ..utils import check_resource
+
+
+class NetworkApiMixin(object):
+    def networks(self, names=None, ids=None):
+        filters = {}
+        if names:
+            filters['name'] = names
+        if ids:
+            filters['id'] = ids
+
+        params = {'filters': json.dumps(filters)}
+
+        url = self._url("/networks")
+        res = self._get(url, params=params)
+        return self._result(res, json=True)
+
+    def create_network(self, name, driver=None):
+        data = {
+            'name': name,
+            'driver': driver,
+        }
+        url = self._url("/networks/create")
+        res = self._post_json(url, data=data)
+        return self._result(res, json=True)
+
+    def remove_network(self, net_id):
+        url = self._url("/networks/{0}", net_id)
+        res = self._delete(url)
+        self._raise_for_status(res)
+
+    def inspect_network(self, net_id):
+        url = self._url("/networks/{0}", net_id)
+        res = self._get(url)
+        return self._result(res, json=True)
+
+    @check_resource
+    def connect_container_to_network(self, container, net_id):
+        data = {"container": container}
+        url = self._url("/networks/{0}/connect", net_id)
+        self._post_json(url, data=data)
+
+    @check_resource
+    def disconnect_container_from_network(self, container, net_id):
+        data = {"container": container}
+        url = self._url("/networks/{0}/disconnect", net_id)
+        self._post_json(url, data=data)

docker/client.py

@@ -39,7 +39,8 @@ class Client(
         api.DaemonApiMixin,
         api.ExecApiMixin,
         api.ImageApiMixin,
-        api.VolumeApiMixin):
+        api.VolumeApiMixin,
+        api.NetworkApiMixin):
     def __init__(self, base_url=None, version=None,
                  timeout=constants.DEFAULT_TIMEOUT_SECONDS, tls=False):
         super(Client, self).__init__()

tests/integration_test.py

@@ -21,6 +21,7 @@
 import shutil
 import signal
 import socket
+import sys
 import tarfile
 import tempfile
 import threading
@@ -95,6 +96,7 @@ def setUp(self):
         self.tmp_containers = []
         self.tmp_folders = []
         self.tmp_volumes = []
+        self.tmp_networks = []
 
     def tearDown(self):
         for img in self.tmp_imgs:
@@ -108,6 +110,11 @@ def tearDown(self):
                 self.client.remove_container(container)
             except docker.errors.APIError:
                 pass
+        for network in self.tmp_networks:
+            try:
+                self.client.remove_network(network)
+            except docker.errors.APIError:
+                pass
         for folder in self.tmp_folders:
             shutil.rmtree(folder)
 
@@ -1590,6 +1597,103 @@ def runTest(self):
             ['not-ignored'],
         )
 
+
+#######################
+#    NETWORK TESTS    #
+#######################
+
+
+@requires_api_version('1.21')
+class TestNetworks(BaseTestCase):
+    def create_network(self, *args, **kwargs):
+        net_name = 'dockerpy{}'.format(random.randrange(sys.maxint))[:14]
+        net_id = self.client.create_network(net_name, *args, **kwargs)['id']
+        self.tmp_networks.append(net_id)
+        return (net_name, net_id)
+
+    def test_list_networks(self):
+        networks = self.client.networks()
+        initial_size = len(networks)
+
+        net_name, net_id = self.create_network()
+
+        networks = self.client.networks()
+        self.assertEqual(len(networks), initial_size + 1)
+        self.assertTrue(net_id in [n['id'] for n in networks])
+
+        networks_by_name = self.client.networks(names=[net_name])
+        self.assertEqual([n['id'] for n in networks_by_name], [net_id])
+
+        networks_by_partial_id = self.client.networks(ids=[net_id[:8]])
+        self.assertEqual([n['id'] for n in networks_by_partial_id], [net_id])
+
+    def test_inspect_network(self):
+        net_name, net_id = self.create_network()
+
+        net = self.client.inspect_network(net_id)
+        self.assertEqual(net, {
+            u'name': net_name,
+            u'id': net_id,
+            u'driver': 'bridge',
+            u'containers': {},
+        })
+
+    def test_create_network_with_host_driver_fails(self):
+        net_name = 'dockerpy{}'.format(random.randrange(sys.maxint))[:14]
+
+        with pytest.raises(APIError):
+            self.client.create_network(net_name, driver='host')
+
+    def test_remove_network(self):
+        initial_size = len(self.client.networks())
+
+        net_name, net_id = self.create_network()
+        self.assertEqual(len(self.client.networks()), initial_size + 1)
+
+        self.client.remove_network(net_id)
+        self.assertEqual(len(self.client.networks()), initial_size)
+
+    def test_connect_and_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)
+        network_data = self.client.inspect_network(net_id)
+        self.assertFalse(network_data.get('containers'))
+
+    def test_connect_on_container_create(self):
+        net_name, net_id = self.create_network()
+
+        container = self.client.create_container(
+            image='busybox',
+            command='top',
+            host_config=self.client.create_host_config(network_mode=net_name),
+        )
+        self.tmp_containers.append(container)
+        self.client.start(container)
+
+        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)
+        network_data = self.client.inspect_network(net_id)
+        self.assertFalse(network_data.get('containers'))
+
+
 #######################
 #  PY SPECIFIC TESTS  #
 #######################

tests/test.py

@@ -369,6 +369,41 @@ def test_list_containers(self):
             timeout=DEFAULT_TIMEOUT_SECONDS
         )
 
+    def test_list_networks(self):
+        networks = [
+            {
+                "name": "none",
+                "id": "8e4e55c6863ef424",
+                "type": "null",
+                "endpoints": []
+            },
+            {
+                "name": "host",
+                "id": "062b6d9ea7913fde",
+                "type": "host",
+                "endpoints": []
+            },
+        ]
+
+        get = mock.Mock(return_value=response(
+            status_code=200, content=json.dumps(networks).encode('utf-8')))
+
+        with mock.patch('docker.Client.get', get):
+            self.assertEqual(self.client.networks(), networks)
+
+            self.assertEqual(get.call_args[0][0], url_prefix + 'networks')
+
+            filters = json.loads(get.call_args[1]['params']['filters'])
+            self.assertFalse(filters)
+
+            self.client.networks(names=['foo'])
+            filters = json.loads(get.call_args[1]['params']['filters'])
+            self.assertEqual(filters, {'name': ['foo']})
+
+            self.client.networks(ids=['123'])
+            filters = json.loads(get.call_args[1]['params']['filters'])
+            self.assertEqual(filters, {'id': ['123']})
+
     #####################
     #  CONTAINER TESTS  #
     #####################
@@ -2229,6 +2264,105 @@ def test_remove_volume(self):
         self.assertEqual(args[0][0], 'DELETE')
         self.assertEqual(args[0][1], '{0}volumes/{1}'.format(url_prefix, name))
 
+    #####################
+    #   NETWORK TESTS   #
+    #####################
+
+    def test_create_network(self):
+        network_data = {
+            "id": 'abc12345',
+            "warning": "",
+        }
+
+        network_response = response(status_code=200, content=network_data)
+        post = mock.Mock(return_value=network_response)
+
+        with mock.patch('docker.Client.post', post):
+            result = self.client.create_network('foo')
+            self.assertEqual(result, network_data)
+
+            self.assertEqual(
+                post.call_args[0][0],
+                url_prefix + 'networks/create')
+
+            self.assertEqual(
+                json.loads(post.call_args[1]['data']),
+                {"name": "foo"})
+
+            self.client.create_network('foo', 'bridge')
+
+            self.assertEqual(
+                json.loads(post.call_args[1]['data']),
+                {"name": "foo", "driver": "bridge"})
+
+    def test_remove_network(self):
+        network_id = 'abc12345'
+        delete = mock.Mock(return_value=response(status_code=200))
+
+        with mock.patch('docker.Client.delete', delete):
+            self.client.remove_network(network_id)
+
+        args = delete.call_args
+        self.assertEqual(args[0][0],
+                         url_prefix + 'networks/{0}'.format(network_id))
+
+    def test_inspect_network(self):
+        network_id = 'abc12345'
+        network_name = 'foo'
+        network_data = {
+            six.u('name'): network_name,
+            six.u('id'): network_id,
+            six.u('driver'): 'bridge',
+            six.u('containers'): {},
+        }
+
+        network_response = response(status_code=200, content=network_data)
+        get = mock.Mock(return_value=network_response)
+
+        with mock.patch('docker.Client.get', get):
+            result = self.client.inspect_network(network_id)
+            self.assertEqual(result, network_data)
+
+        args = get.call_args
+        self.assertEqual(args[0][0],
+                         url_prefix + 'networks/{0}'.format(network_id))
+
+    def test_connect_container_to_network(self):
+        network_id = 'abc12345'
+        container_id = 'def45678'
+
+        post = mock.Mock(return_value=response(status_code=201))
+
+        with mock.patch('docker.Client.post', post):
+            self.client.connect_container_to_network(
+                {'Id': container_id}, network_id)
+
+        self.assertEqual(
+            post.call_args[0][0],
+            url_prefix + 'networks/{0}/connect'.format(network_id))
+
+        self.assertEqual(
+            json.loads(post.call_args[1]['data']),
+            {'container': container_id})
+
+    def test_disconnect_container_from_network(self):
+        network_id = 'abc12345'
+        container_id = 'def45678'
+
+        post = mock.Mock(return_value=response(status_code=201))
+
+        with mock.patch('docker.Client.post', post):
+            self.client.disconnect_container_from_network(
+                {'Id': container_id}, network_id)
+
+        self.assertEqual(
+            post.call_args[0][0],
+            url_prefix + 'networks/{0}/disconnect'.format(network_id))
+
+        self.assertEqual(
+            json.loads(post.call_args[1]['data']),
+            {'container': container_id})
+
     #######################
     #  PY SPECIFIC TESTS  #
     #######################