eventlet.monkey_patch()
import netaddr
-import os
from oslo.config import cfg
from oslo import messaging
from oslo.utils import excutils
from neutron.agent.linux import interface
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ra
+from neutron.agent.metadata import driver as metadata_driver
from neutron.agent import rpc as agent_rpc
from neutron.common import config as common_config
from neutron.common import constants as l3_constants
self.target_ex_net_id = None
self.use_ipv6 = ipv6_utils.is_enabled()
+ if self.conf.enable_metadata_proxy:
+ driver = metadata_driver.MetadataDriver.instance(self)
+ self.event_observers.add(driver)
+
def _check_config_params(self):
"""Check items in configuration files.
def _destroy_router_namespace(self, ns):
router_id = self.get_router_id(ns)
ra.disable_ipv6_ra(router_id, ns, self.root_helper)
- if self.conf.enable_metadata_proxy:
- self._destroy_metadata_proxy(router_id, ns)
ns_ip = ip_lib.IPWrapper(self.root_helper, namespace=ns)
for d in ns_ip.get_devices(exclude_loopback=True):
if d.name.startswith(INTERNAL_DEV_PREFIX):
self.router_info[router_id] = ri
if self.conf.use_namespaces:
self._create_router_namespace(ri)
- for c, r in self.metadata_filter_rules():
- ri.iptables_manager.ipv4['filter'].add_rule(c, r)
- for c, r in self.metadata_nat_rules():
- ri.iptables_manager.ipv4['nat'].add_rule(c, r)
- ri.iptables_manager.apply()
self.process_router_add(ri)
if ri.is_ha:
self.process_ha_router_added(ri)
- if self.conf.enable_metadata_proxy:
- if ri.is_ha:
- self._add_keepalived_notifiers(ri)
- else:
- self._spawn_metadata_proxy(ri.router_id, ri.ns_name)
-
def _router_removed(self, router_id):
ri = self.router_info.get(router_id)
if ri is None:
ri.router[l3_constants.INTERFACE_KEY] = []
ri.router[l3_constants.FLOATINGIP_KEY] = []
self.process_router(ri)
- for c, r in self.metadata_filter_rules():
- ri.iptables_manager.ipv4['filter'].remove_rule(c, r)
- for c, r in self.metadata_nat_rules():
- ri.iptables_manager.ipv4['nat'].remove_rule(c, r)
- ri.iptables_manager.apply()
del self.router_info[router_id]
self._destroy_router_namespace(ri.ns_name)
self.event_observers.notify(
adv_svc.AdvancedService.after_router_removed, ri)
- def _get_metadata_proxy_user_group(self):
- user = self.conf.metadata_proxy_user or os.geteuid()
- group = self.conf.metadata_proxy_group or os.getegid()
- return user, group
-
- def _get_metadata_proxy_callback(self, router_id):
-
- def callback(pid_file):
- metadata_proxy_socket = self.conf.metadata_proxy_socket
- user, group = self._get_metadata_proxy_user_group()
- proxy_cmd = ['neutron-ns-metadata-proxy',
- '--pid_file=%s' % pid_file,
- '--metadata_proxy_socket=%s' % metadata_proxy_socket,
- '--router_id=%s' % router_id,
- '--state_path=%s' % self.conf.state_path,
- '--metadata_port=%s' % self.conf.metadata_port,
- '--metadata_proxy_user=%s' % user,
- '--metadata_proxy_group=%s' % group]
- proxy_cmd.extend(config.get_log_args(
- self.conf, 'neutron-ns-metadata-proxy-%s.log' %
- router_id))
- return proxy_cmd
-
- return callback
-
- def _get_metadata_proxy_process_manager(self, router_id, ns_name):
- return external_process.ProcessManager(
- self.conf,
- router_id,
- self.root_helper,
- ns_name)
-
- def _spawn_metadata_proxy(self, router_id, ns_name):
- callback = self._get_metadata_proxy_callback(router_id)
- pm = self._get_metadata_proxy_process_manager(router_id, ns_name)
- pm.enable(callback)
-
- def _destroy_metadata_proxy(self, router_id, ns_name):
- pm = self._get_metadata_proxy_process_manager(router_id, ns_name)
- pm.disable()
-
def _set_subnet_info(self, port):
ips = port['fixed_ips']
if not ips:
if ri.router['distributed']:
self._destroy_snat_namespace(ns_name)
- def metadata_filter_rules(self):
- rules = []
- if self.conf.enable_metadata_proxy:
- rules.append(('INPUT', '-s 0.0.0.0/0 -d 127.0.0.1 '
- '-p tcp -m tcp --dport %s '
- '-j ACCEPT' % self.conf.metadata_port))
- return rules
-
- def metadata_nat_rules(self):
- rules = []
- if self.conf.enable_metadata_proxy:
- rules.append(('PREROUTING', '-s 0.0.0.0/0 -d 169.254.169.254/32 '
- '-p tcp -m tcp --dport 80 -j REDIRECT '
- '--to-port %s' % self.conf.metadata_port))
- return rules
-
def external_gateway_nat_rules(self, ex_gw_ip, interface_name):
rules = [('POSTROUTING', '! -i %(interface_name)s '
'! -o %(interface_name)s -m conntrack ! '
from oslo.config import cfg
from neutron.agent.linux import keepalived
+from neutron.agent.metadata import driver as metadata_driver
from neutron.common import constants as l3_constants
from neutron.i18n import _LE
from neutron.openstack.common import log as logging
ri.ha_port = ha_port
self._init_keepalived_manager(ri)
+ self._add_keepalived_notifiers(ri)
def process_ha_router_removed(self, ri):
self.ha_network_removed(ri)
instance.remove_vips_vroutes_by_interface(interface)
def _add_keepalived_notifiers(self, ri):
- callback = self._get_metadata_proxy_callback(ri.router_id)
- pm = self._get_metadata_proxy_process_manager(ri.router_id, ri.ns_name)
+ callback = (
+ metadata_driver.MetadataDriver._get_metadata_proxy_callback(
+ ri.router_id, self.conf))
+ pm = (
+ metadata_driver.MetadataDriver.
+ _get_metadata_proxy_process_manager(ri.router_id,
+ ri.ns_name,
+ self.conf))
pid = pm.get_pid_file_name(ensure_pids_dir=True)
ri.keepalived_manager.add_notifier(
callback(pid), 'master', ri.ha_vr_id)
--- /dev/null
+# Copyright 2014 OpenStack Foundation.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import os
+
+from neutron.agent.common import config
+from neutron.agent.linux import external_process
+from neutron.openstack.common import log as logging
+from neutron.services import advanced_service
+
+LOG = logging.getLogger(__name__)
+
+
+class MetadataDriver(advanced_service.AdvancedService):
+ def __init__(self, l3_agent):
+ super(MetadataDriver, self).__init__(l3_agent)
+ self.metadata_port = l3_agent.conf.metadata_port
+
+ def after_router_added(self, router):
+ for c, r in self.metadata_filter_rules(self.metadata_port):
+ router.iptables_manager.ipv4['filter'].add_rule(c, r)
+ for c, r in self.metadata_nat_rules(self.metadata_port):
+ router.iptables_manager.ipv4['nat'].add_rule(c, r)
+ router.iptables_manager.apply()
+
+ if not router.is_ha:
+ self._spawn_metadata_proxy(router.router_id,
+ router.ns_name,
+ self.l3_agent.conf)
+
+ def before_router_removed(self, router):
+ for c, r in self.metadata_filter_rules(self.metadata_port):
+ router.iptables_manager.ipv4['filter'].remove_rule(c, r)
+ for c, r in self.metadata_nat_rules(self.metadata_port):
+ router.iptables_manager.ipv4['nat'].remove_rule(c, r)
+ router.iptables_manager.apply()
+
+ self._destroy_metadata_proxy(router.router['id'],
+ router.ns_name,
+ self.l3_agent.conf)
+
+ @classmethod
+ def metadata_filter_rules(cls, port):
+ return [('INPUT', '-s 0.0.0.0/0 -d 127.0.0.1 '
+ '-p tcp -m tcp --dport %s '
+ '-j ACCEPT' % port)]
+
+ @classmethod
+ def metadata_nat_rules(cls, port):
+ return [('PREROUTING', '-s 0.0.0.0/0 -d 169.254.169.254/32 '
+ '-p tcp -m tcp --dport 80 -j REDIRECT '
+ '--to-port %s' % port)]
+
+ @classmethod
+ def _get_metadata_proxy_user_group(cls, conf):
+ user = conf.metadata_proxy_user or os.geteuid()
+ group = conf.metadata_proxy_group or os.getegid()
+ return user, group
+
+ @classmethod
+ def _get_metadata_proxy_callback(cls, router_id, conf):
+
+ def callback(pid_file):
+ metadata_proxy_socket = conf.metadata_proxy_socket
+ user, group = cls._get_metadata_proxy_user_group(conf)
+ proxy_cmd = ['neutron-ns-metadata-proxy',
+ '--pid_file=%s' % pid_file,
+ '--metadata_proxy_socket=%s' % metadata_proxy_socket,
+ '--router_id=%s' % router_id,
+ '--state_path=%s' % conf.state_path,
+ '--metadata_port=%s' % conf.metadata_port,
+ '--metadata_proxy_user=%s' % user,
+ '--metadata_proxy_group=%s' % group]
+ proxy_cmd.extend(config.get_log_args(
+ conf, 'neutron-ns-metadata-proxy-%s.log' %
+ router_id))
+ return proxy_cmd
+
+ return callback
+
+ @classmethod
+ def _get_metadata_proxy_process_manager(cls, router_id, ns_name, conf):
+ return external_process.ProcessManager(
+ conf,
+ router_id,
+ config.get_root_helper(conf),
+ ns_name)
+
+ @classmethod
+ def _spawn_metadata_proxy(cls, router_id, ns_name, conf):
+ callback = cls._get_metadata_proxy_callback(router_id, conf)
+ pm = cls._get_metadata_proxy_process_manager(router_id, ns_name, conf)
+ pm.enable(callback)
+
+ @classmethod
+ def _destroy_metadata_proxy(cls, router_id, ns_name, conf):
+ pm = cls._get_metadata_proxy_process_manager(router_id, ns_name, conf)
+ pm.disable()
--- /dev/null
+# Copyright 2014 OpenStack Foundation.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import contextlib
+
+import mock
+
+from oslo.config import cfg
+
+from neutron.agent.common import config as agent_config
+from neutron.agent.l3 import agent as l3_agent
+from neutron.agent.metadata import driver as metadata_driver
+from neutron.openstack.common import uuidutils
+from neutron.tests import base
+
+
+_uuid = uuidutils.generate_uuid
+
+
+class TestMetadataDriver(base.BaseTestCase):
+
+ EUID = 123
+ EGID = 456
+
+ def setUp(self):
+ super(TestMetadataDriver, self).setUp()
+ cfg.CONF.register_opts(l3_agent.L3NATAgent.OPTS)
+ agent_config.register_root_helper(cfg.CONF)
+
+ def test_metadata_nat_rules(self):
+ rules = ('PREROUTING', '-s 0.0.0.0/0 -d 169.254.169.254/32 '
+ '-p tcp -m tcp --dport 80 -j REDIRECT --to-port 8775')
+ self.assertEqual(
+ [rules],
+ metadata_driver.MetadataDriver.metadata_nat_rules(8775))
+
+ def test_metadata_filter_rules(self):
+ rules = ('INPUT', '-s 0.0.0.0/0 -d 127.0.0.1 '
+ '-p tcp -m tcp --dport 8775 -j ACCEPT')
+ self.assertEqual(
+ [rules],
+ metadata_driver.MetadataDriver.metadata_filter_rules(8775))
+
+ def _test_spawn_metadata_proxy(self, expected_user, expected_group,
+ user='', group=''):
+ router_id = _uuid()
+ router_ns = 'qrouter-%s' % router_id
+ metadata_port = 8080
+ ip_class_path = 'neutron.agent.linux.ip_lib.IPWrapper'
+
+ cfg.CONF.set_override('metadata_port', metadata_port)
+ cfg.CONF.set_override('metadata_proxy_user', user)
+ cfg.CONF.set_override('metadata_proxy_group', group)
+ cfg.CONF.set_override('log_file', 'test.log')
+ cfg.CONF.set_override('debug', True)
+
+ driver = metadata_driver.MetadataDriver
+ with contextlib.nested(
+ mock.patch('os.geteuid', return_value=self.EUID),
+ mock.patch('os.getegid', return_value=self.EGID),
+ mock.patch(ip_class_path)) as (geteuid, getegid, ip_mock):
+ driver._spawn_metadata_proxy(router_id, router_ns, cfg.CONF)
+ ip_mock.assert_has_calls([
+ mock.call('sudo', router_ns),
+ mock.call().netns.execute([
+ 'neutron-ns-metadata-proxy',
+ mock.ANY,
+ mock.ANY,
+ '--router_id=%s' % router_id,
+ mock.ANY,
+ '--metadata_port=%s' % metadata_port,
+ '--metadata_proxy_user=%s' % expected_user,
+ '--metadata_proxy_group=%s' % expected_group,
+ '--debug',
+ '--verbose',
+ '--log-file=neutron-ns-metadata-proxy-%s.log' %
+ router_id
+ ], addl_env=None)
+ ])
+
+ def test_spawn_metadata_proxy_with_user(self):
+ self._test_spawn_metadata_proxy('user', self.EGID, user='user')
+
+ def test_spawn_metadata_proxy_with_uid(self):
+ self._test_spawn_metadata_proxy('321', self.EGID, user='321')
+
+ def test_spawn_metadata_proxy_with_group(self):
+ self._test_spawn_metadata_proxy(self.EUID, 'group', group='group')
+
+ def test_spawn_metadata_proxy_with_gid(self):
+ self._test_spawn_metadata_proxy(self.EUID, '654', group='654')
+
+ def test_spawn_metadata_proxy(self):
+ self._test_spawn_metadata_proxy(self.EUID, self.EGID)
from neutron.agent.l3 import router_info as l3router
from neutron.agent.linux import interface
from neutron.agent.linux import ra
+from neutron.agent.metadata import driver as metadata_driver
from neutron.common import config as base_config
from neutron.common import constants as l3_constants
from neutron.common import exceptions as n_exc
from neutron.i18n import _LE
+from neutron.openstack.common import log
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants as p_const
from neutron.tests import base
super(TestBasicRouterOperations, self).setUp()
self.conf = agent_config.setup_conf()
self.conf.register_opts(base_config.core_opts)
+ self.conf.register_cli_opts(log.common_cli_opts)
+ self.conf.register_cli_opts(log.logging_cli_opts)
self.conf.register_opts(l3_agent.L3NATAgent.OPTS)
self.conf.register_opts(ha.OPTS)
agent_config.register_interface_driver_opts_helper(self.conf)
self.assertEqual(agent.process_router_floating_ip_nat_rules.called,
distributed)
- def test_ha_router_keepalived_config(self):
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- router = prepare_router_data(enable_ha=True)
- router['routes'] = [
- {'destination': '8.8.8.8/32', 'nexthop': '35.4.0.10'},
- {'destination': '8.8.4.4/32', 'nexthop': '35.4.0.11'}]
- ri = l3router.RouterInfo(router['id'], self.conf.root_helper,
- router=router)
- ri.router = router
- with contextlib.nested(mock.patch.object(agent,
- '_spawn_metadata_proxy'),
- mock.patch('neutron.agent.linux.'
- 'utils.replace_file'),
- mock.patch('neutron.agent.linux.'
- 'utils.execute'),
- mock.patch('os.makedirs')):
- agent.process_ha_router_added(ri)
- agent.process_router(ri)
- config = ri.keepalived_manager.config
- ha_iface = agent.get_ha_device_name(ri.ha_port['id'])
- ex_iface = agent.get_external_device_name(ri.ex_gw_port['id'])
- int_iface = agent.get_internal_device_name(
- ri.internal_ports[0]['id'])
-
- expected = """vrrp_sync_group VG_1 {
- group {
- VR_1
- }
-}
-vrrp_instance VR_1 {
- state BACKUP
- interface %(ha_iface)s
- virtual_router_id 1
- priority 50
- nopreempt
- advert_int 2
- track_interface {
- %(ha_iface)s
- }
- virtual_ipaddress {
- 19.4.4.4/24 dev %(ex_iface)s
- }
- virtual_ipaddress_excluded {
- 35.4.0.4/24 dev %(int_iface)s
- }
- virtual_routes {
- 0.0.0.0/0 via 19.4.4.1 dev %(ex_iface)s
- 8.8.8.8/32 via 35.4.0.10
- 8.8.4.4/32 via 35.4.0.11
- }
-}""" % {'ha_iface': ha_iface, 'ex_iface': ex_iface, 'int_iface': int_iface}
-
- self.assertEqual(expected, config.get_config_str())
-
@mock.patch('neutron.agent.linux.ip_lib.IPDevice')
def _test_process_router_floating_ip_addresses_add(self, ri,
agent, IPDevice):
self.conf.set_override('enable_metadata_proxy', False)
agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
router_id = _uuid()
- router = {'id': _uuid(),
+ router = {'id': router_id,
'external_gateway_info': {},
'routes': [],
'distributed': False}
+ driver = metadata_driver.MetadataDriver
with mock.patch.object(
- agent, '_destroy_metadata_proxy') as destroy_proxy:
+ driver, '_destroy_metadata_proxy') as destroy_proxy:
with mock.patch.object(
- agent, '_spawn_metadata_proxy') as spawn_proxy:
- agent._router_added(router_id, router)
+ driver, '_spawn_metadata_proxy') as spawn_proxy:
+ agent._process_added_router(router)
if enableflag:
- spawn_proxy.assert_called_with(router_id, mock.ANY)
+ spawn_proxy.assert_called_with(router_id,
+ mock.ANY,
+ mock.ANY)
else:
self.assertFalse(spawn_proxy.call_count)
agent._router_removed(router_id)
if enableflag:
- destroy_proxy.assert_called_with(mock.ANY, mock.ANY)
+ destroy_proxy.assert_called_with(router_id,
+ mock.ANY,
+ mock.ANY)
else:
self.assertFalse(destroy_proxy.call_count)
def test_disable_metadata_proxy_spawn(self):
self._configure_metadata_proxy(enableflag=False)
- def test_metadata_nat_rules(self):
- self.conf.set_override('enable_metadata_proxy', False)
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- self.assertEqual([], agent.metadata_nat_rules())
-
- self.conf.set_override('metadata_port', '8775')
- self.conf.set_override('enable_metadata_proxy', True)
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- rules = ('PREROUTING', '-s 0.0.0.0/0 -d 169.254.169.254/32 '
- '-p tcp -m tcp --dport 80 -j REDIRECT --to-port 8775')
- self.assertEqual([rules], agent.metadata_nat_rules())
-
def test_router_id_specified_in_conf(self):
self.conf.set_override('use_namespaces', False)
self.conf.set_override('router_id', '')
msg = _LE("Error importing interface driver '%s'")
log.error.assert_called_once_with(msg, 'wrong_driver')
- def test_metadata_filter_rules(self):
- self.conf.set_override('enable_metadata_proxy', False)
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- self.assertEqual([], agent.metadata_filter_rules())
-
- self.conf.set_override('metadata_port', '8775')
- self.conf.set_override('enable_metadata_proxy', True)
- agent = l3_agent.L3NATAgent(HOSTNAME, self.conf)
- rules = ('INPUT', '-s 0.0.0.0/0 -d 127.0.0.1 '
- '-p tcp -m tcp --dport 8775 -j ACCEPT')
- self.assertEqual([rules], agent.metadata_filter_rules())
-
def _cleanup_namespace_test(self,
stale_namespace_list,
router_list,
asserter = self.assertIn if flag_set else self.assertNotIn
asserter('AdvOtherConfigFlag on;',
self.utils_replace_file.call_args[0][1])
-
-
-class TestL3AgentEventHandler(base.BaseTestCase):
-
- EUID = '123'
- EGID = '456'
-
- def setUp(self):
- super(TestL3AgentEventHandler, self).setUp()
- cfg.CONF.register_opts(l3_agent.L3NATAgent.OPTS)
- cfg.CONF.register_opts(ha.OPTS)
- agent_config.register_interface_driver_opts_helper(cfg.CONF)
- agent_config.register_use_namespaces_opts_helper(cfg.CONF)
- cfg.CONF.set_override(
- 'interface_driver', 'neutron.agent.linux.interface.NullDriver'
- )
- cfg.CONF.set_override('use_namespaces', True)
- cfg.CONF.set_override('verbose', False)
- agent_config.register_root_helper(cfg.CONF)
-
- device_exists_p = mock.patch(
- 'neutron.agent.linux.ip_lib.device_exists')
- device_exists_p.start()
-
- utils_exec_p = mock.patch(
- 'neutron.agent.linux.utils.execute')
- utils_exec_p.start()
-
- drv_cls_p = mock.patch('neutron.agent.linux.interface.NullDriver')
- driver_cls = drv_cls_p.start()
- mock_driver = mock.MagicMock()
- mock_driver.DEV_NAME_LEN = (
- interface.LinuxInterfaceDriver.DEV_NAME_LEN)
- driver_cls.return_value = mock_driver
-
- l3_plugin_p = mock.patch(
- 'neutron.agent.l3.agent.L3PluginApi')
- l3_plugin_cls = l3_plugin_p.start()
- l3_plugin_cls.return_value = mock.MagicMock()
-
- self.external_process_p = mock.patch(
- 'neutron.agent.linux.external_process.ProcessManager'
- )
- self.external_process_p.start()
- looping_call_p = mock.patch(
- 'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
- looping_call_p.start()
- self.agent = l3_agent.L3NATAgent(HOSTNAME)
-
- def _test_spawn_metadata_proxy(self, expected_user, expected_group,
- user='', group=''):
- router_id = _uuid()
- metadata_port = 8080
- ip_class_path = 'neutron.agent.linux.ip_lib.IPWrapper'
-
- cfg.CONF.set_override('metadata_port', metadata_port)
- cfg.CONF.set_override('log_file', 'test.log')
- cfg.CONF.set_override('debug', True)
- cfg.CONF.set_override('metadata_proxy_user', user)
- cfg.CONF.set_override('metadata_proxy_group', group)
-
- self.external_process_p.stop()
- ri = l3router.RouterInfo(router_id, None, None)
- with contextlib.nested(
- mock.patch('os.geteuid', return_value=self.EUID),
- mock.patch('os.getegid', return_value=self.EGID),
- mock.patch(ip_class_path)) as (geteuid, getegid, ip_mock):
- self.agent._spawn_metadata_proxy(ri.router_id, ri.ns_name)
- ip_mock.assert_has_calls([
- mock.call('sudo', ri.ns_name),
- mock.call().netns.execute([
- 'neutron-ns-metadata-proxy',
- mock.ANY,
- mock.ANY,
- '--router_id=%s' % router_id,
- mock.ANY,
- '--metadata_port=%s' % metadata_port,
- '--metadata_proxy_user=%s' % expected_user,
- '--metadata_proxy_group=%s' % expected_group,
- '--debug',
- '--log-file=neutron-ns-metadata-proxy-%s.log' %
- router_id
- ], addl_env=None)
- ])
-
- def test_spawn_metadata_proxy_with_user(self):
- self._test_spawn_metadata_proxy('user', self.EGID, user='user')
-
- def test_spawn_metadata_proxy_with_uid(self):
- self._test_spawn_metadata_proxy('321', self.EGID, user='321')
-
- def test_spawn_metadata_proxy_with_group(self):
- self._test_spawn_metadata_proxy(self.EUID, 'group', group='group')
-
- def test_spawn_metadata_proxy_with_gid(self):
- self._test_spawn_metadata_proxy(self.EUID, '654', group='654')
-
- def test_spawn_metadata_proxy(self):
- self._test_spawn_metadata_proxy(self.EUID, self.EGID)
Code Review / openstack-build / neutron-build.git / commitdiff