From 1ab6c6a3a03f004d8bb76de9114687528426b560 Mon Sep 17 00:00:00 2001 From: Dima Kuznetsov Date: Wed, 19 Apr 2017 13:27:26 +0300 Subject: [PATCH] Split tests --- wan_qos/tests/unit/test_db.py | 101 ----------------------------- wan_qos/tests/unit/test_plugin.py | 104 ++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 101 deletions(-) create mode 100644 wan_qos/tests/unit/test_plugin.py diff --git a/wan_qos/tests/unit/test_db.py b/wan_qos/tests/unit/test_db.py index e85aa7e..49cefd5 100644 --- a/wan_qos/tests/unit/test_db.py +++ b/wan_qos/tests/unit/test_db.py @@ -2,7 +2,6 @@ from neutron.tests.unit import testlib_api from neutron_lib import context as ctx from wan_qos.db import wan_qos_db -from wan_qos.services import plugin class TestTcDb(testlib_api.SqlTestCase): @@ -73,103 +72,3 @@ class TestTcDb(testlib_api.SqlTestCase): wtc_class['max'] = max return self.db.create_wan_tc_class(self.context, wtc_class) - - -class TestPlugin(testlib_api.SqlTestCase): - def setUp(self): - super(TestPlugin, self).setUp() - self.plugin = plugin.WanQosPlugin() - - def test_create_class(self): - wtc_class = { - 'wan_tc_class': { - 'direction': 'both', - 'min': '1mbit' - } - } - - wan_tc = self.plugin.create_wan_tc_class(ctx.get_admin_context(), - wtc_class) - - assert wan_tc is not None - print (wan_tc) - - def test_get_class_by_id(self): - - class_db_1 = self._add_class(None, 'both', '1mbit', '2mbit') - class_db_2 = self._add_class(class_db_1['id'], 'both', '2mbit', - '3mbit') - self._add_class(class_db_2['id'], 'both', '3mbit', - '4mbit') - - tc_class = self.plugin.get_wan_tc_class(ctx.get_admin_context(), - class_db_1['id']) - - print(tc_class) - tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context()) - - print(tc_classes) - - def test_get_all_classes_by_id(self): - - class_db_1 = self._add_class(None, 'both', '1mbit', '2mbit') - print ('class_1: %s ' % class_db_1) - - filters = {'id': [class_db_1['id']]} - # filters = {'id': ['11']} - # filters = {'name': ['111']} - - tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context(), - filters=filters) - - print(tc_classes) - - filters = {'name': ['111']} - - tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context(), - filters=filters) - - print(tc_classes) - - def test_add_filter(self): - - class_db = self._add_class(None, 'both', '1mbit', '2mbit') - filter = self._get_filter(class_db['id']) - filter_db = self.plugin.create_wan_tc_filter(ctx.get_admin_context(), - filter) - - print ('filter: %s' % filter_db) - - filters = { - 'id': [filter_db['id']] - # 'name': ['123'] - } - filter_by_id = self.plugin.get_wan_tc_filters(ctx.get_admin_context(), - filters=filters) - - print('filter by id: %s' % filter_by_id) - - def _get_filter(self, class_id): - - filter = {'wan_tc_filter': { - 'protocol': 'vxlan', - 'match': 'vni=123', - 'class_id': class_id - } - } - - return filter - - def _add_class(self, parent, direction, min, max): - wtc_class = { - 'direction': direction, - } - if min: - wtc_class['min'] = min - if parent: - wtc_class['parent'] = parent - if max: - wtc_class['max'] = max - - return self.plugin.db.create_wan_tc_class(ctx.get_admin_context(), - wtc_class) diff --git a/wan_qos/tests/unit/test_plugin.py b/wan_qos/tests/unit/test_plugin.py new file mode 100644 index 0000000..1dc9780 --- /dev/null +++ b/wan_qos/tests/unit/test_plugin.py @@ -0,0 +1,104 @@ +from neutron.tests.unit import testlib_api +from neutron_lib import context as ctx + +from wan_qos.services import plugin + + +class TestPlugin(testlib_api.SqlTestCase): + def setUp(self): + super(TestPlugin, self).setUp() + self.plugin = plugin.WanQosPlugin() + + def test_create_class(self): + wtc_class = { + 'wan_tc_class': { + 'direction': 'both', + 'min': '1mbit' + } + } + + wan_tc = self.plugin.create_wan_tc_class(ctx.get_admin_context(), + wtc_class) + + assert wan_tc is not None + print (wan_tc) + + def test_get_class_by_id(self): + + class_db_1 = self._add_class(None, 'both', '1mbit', '2mbit') + class_db_2 = self._add_class(class_db_1['id'], 'both', '2mbit', + '3mbit') + self._add_class(class_db_2['id'], 'both', '3mbit', + '4mbit') + + tc_class = self.plugin.get_wan_tc_class(ctx.get_admin_context(), + class_db_1['id']) + + print(tc_class) + tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context()) + + print(tc_classes) + + def test_get_all_classes_by_id(self): + + class_db_1 = self._add_class(None, 'both', '1mbit', '2mbit') + print ('class_1: %s ' % class_db_1) + + filters = {'id': [class_db_1['id']]} + # filters = {'id': ['11']} + # filters = {'name': ['111']} + + tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context(), + filters=filters) + + print(tc_classes) + + filters = {'name': ['111']} + + tc_classes = self.plugin.get_wan_tc_classs(ctx.get_admin_context(), + filters=filters) + + print(tc_classes) + + def test_add_filter(self): + + class_db = self._add_class(None, 'both', '1mbit', '2mbit') + filter = self._get_filter(class_db['id']) + filter_db = self.plugin.create_wan_tc_filter(ctx.get_admin_context(), + filter) + + print ('filter: %s' % filter_db) + + filters = { + 'id': [filter_db['id']] + # 'name': ['123'] + } + filter_by_id = self.plugin.get_wan_tc_filters(ctx.get_admin_context(), + filters=filters) + + print('filter by id: %s' % filter_by_id) + + def _get_filter(self, class_id): + + filter = {'wan_tc_filter': { + 'protocol': 'vxlan', + 'match': 'vni=123', + 'class_id': class_id + } + } + + return filter + + def _add_class(self, parent, direction, min, max): + wtc_class = { + 'direction': direction, + } + if min: + wtc_class['min'] = min + if parent: + wtc_class['parent'] = parent + if max: + wtc_class['max'] = max + + return self.plugin.db.create_wan_tc_class(ctx.get_admin_context(), + wtc_class)