neutron/tests/unit/agent/metadata/test_driver.py

   1 # Copyright 2014 OpenStack Foundation.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17 from oslo_config import cfg
  18 from oslo_utils import uuidutils
  19
  20 from neutron.agent.common import config as agent_config
  21 from neutron.agent.l3 import agent as l3_agent
  22 from neutron.agent.l3 import config as l3_config
  23 from neutron.agent.l3 import ha as l3_ha_agent
  24 from neutron.agent.metadata import config
  25 from neutron.agent.metadata import driver as metadata_driver
  26 from neutron.common import constants
  27 from neutron.tests import base
  28
  29
  30 _uuid = uuidutils.generate_uuid
  31
  32
  33 class TestMetadataDriverRules(base.BaseTestCase):
  34
  35     def test_metadata_nat_rules(self):
  36         rules = ('PREROUTING', '-d 169.254.169.254/32 -i qr-+ '
  37                  '-p tcp -m tcp --dport 80 -j REDIRECT --to-port 8775')
  38         self.assertEqual(
  39             [rules],
  40             metadata_driver.MetadataDriver.metadata_nat_rules(8775))
  41
  42     def test_metadata_filter_rules(self):
  43         rules = [('INPUT', '-m mark --mark 0x1/%s -j ACCEPT' %
  44                   constants.ROUTER_MARK_MASK),
  45                  ('INPUT', '-p tcp -m tcp --dport 8775 -j DROP')]
  46         self.assertEqual(
  47             rules,
  48             metadata_driver.MetadataDriver.metadata_filter_rules(8775, '0x1'))
  49
  50     def test_metadata_mangle_rules(self):
  51         rule = ('PREROUTING', '-d 169.254.169.254/32 -i qr-+ '
  52                 '-p tcp -m tcp --dport 80 '
  53                 '-j MARK --set-xmark 0x1/%s' %
  54                 constants.ROUTER_MARK_MASK)
  55         self.assertEqual(
  56             [rule],
  57             metadata_driver.MetadataDriver.metadata_mangle_rules('0x1'))
  58
  59
  60 class TestMetadataDriverProcess(base.BaseTestCase):
  61
  62     EUID = 123
  63     EGID = 456
  64     EUNAME = 'neutron'
  65
  66     def setUp(self):
  67         super(TestMetadataDriverProcess, self).setUp()
  68         mock.patch('eventlet.spawn').start()
  69         agent_config.register_interface_driver_opts_helper(cfg.CONF)
  70         cfg.CONF.set_override('interface_driver',
  71                               'neutron.agent.linux.interface.NullDriver')
  72
  73         mock.patch('neutron.agent.l3.agent.L3PluginApi').start()
  74         mock.patch('neutron.agent.l3.ha.AgentMixin'
  75                    '._init_ha_conf_path').start()
  76
  77         cfg.CONF.register_opts(l3_config.OPTS)
  78         cfg.CONF.register_opts(l3_ha_agent.OPTS)
  79         cfg.CONF.register_opts(config.SHARED_OPTS)
  80         cfg.CONF.register_opts(config.DRIVER_OPTS)
  81
  82     def _test_spawn_metadata_proxy(self, expected_user, expected_group,
  83                                    user='', group='', watch_log=True):
  84         router_id = _uuid()
  85         router_ns = 'qrouter-%s' % router_id
  86         metadata_port = 8080
  87         ip_class_path = 'neutron.agent.linux.ip_lib.IPWrapper'
  88         is_effective_user = 'neutron.agent.linux.utils.is_effective_user'
  89         fake_is_effective_user = lambda x: x in [self.EUNAME, str(self.EUID)]
  90
  91         cfg.CONF.set_override('metadata_proxy_user', user)
  92         cfg.CONF.set_override('metadata_proxy_group', group)
  93         cfg.CONF.set_override('log_file', 'test.log')
  94         cfg.CONF.set_override('debug', True)
  95
  96         agent = l3_agent.L3NATAgent('localhost')
  97         with mock.patch('os.geteuid', return_value=self.EUID),\
  98                 mock.patch('os.getegid', return_value=self.EGID),\
  99                 mock.patch(is_effective_user,
 100                            side_effect=fake_is_effective_user),\
 101                 mock.patch(ip_class_path) as ip_mock:
 102             agent.metadata_driver.spawn_monitored_metadata_proxy(
 103                 agent.process_monitor,
 104                 router_ns,
 105                 metadata_port,
 106                 agent.conf,
 107                 router_id=router_id)
 108             netns_execute_args = [
 109                 'neutron-ns-metadata-proxy',
 110                 mock.ANY,
 111                 mock.ANY,
 112                 '--router_id=%s' % router_id,
 113                 mock.ANY,
 114                 '--metadata_port=%s' % metadata_port,
 115                 '--metadata_proxy_user=%s' % expected_user,
 116                 '--metadata_proxy_group=%s' % expected_group,
 117                 '--debug',
 118                 '--verbose',
 119                 '--log-file=neutron-ns-metadata-proxy-%s.log' %
 120                 router_id]
 121             if not watch_log:
 122                 netns_execute_args.append(
 123                     '--nometadata_proxy_watch_log')
 124             ip_mock.assert_has_calls([
 125                 mock.call(namespace=router_ns),
 126                 mock.call().netns.execute(netns_execute_args, addl_env=None,
 127                                           run_as_root=False)
 128             ])
 129
 130     def test_spawn_metadata_proxy_with_agent_user(self):
 131         self._test_spawn_metadata_proxy(
 132             self.EUNAME, str(self.EGID), user=self.EUNAME)
 133
 134     def test_spawn_metadata_proxy_with_nonagent_user(self):
 135         self._test_spawn_metadata_proxy(
 136             'notneutron', str(self.EGID), user='notneutron', watch_log=False)
 137
 138     def test_spawn_metadata_proxy_with_agent_uid(self):
 139         self._test_spawn_metadata_proxy(
 140             str(self.EUID), str(self.EGID), user=str(self.EUID))
 141
 142     def test_spawn_metadata_proxy_with_nonagent_uid(self):
 143         self._test_spawn_metadata_proxy(
 144             '321', str(self.EGID), user='321', watch_log=False)
 145
 146     def test_spawn_metadata_proxy_with_group(self):
 147         self._test_spawn_metadata_proxy(str(self.EUID), 'group', group='group')
 148
 149     def test_spawn_metadata_proxy_with_gid(self):
 150         self._test_spawn_metadata_proxy(str(self.EUID), '654', group='654')
 151
 152     def test_spawn_metadata_proxy(self):
 153         self._test_spawn_metadata_proxy(str(self.EUID), str(self.EGID))