import random
+import netaddr
+
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ovs_lib
from neutron.agent.linux import utils
BR_PREFIX = 'test-br'
+PORT_PREFIX = 'test-port'
ICMP_BLOCK_RULE = '-p icmp -j DROP'
VETH_PREFIX = 'tst-vth'
self.skipTest(skip_msg)
raise
+ def _create_namespace(self):
+ ip_cmd = ip_lib.IPWrapper(self.root_helper)
+ name = "func-%s" % uuidutils.generate_uuid()
+ namespace = ip_cmd.ensure_namespace(name)
+ self.addCleanup(namespace.netns.delete, namespace.namespace)
+
+ return namespace
+
def create_resource(self, name_prefix, creation_func, *args, **kwargs):
"""Create a new resource that does not already exist.
veth1, veth2 = ip_wrapper.add_veth(name1, name2)
return veth1, veth2
+ def set_namespace_gateway(self, port_dev, gateway_ip):
+ """Set gateway for the namespace associated to the port."""
+ if not port_dev.namespace:
+ self.fail('tests should not change test machine gateway')
+ port_dev.route.add_gateway(gateway_ip)
+
+ def shift_ip_cidr(self, ip_cidr, offset=1):
+ """Shift ip_cidr offset times.
+
+ example: shift_ip_cidr("1.2.3.4/24", 2) ==> "1.2.3.6/24"
+ """
+ net = netaddr.IPNetwork(ip_cidr)
+ net.value += offset
+ return str(net)
+
class BaseOVSLinuxTestCase(BaseLinuxTestCase):
def setUp(self):
super(BaseOVSLinuxTestCase, self).setUp()
self.ovs = ovs_lib.BaseOVS(self.root_helper)
+ self.ip = ip_lib.IPWrapper(self.root_helper)
def create_ovs_bridge(self, br_prefix=BR_PREFIX):
br = self.create_resource(br_prefix, self.ovs.add_bridge)
def get_ovs_bridge(self, br_name):
return ovs_lib.OVSBridge(br_name, self.root_helper)
+ def create_ovs_port_in_ns(self, br, ns):
+ def create_port(name):
+ br.add_port(name)
+ self.addCleanup(br.delete_port, name)
+ br.set_db_attribute('Interface', name, 'type', 'internal')
+ return name
+ port_name = self.create_resource(PORT_PREFIX, create_port)
+ port_dev = self.ip.device(port_name)
+ ns.add_device_to_namespace(port_dev)
+ port_dev.link.set_up()
+ return port_dev
+
+ def bind_namespace_to_cidr(self, namespace, br, ip_cidr):
+ """Bind namespace to cidr (on layer2 and 3).
+
+ Bind the namespace to a subnet by creating an ovs port in the namespace
+ and configuring port ip.
+ """
+ net = netaddr.IPNetwork(ip_cidr)
+ port_dev = self.create_ovs_port_in_ns(br, namespace)
+ port_dev.addr.add(net.version, str(net), net.broadcast)
+ return port_dev
+
class BaseIPVethTestCase(BaseLinuxTestCase):
SRC_ADDRESS = '192.168.0.1'
device.addr.add(ip_version=ip_version, cidr=cidr, broadcast=broadcast)
device.link.set_up()
- def _create_namespace(self):
- ip_cmd = ip_lib.IPWrapper(self.root_helper)
- name = "func-%s" % uuidutils.generate_uuid()
- namespace = ip_cmd.ensure_namespace(name)
- self.addCleanup(namespace.netns.delete, namespace.namespace)
-
- return namespace
-
def prepare_veth_pairs(self, src_addr=None,
dst_addr=None,
broadcast_addr=None,
import fixtures
import mock
from oslo.config import cfg
+import webob
+import webob.dec
+import webob.exc
from neutron.agent.common import config as agent_config
from neutron.agent.l3 import agent as l3_agent
+from neutron.agent.linux import dhcp
from neutron.agent.linux import external_process
from neutron.agent.linux import ip_lib
+from neutron.agent.metadata import agent as metadata_agent
from neutron.common import config as common_config
from neutron.common import constants as l3_constants
from neutron.openstack.common import log as logging
LOG = logging.getLogger(__name__)
_uuid = uuidutils.generate_uuid
+METADATA_REQUEST_TIMEOUT = 60
+
class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
def setUp(self):
helpers.wait_until_true(lambda: router2.ha_state == 'master')
helpers.wait_until_true(lambda: router1.ha_state == 'fault')
+
+
+class MetadataFakeProxyHandler(object):
+
+ def __init__(self, status):
+ self.status = status
+
+ @webob.dec.wsgify()
+ def __call__(self, req):
+ return webob.Response(status=self.status)
+
+
+class MetadataL3AgentTestCase(L3AgentTestFramework):
+
+ def _create_metadata_fake_server(self, status):
+ server = metadata_agent.UnixDomainWSGIServer('metadata-fake-server')
+ self.addCleanup(server.stop)
+ server.start(MetadataFakeProxyHandler(status),
+ self.agent.conf.metadata_proxy_socket,
+ workers=0, backlog=4096)
+
+ def test_access_to_metadata_proxy(self):
+ """Test access to the l3-agent metadata proxy.
+
+ The test creates:
+ * A l3-agent metadata service:
+ * A router (which creates a metadata proxy in the router namespace),
+ * A fake metadata server
+ * A "client" namespace (simulating a vm) with a port on router
+ internal subnet.
+
+ The test queries from the "client" namespace the metadata proxy on
+ http://169.254.169.254 and asserts that the metadata proxy added
+ the X-Forwarded-For and X-Neutron-Router-Id headers to the request
+ and forwarded the http request to the fake metadata server and the
+ response to the "client" namespace.
+ """
+ router_info = self.generate_router_info(enable_ha=False)
+ router = self.manage_router(self.agent, router_info)
+ self._create_metadata_fake_server(webob.exc.HTTPOk.code)
+
+ # Create and configure client namespace
+ client_ns = self._create_namespace()
+ router_ip_cidr = router.internal_ports[0]['ip_cidr']
+ ip_cidr = self.shift_ip_cidr(router_ip_cidr)
+ br_int = self.get_ovs_bridge(self.agent.conf.ovs_integration_bridge)
+ port = self.bind_namespace_to_cidr(client_ns, br_int, ip_cidr)
+ self.set_namespace_gateway(port, router_ip_cidr.partition('/')[0])
+
+ # Query metadata proxy
+ url = 'http://%(host)s:%(port)s' % {'host': dhcp.METADATA_DEFAULT_IP,
+ 'port': dhcp.METADATA_PORT}
+ cmd = 'curl', '--max-time', METADATA_REQUEST_TIMEOUT, '-D-', url
+ try:
+ raw_headers = client_ns.netns.execute(cmd)
+ except RuntimeError:
+ self.fail('metadata proxy unreachable on %s before timeout' % url)
+
+ # Check status code
+ firstline = raw_headers.splitlines()[0]
+ self.assertIn(str(webob.exc.HTTPOk.code), firstline.split())
Code Review / openstack-build / neutron-build.git / commitdiff