openstack-attic/melange
Code Issues Proposed changes

CLI commands for shared ips

Browse Source
This commit is contained in:
Rajaram Mallya 2011-11-21 10:23:33 +05:30
parent 643e2c30a0
commit 8535ff4a8d
3 changed files with 123 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
bin/melange-client
View File

@ -124,6 +124,7 @@ categories = {
'ip_route': client.IpRouteClient,
'interface': client.InterfaceClient,
'mac_address_range': client.MacAddressRangeClient,
'allowed_ip': client.AllowedIpClient,
}

28
melange/ipam/client.py
View File

@ -303,3 +303,31 @@ class MacAddressRangeClient(BaseClient):
def create(self, cidr):
return self._resource.create(cidr=cidr)
class AllowedIpClient(BaseClient):
def __init__(self, client, auth_client, tenant_id=None):
self.client = client
self.auth_client = auth_client
self.tenant_id = tenant_id
def _resource(self, interface_id):
return Resource("interfaces/{0}/allowed_ips".format(interface_id),
"allowed_ip",
self.client,
self.auth_client,
self.tenant_id)
def create(self, interface_id, network_id, ip_address):
return self._resource(interface_id).create(network_id=network_id,
ip_address=ip_address)
def show(self, interface_id, ip_address):
return self._resource(interface_id).find(ip_address)
def list(self, interface_id):
return self._resource(interface_id).all()
def delete(self, interface_id, ip_address):
return self._resource(interface_id).delete(ip_address)

113
melange/tests/functional/test_cli.py
View File

@ -26,11 +26,18 @@ from melange.tests import functional
def run(command):
return functional.execute("{0} --port={1} {2} --auth-token=test".format(
return functional.execute("{0} --port={1} {2} -v --auth-token=test".format(
melange.melange_bin_path('melange-client'),
functional.get_api_port(), command))
def run_melange_manage(command):
melange_manage = melange.melange_bin_path('melange-manage')
config_file = functional.test_config_file()
return functional.execute("%(melange_manage)s %(command)s "
"--config-file=%(config_file)s" % locals())
class TestIpBlockCLI(tests.BaseTest):
def test_create(self):
@ -494,6 +501,92 @@ class TestInterfaceCLI(tests.BaseTest):
self.assertTrue(models.IpAddress.get(ip2.id).marked_for_deallocation)
class TestMacAddressRangeCLI(tests.BaseTest):
def test_create(self):
exitcode, out, err = run("mac_address_range create "
"ab-bc-cd-12-23-34/24")
self.assertEqual(exitcode, 0)
self.assertIsNotNone(models.MacAddressRange.get_by(
cidr="ab-bc-cd-12-23-34/24"))
class TestAllowedIpCLI(tests.BaseTest):
def test_create(self):
interface = factory_models.InterfaceFactory(network_id="123",
virtual_interface_id="x",
tenant_id="RAX")
block = factory_models.IpBlockFactory(network_id="123",
tenant_id="RAX")
ip_plugged_into_interface = block.allocate_ip(interface)
ip_to_allow = block.allocate_ip(
factory_models.InterfaceFactory(network_id="123"))
exitcode, out, err = run("allowed_ip create "
"%(iface)s %(net)s %(ip)s -t RAX"
% {'iface': interface.virtual_interface_id,
'net': "123",
'ip': ip_to_allow.address})
self.assertEqual(exitcode, 0)
self.assertModelsEqual(interface.ips_allowed(),
[ip_plugged_into_interface, ip_to_allow])
def test_index(self):
interface = factory_models.InterfaceFactory(
tenant_id="RAX", virtual_interface_id="vif_id")
ip_factory = factory_models.IpAddressFactory
block_factory = factory_models.IpBlockFactory
ip_on_interface = block_factory(network_id="1",
tenant_id="RAX").allocate_ip(interface)
allowed_ip = ip_factory(ip_block_id=block_factory(network_id="1").id)
interface.allow_ip(allowed_ip)
exitcode, out, err = run("allowed_ip list "
"%(iface)s -t RAX"
% {'iface': interface.virtual_interface_id})
self.assertEqual(exitcode, 0)
self.assertIn("ip_addresses", out)
self.assertIn('"address": "%s"' % ip_on_interface.address, out)
self.assertIn('"address": "%s"' % allowed_ip.address, out)
def test_show(self):
interface = factory_models.InterfaceFactory(
tenant_id="RAX", virtual_interface_id="vif_id")
block = factory_models.IpBlockFactory(network_id="net123",
tenant_id="RAX")
ip_on_interface = block.allocate_ip(interface)
exitcode, out, err = run("allowed_ip show "
"%(iface)s %(ip)s -t RAX"
% {'iface': interface.virtual_interface_id,
'ip': ip_on_interface.address})
self.assertEqual(exitcode, 0)
self.assertIn("ip_address", out)
self.assertIn('"address": "%s"' % ip_on_interface.address, out)
def test_delete(self):
interface = factory_models.InterfaceFactory(
tenant_id="RAX", virtual_interface_id="vif_id")
block = factory_models.IpBlockFactory(network_id="net123",
tenant_id="RAX")
ip_on_interface = block.allocate_ip(interface)
allowed_ip = block.allocate_ip(factory_models.InterfaceFactory())
interface.allow_ip(allowed_ip)
exitcode, out, err = run("allowed_ip delete "
"%(iface)s %(ip)s -t RAX"
% {'iface': interface.virtual_interface_id,
'ip': allowed_ip.address})
self.assertEqual(exitcode, 0)
self.assertEqual(interface.ips_allowed(), [ip_on_interface])
class TestDBSyncCLI(tests.BaseTest):
def test_db_sync_executes(self):
@ -529,21 +622,3 @@ class TestDeleteDeallocatedIps(tests.BaseTest):
deallocated_ip = models.IpAddress.find(ip.id)
new_deallocated_date = deallocated_ip.deallocated_at - days_to_subtract
deallocated_ip.update(deallocated_at=(new_deallocated_date))
class TestMacAddressRangeCLI(tests.BaseTest):
def test_create(self):
exitcode, out, err = run("mac_address_range create "
"ab-bc-cd-12-23-34/24")
self.assertEqual(exitcode, 0)
self.assertIsNotNone(models.MacAddressRange.get_by(
cidr="ab-bc-cd-12-23-34/24"))
def run_melange_manage(command):
melange_manage = melange.melange_bin_path('melange-manage')
config_file = functional.test_config_file()
return functional.execute("%(melange_manage)s %(command)s "
"--config-file=%(config_file)s" % locals())