212 lines
8.4 KiB
Python
212 lines
8.4 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
# Copyright 2012 Nicira Networks, Inc.
|
|
# 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.
|
|
#
|
|
# @author: Dave Lapsley, Nicira Networks, Inc.
|
|
|
|
import logging
|
|
import mox
|
|
import os
|
|
import unittest
|
|
from agent import ovs_quantum_agent
|
|
|
|
LOG = logging.getLogger("quantum.plugins.openvswitch.tests.unit.test_tunnel")
|
|
LOG.setLevel(logging.INFO)
|
|
|
|
LOCAL_DIR = os.path.dirname(__file__)
|
|
REMOTE_IP_FILE = LOCAL_DIR + '/remote-ip-file.txt'
|
|
|
|
# Useful global dummy variables.
|
|
NET_UUID = '3faeebfe-5d37-11e1-a64b-000c29d5f0a7'
|
|
LS_ID = '42'
|
|
LV_ID = 42
|
|
LV_IDS = [42, 43]
|
|
LVM = ovs_quantum_agent.LocalVLANMapping(LV_ID, LS_ID, LV_IDS)
|
|
VIF_ID = '404deaec-5d37-11e1-a64b-000c29d5f0a8'
|
|
VIF_MAC = '3c:09:24:1e:78:23'
|
|
VIF_PORT = ovs_quantum_agent.VifPort('port', 'ofport', VIF_ID, VIF_MAC,
|
|
'switch')
|
|
|
|
|
|
class DummyPort:
|
|
def __init__(self, interface_id):
|
|
self.interface_id = interface_id
|
|
|
|
|
|
class DummyVlanBinding:
|
|
def __init__(self, network_id, vlan_id):
|
|
self.network_id = network_id
|
|
self.vlan_id = vlan_id
|
|
|
|
|
|
class TunnelTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
print LOCAL_DIR
|
|
self.mox = mox.Mox()
|
|
|
|
self.INT_BRIDGE = 'integration_bridge'
|
|
self.TUN_BRIDGE = 'tunnel_bridge'
|
|
self.INT_OFPORT = 'PATCH_INT_OFPORT'
|
|
self.TUN_OFPORT = 'PATCH_TUN_OFPORT'
|
|
|
|
self.mox.StubOutClassWithMocks(ovs_quantum_agent, 'OVSBridge')
|
|
self.mock_int_bridge = ovs_quantum_agent.OVSBridge(self.INT_BRIDGE,
|
|
'sudo')
|
|
self.mock_int_bridge.delete_port('patch-tun')
|
|
self.mock_int_bridge.add_patch_port(
|
|
'patch-tun', 'patch-int').AndReturn(self.TUN_OFPORT)
|
|
self.mock_int_bridge.remove_all_flows()
|
|
self.mock_int_bridge.add_flow(priority=1, actions='normal')
|
|
|
|
self.mock_tun_bridge = ovs_quantum_agent.OVSBridge(self.TUN_BRIDGE,
|
|
'sudo')
|
|
self.mock_tun_bridge.reset_bridge()
|
|
self.mock_tun_bridge.add_patch_port(
|
|
'patch-int', 'patch-tun').AndReturn(self.INT_OFPORT)
|
|
self.mock_tun_bridge.remove_all_flows()
|
|
self.mock_tun_bridge.add_flow(priority=1, actions='drop')
|
|
|
|
def tearDown(self):
|
|
self.mox.UnsetStubs()
|
|
|
|
def testConstruct(self):
|
|
self.mox.ReplayAll()
|
|
|
|
b = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
self.mox.VerifyAll()
|
|
|
|
def testProvisionLocalVlan(self):
|
|
match_string = 'in_port=%s,dl_vlan=%s' % (self.INT_OFPORT, LV_ID)
|
|
action_string = 'set_tunnel:%s,normal' % LS_ID
|
|
self.mock_tun_bridge.add_flow(priority=4, match=match_string,
|
|
actions=action_string)
|
|
|
|
match_string = 'tun_id=%s' % LS_ID
|
|
action_string = 'mod_vlan_vid:%s,output:%s' % (LV_ID, self.INT_OFPORT)
|
|
self.mock_tun_bridge.add_flow(priority=3, match=match_string,
|
|
actions=action_string)
|
|
|
|
self.mox.ReplayAll()
|
|
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
a.available_local_vlans = set([LV_ID])
|
|
a.provision_local_vlan(NET_UUID, LS_ID)
|
|
self.mox.VerifyAll()
|
|
|
|
def testReclaimLocalVlan(self):
|
|
match_string = 'tun_id=%s' % LVM.lsw_id
|
|
self.mock_tun_bridge.delete_flows(match=match_string)
|
|
|
|
match_string = 'dl_vlan=%s' % LVM.vlan
|
|
self.mock_tun_bridge.delete_flows(match=match_string)
|
|
|
|
self.mox.ReplayAll()
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
a.available_local_vlans = set()
|
|
a.local_vlan_map[NET_UUID] = LVM
|
|
a.reclaim_local_vlan(NET_UUID, LVM)
|
|
self.assertTrue(LVM.vlan in a.available_local_vlans)
|
|
self.mox.VerifyAll()
|
|
|
|
def testPortBound(self):
|
|
self.mock_int_bridge.set_db_attribute('Port', VIF_PORT.port_name,
|
|
'tag', str(LVM.vlan))
|
|
self.mock_int_bridge.delete_flows(match='in_port=%s' % VIF_PORT.ofport)
|
|
|
|
self.mox.ReplayAll()
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
a.local_vlan_map[NET_UUID] = LVM
|
|
a.port_bound(VIF_PORT, NET_UUID, LS_ID)
|
|
self.mox.VerifyAll()
|
|
|
|
def testPortUnbound(self):
|
|
self.mox.ReplayAll()
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
a.available_local_vlans = set([LV_ID])
|
|
a.local_vlan_map[NET_UUID] = LVM
|
|
a.port_unbound(VIF_PORT, NET_UUID)
|
|
self.mox.VerifyAll()
|
|
|
|
def testPortDead(self):
|
|
self.mock_int_bridge.set_db_attribute('Port', VIF_PORT.port_name,
|
|
'tag', ovs_quantum_agent.DEAD_VLAN_TAG)
|
|
|
|
match_string = 'in_port=%s' % VIF_PORT.ofport
|
|
self.mock_int_bridge.add_flow(priority=2, match=match_string,
|
|
actions='drop')
|
|
|
|
self.mox.ReplayAll()
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
a.available_local_vlans = set([LV_ID])
|
|
a.local_vlan_map[NET_UUID] = LVM
|
|
a.port_dead(VIF_PORT)
|
|
self.mox.VerifyAll()
|
|
|
|
def testDbBindings(self):
|
|
db = self.mox.CreateMockAnything()
|
|
db.ports = self.mox.CreateMockAnything()
|
|
interface_ids = ['interface-id-%d' % x for x in range(3)]
|
|
db.ports.all().AndReturn([DummyPort(x) for x in interface_ids])
|
|
|
|
db.vlan_bindings = self.mox.CreateMockAnything()
|
|
vlan_bindings = [
|
|
['network-id-%d' % x, 'vlan-id-%d' % x] for x in range(3)]
|
|
db.vlan_bindings.all().AndReturn(
|
|
[DummyVlanBinding(*x) for x in vlan_bindings])
|
|
|
|
self.mox.ReplayAll()
|
|
a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
|
|
self.TUN_BRIDGE,
|
|
REMOTE_IP_FILE,
|
|
'10.0.0.1',
|
|
'sudo')
|
|
|
|
all_bindings = a.get_db_port_bindings(db)
|
|
lsw_id_bindings = a.get_db_vlan_bindings(db)
|
|
|
|
for interface_id, port in all_bindings.iteritems():
|
|
self.assertTrue(interface_id in interface_ids)
|
|
|
|
for network_id, vlan_id in lsw_id_bindings.iteritems():
|
|
self.assertTrue(network_id in [x[0] for x in vlan_bindings])
|
|
self.assertTrue(vlan_id in [x[1] for x in vlan_bindings])
|
|
|
|
self.mox.VerifyAll()
|