openstack
/
ovsdbapp
Code Issues Proposed changes

Merge "Allow functional tests to pass on older OVN w/o IC"

This commit is contained in:
Zuul 2022-01-04 20:34:21 +00:00 committed by Gerrit Code Review
parent 48cd4da530 8a21d5f3a3
commit 2c6b6c9c8e
3 changed files with 81 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
ovsdbapp/tests/functional/base.py
View File

@ -22,23 +22,29 @@ from ovsdbapp import venv
class FunctionalTestCase(base.TestCase):
_connections = None
ovsvenv = venv.OvsOvnVenvFixture(tempfile.mkdtemp(),
ovsdir=os.getenv('OVS_SRCDIR'),
ovndir=os.getenv('OVN_SRCDIR'),
remove=not bool(os.getenv('KEEP_VENV')))
atexit.register(ovsvenv.cleanUp)
ovsvenv.setUp()
schema_map = {'Open_vSwitch': ovsvenv.ovs_connection,
'OVN_Northbound': ovsvenv.ovnnb_connection,
'OVN_Southbound': ovsvenv.ovnsb_connection,
'OVN_IC_Northbound': ovsvenv.ovn_icnb_connection,
}
ovsvenvlog = None
if os.getenv('KEEP_VENV') and os.getenv('VIRTUAL_ENV'):
ovsvenvlog = open(os.path.join(os.getenv('VIRTUAL_ENV'),
'ovsvenv.%s' % os.getpid()), 'a+')
atexit.register(ovsvenvlog.close)
ovsvenvlog.write("%s\n" % ovsvenv.venv)
fixture_class = venv.OvsOvnVenvFixture
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.ovsvenv = cls.fixture_class(
tempfile.mkdtemp(),
ovsdir=os.getenv('OVS_SRCDIR'),
ovndir=os.getenv('OVN_SRCDIR'),
remove=not bool(os.getenv('KEEP_VENV')))
atexit.register(cls.ovsvenv.cleanUp)
cls.ovsvenv.setUp()
cls.schema_map = {'Open_vSwitch': cls.ovsvenv.ovs_connection,
'OVN_Northbound': cls.ovsvenv.ovnnb_connection,
'OVN_Southbound': cls.ovsvenv.ovnsb_connection,
}
cls.ovsvenvlog = None
if os.getenv('KEEP_VENV') and os.getenv('VIRTUAL_ENV'):
cls.ovsvenvlog = open(
os.path.join(os.getenv('VIRTUAL_ENV'),
'ovsvenv.%s' % os.getpid()), 'a+')
atexit.register(cls.ovsvenvlog.close)
cls.ovsvenvlog.write("%s\n" % cls.ovsvenv.venv)
@classmethod
def venv_log(cls, val):

10
ovsdbapp/tests/functional/schema/ovn_ic_northbound/test_impl_idl.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
from ovsdbapp.tests.functional import base
from ovsdbapp.tests.functional.schema.ovn_ic_northbound import fixtures
from ovsdbapp.tests import utils
@ -18,8 +17,17 @@ from ovsdbapp.tests import utils
class OvnIcNorthboundTest(base.FunctionalTestCase):
schemas = ['OVN_IC_Northbound']
fixture_class = base.venv.OvsOvnIcVenvFixture
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.schema_map = cls.schema_map.copy()
cls.schema_map['OVN_IC_Northbound'] = cls.ovsvenv.ovn_icnb_connection
def setUp(self):
if not self.ovsvenv.has_icnb():
self.skipTest("Installed version of OVN does not support ICNB")
super(OvnIcNorthboundTest, self).setUp()
self.api = self.useFixture(
fixtures.IcNbApiFixture(self.connection)).obj

66
ovsdbapp/venv.py
View File

@ -164,8 +164,7 @@ class OvsOvnVenvFixture(OvsVenvFixture):
":{0}/controller:{0}/northd:{0}/utilities".format(ovndir))
super().__init__(venv, **kwargs)
self.ovndir = self._share_path(self.OVN_PATHS, ovndir,
[self.SBSCHEMA, self.NBSCHEMA,
self.IC_NBSCHEMA])
[self.SBSCHEMA, self.NBSCHEMA])
self.env.update({'OVN_RUNDIR': self.venv})
@property
@ -176,10 +175,6 @@ class OvsOvnVenvFixture(OvsVenvFixture):
def ovnnb_schema(self):
return os.path.join(self.ovndir, self.NBSCHEMA)
@property
def ovn_icnb_schema(self):
return os.path.join(self.ovndir, self.IC_NBSCHEMA)
@property
def ovnnb_connection(self):
return 'unix:' + os.path.join(self.venv, 'ovnnb_db.sock')
@ -188,26 +183,20 @@ class OvsOvnVenvFixture(OvsVenvFixture):
def ovnsb_connection(self):
return 'unix:' + os.path.join(self.venv, 'ovnsb_db.sock')
@property
def ovn_icnb_connection(self):
return 'unix:' + os.path.join(self.venv, 'ovn_ic_nb_db.sock')
def setup_dbs(self):
super().setup_dbs()
self.create_db('ovnsb.db', self.ovnsb_schema)
self.create_db('ovnnb.db', self.ovnnb_schema)
self.create_db('ovn_ic_nb.db', self.ovn_icnb_schema)
def start_ovsdb_processes(self):
super().start_ovsdb_processes()
for connection, schema, db_name, table in [
data = [
(self.ovnnb_connection,
"OVN_Northbound", "ovnnb", "NB_Global"),
(self.ovnsb_connection,
"OVN_Southbound", "ovnsb", "SB_Global"),
(self.ovn_icnb_connection,
"OVN_IC_Northbound", "ovn_ic_nb", "IC_NB_Global"),
]:
"OVN_Southbound", "ovnsb", "SB_Global")]
for connection, schema, db_name, table in data:
self.call(['ovsdb-server',
'--detach', '--no-chdir', '-vconsole:off',
'--pidfile=%s' % os.path.join(self.venv,
@ -226,7 +215,6 @@ class OvsOvnVenvFixture(OvsVenvFixture):
super().init_processes()
self.call(['ovn-nbctl', 'init'])
self.call(['ovn-sbctl', 'init'])
self.call(['ovn-ic-nbctl', 'init'])
if self.add_chassis:
self.call([
'ovs-vsctl', 'set', 'open', '.',
@ -243,3 +231,47 @@ class OvsOvnVenvFixture(OvsVenvFixture):
'--ovnnb-db=' + self.ovnnb_connection])
self.call(['ovn-controller', '--detach', '--no-chdir', '--pidfile',
'-vconsole:off', '--log-file'])
class OvsOvnIcVenvFixture(OvsOvnVenvFixture):
def _setUp(self):
if self.has_icnb():
super()._setUp()
@property
def ovn_icnb_connection(self):
return 'unix:' + os.path.join(self.venv, 'ovn_ic_nb_db.sock')
@property
def ovn_icnb_schema(self):
return os.path.join(self.ovndir, self.IC_NBSCHEMA)
def has_icnb(self):
return os.path.isfile(self.ovn_icnb_schema)
def setup_dbs(self):
super().setup_dbs()
self.create_db('ovn_ic_nb.db', self.ovn_icnb_schema)
def start_ovsdb_processes(self):
super().start_ovsdb_processes()
connection, schema, db_name, table = (
self.ovn_icnb_connection,
"OVN_IC_Northbound", "ovn_ic_nb", "IC_NB_Global")
self.call(['ovsdb-server',
'--detach', '--no-chdir', '-vconsole:off',
'--pidfile=%s' % os.path.join(self.venv,
'%s_db.pid' % (db_name)),
'--log-file=%s' % os.path.join(self.venv,
'%s_db.log' % (db_name)),
'--remote=db:%s,%s,connections' % (schema, table),
'--private-key=db:%s,SSL,private_key' % (schema),
'--certificate=db:%s,SSL,certificate' % (schema),
'--ca-cert=db:%s,SSL,ca_cert' % (schema),
'--ssl-protocols=db:%s,SSL,ssl_protocols' % (schema),
'--ssl-ciphers=db:%s,SSL,ssl_ciphers' % (schema),
'--remote=p' + connection, '%s.db' % (db_name)])
def init_processes(self):
super().init_processes()
self.call(['ovn-ic-nbctl', 'init'])