Add some bridge and port functional tests

Change-Id: Iaeb71bc4dff3f076d913d6bf0b34e0dd19dc9b93
This commit is contained in:
Terry Wilson 2017-03-16 17:52:12 -05:00
parent 1af842a207
commit e95b895b31
7 changed files with 185 additions and 0 deletions

17
ovsdbapp/constants.py Normal file
View File

@ -0,0 +1,17 @@
# Copyright (c) 2017 Red Hat, Inc.
#
# 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.
DEFAULT_OVSDB_CONNECTION = 'tcp:127.0.0.1:6640'
DEFAULT_TIMEOUT = 5
DEVICE_NAME_MAX_LEN = 14

View File

View File

@ -0,0 +1,96 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2017 Red Hat, Inc.
# 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.
from ovsdbapp import constants
from ovsdbapp import impl_idl
from ovsdbapp.tests import base
from ovsdbapp.tests import utils
class DefaultContext(object):
ovsdb_connection = constants.DEFAULT_OVSDB_CONNECTION
vsctl_timeout = constants.DEFAULT_TIMEOUT
class TestOvsdbIdl(base.TestCase):
def setUp(self):
super(TestOvsdbIdl, self).setUp()
self.api = impl_idl.OvsdbIdl(DefaultContext())
self.brname = utils.get_rand_device_name()
# Destroying the bridge cleans up most things created by tests
cleanup_cmd = self.api.del_br(self.brname)
self.addCleanup(cleanup_cmd.execute)
def test_br_exists_false(self):
exists = self.api.br_exists(self.brname).execute(check_error=True)
self.assertFalse(exists)
def test_add_br_may_exist(self):
self.api.add_br(self.brname).execute(check_error=True)
with self.api.transaction(check_error=True) as txn:
txn.add(self.api.add_br(self.brname, datapath_type="netdev"))
exists = txn.add(self.api.br_exists(self.brname))
dpt = txn.add(self.api.db_get("Bridge", self.brname,
"datapath_type"))
self.assertTrue(exists)
self.assertEqual("netdev", dpt.result)
def test_add_br_may_not_exist(self):
self.api.add_br(self.brname).execute(check_error=True)
cmd = self.api.add_br(self.brname, may_exist=False)
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
def test_del_br_if_exists_false(self):
cmd = self.api.del_br(self.brname, if_exists=False)
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
def test_del_br_if_exists_true(self):
self.api.del_br(self.brname).execute(check_error=True)
def test_del_br(self):
self.api.add_br(self.brname).execute(check_error=True)
self.api.del_br(self.brname).execute(check_error=True)
exists = self.api.br_exists(self.brname).execute(check_error=True)
self.assertFalse(exists)
def _test_add_port(self):
pname = utils.get_rand_device_name()
with self.api.transaction(check_error=True) as txn:
txn.add(self.api.add_br(self.brname))
txn.add(self.api.add_port(self.brname, pname))
return pname
def test_add_port(self):
pname = self._test_add_port()
plist_cmd = self.api.list_ports(self.brname)
ports = plist_cmd.execute(check_error=True)
self.assertIn(pname, ports)
def test_add_port_may_exist_false(self):
pname = self._test_add_port()
cmd = self.api.add_port(self.brname, pname, may_exist=False)
self.assertRaises(RuntimeError, cmd.execute, check_error=True)
def test_del_port(self):
pname = self._test_add_port()
plist_cmd = self.api.list_ports(self.brname)
self.assertIn(pname, plist_cmd.execute(check_error=True))
self.api.del_port(pname).execute(check_error=True)
self.assertNotIn(pname, plist_cmd.execute(check_error=True))
def test_del_port_if_exists_false(self):
cmd = self.api.del_port(utils.get_rand_device_name(), if_exists=False)
self.assertRaises(RuntimeError, cmd.execute, check_error=True)

View File

68
ovsdbapp/tests/utils.py Normal file
View File

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
# 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 random
from ovsdbapp import constants
# NOTE(twilson) This code was copied from the neutron/neutron-lib trees
def get_random_string(length):
"""Get a random hex string of the specified length.
:param length: The length for the hex string.
:returns: A random hex string of the said length.
"""
return "{0:0{1}x}".format(random.getrandbits(length * 4), length)
def get_rand_name(max_length=None, prefix='test'):
"""Return a random string.
The string will start with 'prefix' and will be exactly 'max_length'.
If 'max_length' is None, then exactly 8 random characters, each
hexadecimal, will be added. In case len(prefix) <= len(max_length),
ValueError will be raised to indicate the problem.
"""
return get_related_rand_names([prefix], max_length)[0]
def get_rand_device_name(prefix='test'):
return get_rand_name(
max_length=constants.DEVICE_NAME_MAX_LEN, prefix=prefix)
def get_related_rand_names(prefixes, max_length=None):
"""Returns a list of the prefixes with the same random characters appended
:param prefixes: A list of prefix strings
:param max_length: The maximum length of each returned string
:returns: A list with each prefix appended with the same random characters
"""
if max_length:
length = max_length - max(len(p) for p in prefixes)
if length <= 0:
raise ValueError(
_("'max_length' must be longer than all prefixes"))
else:
length = 8
rndchrs = get_random_string(length)
return [p + rndchrs for p in prefixes]
def get_related_rand_device_names(prefixes):
return get_related_rand_names(prefixes,
max_length=constants.DEVICE_NAME_MAX_LEN)

View File

@ -9,6 +9,7 @@ install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstac
setenv =
VIRTUAL_ENV={envdir}
PYTHONWARNINGS=default::DeprecationWarning
OS_TEST_PATH=./ovsdbapp/tests/unit
deps = -r{toxinidir}/test-requirements.txt
commands = python setup.py testr --slowest --testr-args='{posargs}'
@ -31,6 +32,9 @@ commands =
[testenv:debug]
commands = oslo_debug_helper {posargs}
[testenv:functional]
setenv = OS_TEST_PATH=./ovsdbapp/tests/functional
[flake8]
# E123, E125 skipped as they are invalid PEP-8.