/usr/lib/python2.7/dist-packages/ovsdbapp/tests/functional/schema/open_vswitch/test_common_db.py is in python-ovsdbapp 0.9.1-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | # 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.backend.ovs_idl import idlutils
from ovsdbapp.schema.open_vswitch import impl_idl
from ovsdbapp.tests.functional import base
from ovsdbapp.tests.functional.schema.open_vswitch import fixtures
from ovsdbapp.tests import utils
class TestBackendDb(base.FunctionalTestCase):
schemas = ["Open_vSwitch"]
def setUp(self):
self.bridges = [
{'name': utils.get_rand_device_name(),
'datapath_type': 'fake1'},
{'name': utils.get_rand_device_name(),
'datapath_type': 'fake1'},
{'name': utils.get_rand_device_name(),
'datapath_type': 'fake2'}
]
super(TestBackendDb, self).setUp()
self.api = impl_idl.OvsdbIdl(self.connection)
for bridge in self.bridges:
self.useFixture(fixtures.BridgeFixture(bridge['name']))
for col, val in bridge.items():
if col == 'name':
continue
self.api.db_set(
'Bridge', bridge['name'], (col, val)).execute(
check_error=True)
def test_db_find(self):
res = self.api.db_find(
'Bridge',
('datapath_type', '=', 'fake1'),
columns=['name', 'datapath_type']).execute(check_error=True)
self.assertItemsEqual(self.bridges[:2], res)
def test_db_find_no_exist(self):
res = self.api.db_find(
'Bridge', ('name', '=', 'unpossible')).execute(check_error=True)
self.assertFalse(res)
def test_db_find_rows(self):
res = self.api.db_find_rows(
'Bridge',
('datapath_type', '=', 'fake1')).execute(check_error=True)
self.assertItemsEqual(
self.bridges[:2],
[{'name': r.name, 'datapath_type': r.datapath_type} for r in res])
def test_db_list(self):
res = self.api.db_list(
'Bridge',
columns=('name', 'datapath_type')).execute(check_error=True)
self.assertTrue(all(b in res for b in self.bridges))
def test_db_list_record(self):
res = self.api.db_list(
'Bridge', [self.bridges[0]['name']],
('name', 'datapath_type')).execute(check_error=True)
self.assertEqual(self.bridges[0], res[0])
def test_db_list_record_no_exist(self):
cmd = self.api.db_list('Bridge', ['unpossible'])
self.assertRaises(idlutils.RowNotFound, cmd.execute, check_error=True)
def test_db_list_record_if_exsists(self):
self.api.db_list('Bridge', ['unpossible'])
def test_db_list_rows(self):
res = self.api.db_list_rows('Bridge').execute(check_error=True)
self.assertTrue(
set(b['name'] for b in self.bridges).issubset(
set(b.name for b in res)))
|