Merge "Add 'ip rule ...' support to ip_lib"

This commit is contained in:
Jenkins 2014-06-15 03:22:19 +00:00 committed by Gerrit Code Review
commit dbedfb7aba
2 changed files with 79 additions and 12 deletions

View File

@ -185,6 +185,18 @@ class IPWrapper(SubProcessBase):
return [l.strip() for l in output.split('\n')]
class IpRule(IPWrapper):
def add_rule_from(self, ip, table, rule_pr):
args = ['add', 'from', ip, 'lookup', table, 'priority', rule_pr]
ip = self._as_root('', 'rule', tuple(args))
return ip
def delete_rule_priority(self, rule_pr):
args = ['del', 'priority', rule_pr]
ip = self._as_root('', 'rule', tuple(args))
return ip
class IPDevice(SubProcessBase):
def __init__(self, name, root_helper=None, namespace=None):
super(IPDevice, self).__init__(root_helper=root_helper,
@ -362,20 +374,23 @@ class IpAddrCommand(IpDeviceCommandBase):
class IpRouteCommand(IpDeviceCommandBase):
COMMAND = 'route'
def add_gateway(self, gateway, metric=None):
def add_gateway(self, gateway, metric=None, table=None):
args = ['replace', 'default', 'via', gateway]
if metric:
args += ['metric', metric]
args += ['dev', self.name]
if table:
args += ['table', table]
self._as_root(*args)
def delete_gateway(self, gateway):
self._as_root('del',
'default',
'via',
gateway,
'dev',
self.name)
def delete_gateway(self, gateway=None, table=None):
args = ['del', 'default']
if gateway:
args += ['via', gateway]
args += ['dev', self.name]
if table:
args += ['table', table]
self._as_root(*args)
def list_onlink_routes(self):
def iterate_routes():
@ -456,6 +471,18 @@ class IpRouteCommand(IpDeviceCommandBase):
self._as_root('append', subnet, 'proto', 'kernel',
'dev', device)
def add_route(self, cidr, ip, table=None):
args = ['replace', cidr, 'via', ip, 'dev', self.name]
if table:
args += ['table', table]
self._as_root(*args)
def delete_route(self, cidr, ip, table=None):
args = ['del', cidr, 'via', ip, 'dev', self.name]
if table:
args += ['table', table]
self._as_root(*args)
class IpNeighCommand(IpDeviceCommandBase):
COMMAND = 'neigh'

View File

@ -406,6 +406,26 @@ class TestIpWrapper(base.BaseTestCase):
self.assertEqual(dev.mock_calls, [])
class TestIpRule(base.BaseTestCase):
def setUp(self):
super(TestIpRule, self).setUp()
self.execute_p = mock.patch.object(ip_lib.IpRule, '_execute')
self.execute = self.execute_p.start()
def test_add_rule_from(self):
ip_lib.IpRule('sudo').add_rule_from('192.168.45.100', 2, 100)
self.execute.assert_called_once_with('', 'rule',
('add', 'from', '192.168.45.100',
'lookup', 2, 'priority', 100),
'sudo', None)
def test_delete_rule_priority(self):
ip_lib.IpRule('sudo').delete_rule_priority(100)
self.execute.assert_called_once_with('', 'rule',
('del', 'priority', 100),
'sudo', None)
class TestIPDevice(base.BaseTestCase):
def test_eq_same_name(self):
dev1 = ip_lib.IPDevice('tap0')
@ -656,18 +676,20 @@ class TestIpRouteCommand(TestIPCmdBase):
def test_add_gateway(self):
gateway = '192.168.45.100'
metric = 100
self.route_cmd.add_gateway(gateway, metric)
table = 14
self.route_cmd.add_gateway(gateway, metric, table)
self._assert_sudo([],
('replace', 'default', 'via', gateway,
'metric', metric,
'dev', self.parent.name))
'dev', self.parent.name, 'table', table))
def test_del_gateway(self):
gateway = '192.168.45.100'
self.route_cmd.delete_gateway(gateway)
table = 14
self.route_cmd.delete_gateway(gateway, table)
self._assert_sudo([],
('del', 'default', 'via', gateway,
'dev', self.parent.name))
'dev', self.parent.name, 'table', table))
def test_get_gateway(self):
test_cases = [{'sample': GATEWAY_SAMPLE1,
@ -718,6 +740,24 @@ class TestIpRouteCommand(TestIPCmdBase):
# Check two calls - device get and subnet get
self.assertEqual(len(self.parent._run.mock_calls), 2)
def test_add_route(self):
cidr = '192.168.45.100/24'
ip = '10.0.0.1'
table = 14
self.route_cmd.add_route(cidr, ip, table)
self._assert_sudo([],
('replace', cidr, 'via', ip,
'dev', self.parent.name, 'table', table))
def test_delete_route(self):
cidr = '192.168.45.100/24'
ip = '10.0.0.1'
table = 14
self.route_cmd.delete_route(cidr, ip, table)
self._assert_sudo([],
('del', cidr, 'via', ip,
'dev', self.parent.name, 'table', table))
class TestIpNetnsCommand(TestIPCmdBase):
def setUp(self):