import docker from docker.types import IPAMConfig, IPAMPool import pytest from ..helpers import random_name, requires_api_version from .base import BaseAPIIntegrationTest, BUSYBOX class TestNetworks(BaseAPIIntegrationTest): def tearDown(self): super(TestNetworks, self).tearDown() self.client.leave_swarm(force=True) def create_network(self, *args, **kwargs): net_name = random_name() net_id = self.client.create_network(net_name, *args, **kwargs)['Id'] self.tmp_networks.append(net_id) return (net_name, net_id) @requires_api_version('1.21') def test_list_networks(self): networks = self.client.networks() net_name, net_id = self.create_network() networks = self.client.networks() 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]) @requires_api_version('1.21') def test_inspect_network(self): net_name, net_id = self.create_network() net = self.client.inspect_network(net_id) self.assertEqual(net['Id'], net_id) self.assertEqual(net['Name'], net_name) self.assertEqual(net['Driver'], 'bridge') self.assertEqual(net['Scope'], 'local') self.assertEqual(net['IPAM']['Driver'], 'default') @requires_api_version('1.21') def test_create_network_with_ipam_config(self): _, net_id = self.create_network( ipam=IPAMConfig( driver='default', pool_configs=[ IPAMPool( subnet="172.28.0.0/16", iprange="172.28.5.0/24", gateway="172.28.5.254", aux_addresses={ "a": "172.28.1.5", "b": "172.28.1.6", "c": "172.28.1.7", }, ), ], ), ) net = self.client.inspect_network(net_id) ipam = net['IPAM'] assert ipam.pop('Options', None) is None assert ipam['Driver'] == 'default' assert ipam['Config'] == [{ 'Subnet': "172.28.0.0/16", 'IPRange': "172.28.5.0/24", 'Gateway': "172.28.5.254", 'AuxiliaryAddresses': { "a": "172.28.1.5", "b": "172.28.1.6", "c": "172.28.1.7", }, }] @requires_api_version('1.21') def test_create_network_with_host_driver_fails(self): with pytest.raises(docker.errors.APIError): self.client.create_network(random_name(), driver='host') @requires_api_version('1.21') 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) @requires_api_version('1.21') 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']] ) with pytest.raises(docker.errors.APIError): self.client.connect_container_to_network(container, net_id) self.client.disconnect_container_from_network(container, net_id) 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) @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() container = self.client.create_container(BUSYBOX, 'top') self.tmp_containers.append(container) self.client.start(container) self.client.connect_container_to_network( container, net_id, aliases=['foo', 'bar']) container_data = self.client.inspect_container(container) aliases = ( container_data['NetworkSettings']['Networks'][net_name]['Aliases'] ) assert 'foo' in aliases assert 'bar' in aliases @requires_api_version('1.21') 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')) @requires_api_version('1.22') def test_create_with_aliases(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, ), networking_config=self.client.create_networking_config({ net_name: self.client.create_endpoint_config( aliases=['foo', 'bar'], ), }), ) self.tmp_containers.append(container) self.client.start(container) container_data = self.client.inspect_container(container) aliases = ( container_data['NetworkSettings']['Networks'][net_name]['Aliases'] ) assert 'foo' in aliases assert 'bar' in aliases @requires_api_version('1.22') def test_create_with_ipv4_address(self): net_name, net_id = self.create_network( ipam=IPAMConfig( driver='default', pool_configs=[IPAMPool(subnet="132.124.0.0/16")], ), ) container = self.client.create_container( image=BUSYBOX, command='top', host_config=self.client.create_host_config(network_mode=net_name), networking_config=self.client.create_networking_config({ net_name: self.client.create_endpoint_config( ipv4_address='132.124.0.23' ) }) ) self.tmp_containers.append(container) self.client.start(container) container_data = self.client.inspect_container(container) self.assertEqual( container_data[ 'NetworkSettings']['Networks'][net_name]['IPAMConfig'][ 'IPv4Address' ], '132.124.0.23' ) @requires_api_version('1.22') def test_create_with_ipv6_address(self): net_name, net_id = self.create_network( ipam=IPAMConfig( driver='default', pool_configs=[IPAMPool(subnet="2001:389::1/64")], ), ) container = self.client.create_container( image=BUSYBOX, command='top', host_config=self.client.create_host_config(network_mode=net_name), networking_config=self.client.create_networking_config({ net_name: self.client.create_endpoint_config( ipv6_address='2001:389::f00d' ) }) ) self.tmp_containers.append(container) self.client.start(container) container_data = self.client.inspect_container(container) self.assertEqual( container_data[ 'NetworkSettings']['Networks'][net_name]['IPAMConfig'][ 'IPv6Address' ], '2001:389::f00d' ) @requires_api_version('1.24') def test_create_with_linklocal_ips(self): container = self.client.create_container( BUSYBOX, 'top', networking_config=self.client.create_networking_config( { 'bridge': self.client.create_endpoint_config( link_local_ips=['169.254.8.8'] ) } ), host_config=self.client.create_host_config(network_mode='bridge') ) self.tmp_containers.append(container) self.client.start(container) container_data = self.client.inspect_container(container) net_cfg = container_data['NetworkSettings']['Networks']['bridge'] assert 'IPAMConfig' in net_cfg assert 'LinkLocalIPs' in net_cfg['IPAMConfig'] assert net_cfg['IPAMConfig']['LinkLocalIPs'] == ['169.254.8.8'] @requires_api_version('1.22') def test_create_with_links(self): net_name, net_id = self.create_network() container = self.create_and_start( host_config=self.client.create_host_config(network_mode=net_name), networking_config=self.client.create_networking_config({ net_name: self.client.create_endpoint_config( links=[('docker-py-test-upstream', 'bar')], ), }), ) container_data = self.client.inspect_container(container) self.assertEqual( container_data['NetworkSettings']['Networks'][net_name]['Links'], ['docker-py-test-upstream:bar']) self.create_and_start( name='docker-py-test-upstream', host_config=self.client.create_host_config(network_mode=net_name), ) self.execute(container, ['nslookup', 'bar']) @requires_api_version('1.21') def test_create_check_duplicate(self): net_name, net_id = self.create_network() with self.assertRaises(docker.errors.APIError): self.client.create_network(net_name, check_duplicate=True) net_id = self.client.create_network(net_name, check_duplicate=False) self.tmp_networks.append(net_id['Id']) @requires_api_version('1.22') def test_connect_with_links(self): net_name, net_id = self.create_network() container = self.create_and_start( host_config=self.client.create_host_config(network_mode=net_name)) self.client.disconnect_container_from_network(container, net_name) self.client.connect_container_to_network( container, net_name, links=[('docker-py-test-upstream', 'bar')]) container_data = self.client.inspect_container(container) self.assertEqual( container_data['NetworkSettings']['Networks'][net_name]['Links'], ['docker-py-test-upstream:bar']) self.create_and_start( name='docker-py-test-upstream', host_config=self.client.create_host_config(network_mode=net_name), ) self.execute(container, ['nslookup', 'bar']) @requires_api_version('1.22') def test_connect_with_ipv4_address(self): net_name, net_id = self.create_network( ipam=IPAMConfig( driver='default', pool_configs=[ IPAMPool( subnet="172.28.0.0/16", iprange="172.28.5.0/24", gateway="172.28.5.254" ) ] ) ) container = self.create_and_start( host_config=self.client.create_host_config(network_mode=net_name)) self.client.disconnect_container_from_network(container, net_name) self.client.connect_container_to_network( container, net_name, ipv4_address='172.28.5.24' ) container_data = self.client.inspect_container(container) net_data = container_data['NetworkSettings']['Networks'][net_name] self.assertEqual( net_data['IPAMConfig']['IPv4Address'], '172.28.5.24' ) @requires_api_version('1.22') def test_connect_with_ipv6_address(self): net_name, net_id = self.create_network( ipam=IPAMConfig( driver='default', pool_configs=[ IPAMPool( subnet="2001:389::1/64", iprange="2001:389::0/96", gateway="2001:389::ffff" ) ] ) ) container = self.create_and_start( host_config=self.client.create_host_config(network_mode=net_name)) self.client.disconnect_container_from_network(container, net_name) self.client.connect_container_to_network( container, net_name, ipv6_address='2001:389::f00d' ) container_data = self.client.inspect_container(container) net_data = container_data['NetworkSettings']['Networks'][net_name] self.assertEqual( net_data['IPAMConfig']['IPv6Address'], '2001:389::f00d' ) @requires_api_version('1.23') def test_create_internal_networks(self): _, net_id = self.create_network(internal=True) net = self.client.inspect_network(net_id) assert net['Internal'] is True @requires_api_version('1.23') def test_create_network_with_labels(self): _, net_id = self.create_network(labels={ 'com.docker.py.test': 'label' }) net = self.client.inspect_network(net_id) assert 'Labels' in net assert len(net['Labels']) == 1 assert net['Labels'] == { 'com.docker.py.test': 'label' } @requires_api_version('1.23') def test_create_network_with_labels_wrong_type(self): with pytest.raises(TypeError): self.create_network(labels=['com.docker.py.test=label', ]) @requires_api_version('1.23') def test_create_network_ipv6_enabled(self): _, net_id = self.create_network( enable_ipv6=True, ipam=IPAMConfig( driver='default', pool_configs=[ IPAMPool( subnet="2001:389::1/64", iprange="2001:389::0/96", gateway="2001:389::ffff" ) ] ) ) net = self.client.inspect_network(net_id) assert net['EnableIPv6'] is True @requires_api_version('1.25') def test_create_network_attachable(self): assert self.client.init_swarm('eth0') _, net_id = self.create_network(driver='overlay', attachable=True) net = self.client.inspect_network(net_id) assert net['Attachable'] is True @requires_api_version('1.25') def test_prune_networks(self): net_name, _ = self.create_network() result = self.client.prune_networks() assert net_name in result['NetworksDeleted']