diff --git a/compass/db/models.py b/compass/db/models.py index ff11328e..48bec2b4 100644 --- a/compass/db/models.py +++ b/compass/db/models.py @@ -94,7 +94,8 @@ class HelperMixin(object): return False def validate(self): - for key, column in self.__mapper__.columns.items(): + columns = self.__mapper__.columns + for key, column in columns.items(): value = getattr(self, key) if not self.type_compatible(value, column.type): raise exception.InvalidParameter( diff --git a/compass/tests/api/test_api.py b/compass/tests/db/api/base.py similarity index 75% rename from compass/tests/api/test_api.py rename to compass/tests/db/api/base.py index 02c72222..6ffac947 100644 --- a/compass/tests/api/test_api.py +++ b/compass/tests/db/api/base.py @@ -1,5 +1,3 @@ -#!/usr/bin/python -# # Copyright 2014 Huawei Technologies Co. Ltd # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -14,35 +12,33 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""test api module.""" -import celery -import copy -import mock + +import datetime +import logging import os import unittest2 +from compass.db.api import adapter_holder as adapter_api +from compass.db.api import database +from compass.db.api import metadata_holder as metadata_api +from compass.db.api import switch +from compass.db.api import user as user_api +from compass.db import exception +from compass.utils import flags +from compass.utils import logsetting +from compass.utils import setting_wrapper as setting os.environ['COMPASS_IGNORE_SETTING'] = 'true' -from compass.utils import setting_wrapper as setting reload(setting) -# from compass.api import app -from compass.db.api import adapter_holder as adapter_api -from compass.db.api import database -from compass.db.api import metadata_holder as metadata_api -from compass.utils import flags -from compass.utils import logsetting -from compass.utils import util - - -class ApiTestCase(unittest2.TestCase): - """base api test class.""" +class BaseTest(unittest2.TestCase): + """Base Class for unit test.""" def setUp(self): - super(ApiTestCase, self).setUp() + super(BaseTest, self).setUp() reload(setting) setting.CONFIG_DIR = os.path.join( os.path.dirname(os.path.abspath(__file__)), @@ -52,17 +48,13 @@ class ApiTestCase(unittest2.TestCase): database.create_db() adapter_api.load_adapters() metadata_api.load_metadatas() + self.user_object = ( + user_api.get_user_object( + setting.COMPASS_ADMIN_EMAIL + ) + ) def tearDown(self): - database.drop_db() + super(BaseTest, self).setUp() reload(setting) - super(ApiTestCase, self).tearDown() - - def test_login(self): - pass - - -if __name__ == '__main__': - flags.init() - logsetting.init() - unittest2.main() + database.drop_db() diff --git a/compass/tests/db/api/test_machine.py b/compass/tests/db/api/test_machine.py index 4a38776e..538f7877 100644 --- a/compass/tests/db/api/test_machine.py +++ b/compass/tests/db/api/test_machine.py @@ -17,6 +17,7 @@ import logging import os import unittest2 +from base import BaseTest from compass.db.api import database from compass.db.api import machine from compass.db.api import switch @@ -25,35 +26,10 @@ from compass.db import exception from compass.utils import flags from compass.utils import logsetting from compass.utils import setting_wrapper as setting -reload(setting) os.environ['COMPASS_IGNORE_SETTING'] = 'true' -class BaseTest(unittest2.TestCase): - """Base Class for unit test.""" - - def setUp(self): - super(BaseTest, self).setUp() - reload(setting) - setting.CONFIG_DIR = os.path.join( - os.path.dirname(os.path.abspath(__file__)), - 'data' - ) - database.init('sqlite://') - database.create_db() - self.user_object = ( - user_api.get_user_object( - setting.COMPASS_ADMIN_EMAIL - ) - ) - - def tearDown(self): - super(BaseTest, self).setUp() - reload(setting) - database.drop_db() - - class TestGetMachine(BaseTest): """Test get machine.""" diff --git a/compass/tests/db/api/test_metadata_holder.py b/compass/tests/db/api/test_metadata_holder.py new file mode 100644 index 00000000..a82a005a --- /dev/null +++ b/compass/tests/db/api/test_metadata_holder.py @@ -0,0 +1,56 @@ +# Copyright 2014 Huawei Technologies Co. Ltd +# +# 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 datetime +import logging +import os +import unittest2 + +from base import BaseTest +from compass.db.api import adapter_holder as adapter +from compass.db.api import database +from compass.db.api import metadata_holder as metadata +from compass.db.api import user as user_api +from compass.db import exception +from compass.utils import flags +from compass.utils import logsetting +from compass.utils import setting_wrapper as setting + + +os.environ['COMPASS_IGNORE_SETTING'] = 'true' + + +class TestGetPackageMetadata(BaseTest): + """test get package metadata.""" + + def setUp(self): + super(TestGetPackageMetadata, self).setUp() + + def tearDown(self): + super(TestGetPackageMetadata, self).tearDown() + + def test_get_package_metadata(self): + adapter_object = adapter.list_adapters(self.user_object) + package_metadata = metadata.get_package_metadata( + self.user_object, + adapter_object[0]['id'] + ) + self.assertIsNotNone(package_metadata) + + +if __name__ == '__main__': + flags.init() + logsetting.init() + unittest2.main() diff --git a/compass/tests/db/api/test_network.py b/compass/tests/db/api/test_network.py new file mode 100644 index 00000000..f112f021 --- /dev/null +++ b/compass/tests/db/api/test_network.py @@ -0,0 +1,188 @@ +# Copyright 2014 Huawei Technologies Co. Ltd +# +# 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 datetime +import logging +import os +import unittest2 + +from base import BaseTest +from compass.db.api import database +from compass.db.api import network +from compass.db.api import user as user_api +from compass.db import exception +from compass.utils import flags +from compass.utils import logsetting +from compass.utils import setting_wrapper as setting + +reload(setting) + + +os.environ['COMPASS_IGNORE_SETTING'] = 'true' + + +class TestListSubnets(BaseTest): + """Test list subnets.""" + + def setUp(self): + super(TestListSubnets, self).setUp() + + def tearDown(self): + super(TestListSubnets, self).tearDown() + + def test_list_subnets(self): + network.add_subnet( + self.user_object, + subnet='10.145.89.0/24' + ) + list_subnet = network.list_subnets( + self.user_object + ) + expected = '10.145.89.0/24' + self.assertTrue( + item in list_subnet[0].items() for item in expected + ) + + +class TestGetSubnet(BaseTest): + """Test get subnet.""" + + def setUp(self): + super(TestGetSubnet, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestGetSubnet, self).tearDown() + + def test_get_subnet(self): + network.add_subnet( + self.user_object, + subnet='10.145.89.0/24' + ) + get_subnet = network.get_subnet( + self.user_object, + 1 + ) + self.assertEqual( + '10.145.89.0/24', + get_subnet['subnet'] + ) + + def tset_get_subnet_no_exist(self): + get_subnet_no_exist = network.get_subnet( + self.user_object, + 2 + ) + self.assertEqual([], get_subnet_no_exist) + + +class TestAddSubnet(BaseTest): + """Test add subnet.""" + + def setUp(self): + super(TestAddSubnet, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestAddSubnet, self).tearDown() + + def test_add_subnet(self): + network.add_subnet( + self.user_object, + subnet='10.145.89.0/24' + ) + add_subnet = network.list_subnets( + self.user_object + ) + expected = '10.145.89.0/24' + self.assertTrue( + item in add_subnet[0].items() for item in expected + ) + + +class TestUpdateSubnet(BaseTest): + """Test update subnet.""" + + def setUp(self): + super(TestUpdateSubnet, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSubnet, self).tearDown() + + def test_update_subnet(self): + network.add_subnet( + self.user_object, + subnet='10.145.89.0/24' + ) + network.update_subnet( + self.user_object, + 1, + subnet='192.168.100.0/24' + ) + update_subnet = network.list_subnets( + self.user_object + ) + expected = '192.168.100.0/24' + self.assertTrue( + item in update_subnet[0].items() for item in expected + ) + + def test_update_subnet_no_exist(self): + self.assertRaises( + exception.DatabaseException, + network.update_subnet, + self.user_object, + 2 + ) + + +class TestDelSubnet(BaseTest): + """Test delete subnet.""" + + def setUp(self): + super(TestDelSubnet, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestDelSubnet, self).tearDown() + + def test_del_subnet(self): + network.add_subnet( + self.user_object, + subnet='10.145.89.0/24' + ) + network.del_subnet( + self.user_object, + 1 + ) + del_subnet = network.list_subnets( + self.user_object + ) + self.assertEqual([], del_subnet) + + def test_del_subnet_not_exist(self): + self.assertRaises( + exception.RecordNotExists, + network.del_subnet, + self.user_object, + 2 + ) + + +if __name__ == '__main__': + flags.init() + logsetting.init() + unittest2.main() diff --git a/compass/tests/db/api/test_permission.py b/compass/tests/db/api/test_permission.py index 7358f9e2..c922714d 100644 --- a/compass/tests/db/api/test_permission.py +++ b/compass/tests/db/api/test_permission.py @@ -17,6 +17,7 @@ import logging import os import unittest2 +from base import BaseTest from compass.db.api import database from compass.db.api import permission from compass.db.api import user as user_api @@ -26,30 +27,7 @@ from compass.utils import logsetting from compass.utils import setting_wrapper as setting os.environ['COMPASS_IGNORE_SETTING'] = 'true' - - -class BaseTest(unittest2.TestCase): - """Base Class for unit test.""" - - def setUp(self): - super(BaseTest, self).setUp() - reload(setting) - setting.CONFIG_DIR = os.path.join( - os.path.dirname(os.path.abspath(__file__)), - 'data' - ) - database.init('sqlite://') - database.create_db() - self.user_object = ( - user_api.get_user_object( - setting.COMPASS_ADMIN_EMAIL - ) - ) - - def tearDown(self): - database.drop_db() - reload(setting) - super(BaseTest, self).tearDown() +reload(setting) class TestListPermissions(BaseTest): diff --git a/compass/tests/db/api/test_switch.py b/compass/tests/db/api/test_switch.py new file mode 100644 index 00000000..5e24accb --- /dev/null +++ b/compass/tests/db/api/test_switch.py @@ -0,0 +1,781 @@ +# Copyright 2014 Huawei Technologies Co. Ltd +# +# 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 datetime +import logging +import os +import unittest2 + +from base import BaseTest +from compass.db.api import database +from compass.db.api import switch +from compass.db.api import user as user_api +from compass.db import exception +from compass.utils import flags +from compass.utils import logsetting +from compass.utils import setting_wrapper as setting + + +os.environ['COMPASS_IGNORE_SETTING'] = 'true' + + +reload(setting) + + +class TestGetSwitch(BaseTest): + """Test get switch.""" + + def setUp(self): + super(TestGetSwitch, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestGetSwitch, self).tearDown() + + def test_get_switch(self): + get_switch = switch.get_switch( + self.user_object, + 1 + ) + self.assertIsNotNone(get_switch) + + +class TestAddSwitch(BaseTest): + """Test add switch.""" + + def setUp(self): + super(TestAddSwitch, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestAddSwitch, self).tearDown() + + def test_add_switch(self): + add_switch = switch.add_switch( + self.user_object, + ip='2887583784' + ) + expected = '172.29.8.40' + self.assertEqual(expected, add_switch['ip']) + + +class TestListSwitches(BaseTest): + """Test list switch.""" + + def setUp(self): + super(TestListSwitches, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitches, self).tearDown() + + def test_list_switches_ip_int_invalid(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + list_switches = switch.list_switches( + self.user_object, + ip_int='test' + ) + self.assertEqual(list_switches, []) + + def test_list_switches_with_ip_int(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + list_switches = switch.list_switches( + self.user_object, + ip_int='2887583784' + ) + expected = '2887583784' + self.assertTrue( + item in expected.items() for item in list_switches[0].items() + ) + + def test_list_switches(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + list_switches = switch.list_switches( + self.user_object + ) + self.assertIsNotNone(list_switches) + + +class TestDelSwitch(BaseTest): + """Test delete switch.""" + + def setUp(self): + super(TestDelSwitch, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestDelSwitch, self).tearDown() + + def test_del_switch(self): + switch.del_switch( + self.user_object, + 1 + ) + del_switch = switch.list_switches( + self.user_object + ) + self.assertEqual([], del_switch) + + +class TestUpdateSwitch(BaseTest): + """Test update switch.""" + + def setUp(self): + super(TestUpdateSwitch, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSwitch, self).tearDown() + + def test_update_switch(self): + switch.update_switch( + self.user_object, + 1, + vendor='test_update' + ) + update_switch = switch.get_switch( + self.user_object, + 1 + ) + expected = 'test_update' + self.assertEqual(expected, update_switch['vendor']) + + +class TestPatchSwitch(BaseTest): + """Test patch switch.""" + + def setUp(self): + super(TestPatchSwitch, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestPatchSwitch, self).tearDown() + + def test_patch_switch(self): + switch.patch_switch( + self.user_object, + 1, + patched_credentials={ + 'version': '2c', + 'community': 'public' + } + ) + patch_switch = switch.get_switch( + self.user_object, + 1 + ) + expected = { + 'version': '2c', + 'community': 'public' + } + self.assertTrue( + item in expected.items() for item in patch_switch.items() + ) + + +class TestListSwitchFilters(BaseTest): + """Test list switch filters.""" + + def setUp(self): + super(TestListSwitchFilters, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitchFilters, self).tearDown() + + def test_list_switch_filters(self): + list_switch_filters = switch.list_switch_filters( + self.user_object + ) + self.assertIsNotNone(list_switch_filters) + + +class TestGetSwitchFilters(BaseTest): + """Test get switch filter.""" + + def setUp(self): + super(TestGetSwitchFilters, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestGetSwitchFilters, self).tearDown() + + def test_get_swtich_filters(self): + get_switch_filter = switch.get_switch_filters( + self.user_object, + 1 + ) + self.assertIsNotNone(get_switch_filter) + + +class TestUpdateSwitchFilters(BaseTest): + """Test update a switch filter.""" + + def setUp(self): + super(TestUpdateSwitchFilters, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSwitchFilters, self).tearDown() + + def test_update_switch_filters(self): + switch.update_switch_filters( + self.user_object, + 1, + filters=[ + { + 'filter_name': 'test', + 'filter_type': 'allow' + } + ] + ) + update_switch_filters = switch.get_switch_filters( + self.user_object, + 1 + ) + expected = { + 'filter_name': 'test', + 'filter_type': 'allow' + } + self.assertTrue( + item in update_switch_filters[0].items() + for item in expected.items() + ) + + +class TestPatchSwitchFilter(BaseTest): + """Test patch a switch filter.""" + + def setUp(self): + super(TestPatchSwitchFilter, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestPatchSwitchFilter, self).tearDown() + + def test_patch_switch_filter(self): + switch.patch_switch_filter( + self.user_object, + 1, + patched_filters=[ + { + 'filter_name': 'test', + 'filter_type': 'allow' + } + ] + ) + patch_switch_filter = switch.get_switch_filters( + self.user_object, + 1 + ) + expected = { + 'filter_name': 'test', + 'filter_type': 'allow' + } + self.assertTrue( + item in patch_switch_filter[0].items() for item in expected.items() + ) + + +class TestAddSwitchMachine(BaseTest): + """Test add switch machine.""" + + def setUp(self): + super(TestAddSwitchMachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestAddSwitchMachine, self).tearDown() + + def test_add_switch_machine(self): + add_switch_machine = switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + expected = '28:6e:d4:46:c4:25' + self.assertEqual(expected, add_switch_machine['mac']) + + +class TestListSwitchMachines(BaseTest): + """Test get switch machines.""" + + def setUp(self): + super(TestListSwitchMachines, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitchMachines, self).tearDown() + + def test_list_switch_machines(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_switch_machines = switch.list_switch_machines( + self.user_object, + 2 + ) + self.assertIsNotNone(list_switch_machines) + + +class TestListSwitchmachines(BaseTest): + """Test list switch machines.""" + + def setUp(self): + super(TestListSwitchmachines, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitchmachines, self).tearDown() + + def test_list_switch_machines_with_ip_int(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_switch_machines = switch.list_switchmachines( + self.user_object, + switch_ip_int='2887583784' + ) + expected = '172.29.8.40' + self.assertTrue(expected for item in list_switch_machines[0].items()) + + def test_list_switch_machines_ip_invalid(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_switch_machines = switch.list_switchmachines( + self.user_object, + switch_ip_int='test' + ) + self.assertEqual(list_switch_machines, []) + + def test_list_switch_machines_without_ip(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_switch_machines = switch.list_switchmachines( + self.user_object + ) + self.assertIsNotNone(list_switch_machines) + + +class TestListSwitchMachinesHosts(BaseTest): + """Test get switch machines hosts.""" + + def setUp(self): + super(TestListSwitchMachinesHosts, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitchMachinesHosts, self).tearDown() + + def test_list_hosts(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_hosts = switch.list_switch_machines_hosts( + self.user_object, + 2 + ) + self.assertIsNotNone(list_hosts) + + +class TestListSwitchmachinesHosts(BaseTest): + """Test list switch machines hosts.""" + + def setUp(self): + super(TestListSwitchmachinesHosts, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestListSwitchmachinesHosts, self).tearDown() + + def test_list_hosts_with_ip_int(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_hosts = switch.list_switchmachines_hosts( + self.user_object, + switch_ip_int='2887583784' + ) + expected = '172.29.8.40' + self.assertTrue(expected for item in list_hosts[0].items()) + + def test_list_hosts_ip_invalid(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_hosts = switch.list_switchmachines_hosts( + self.user_object, + switch_ip_int='test' + ) + self.assertEqual(list_hosts, []) + + def test_list_hosts_without_ip(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + list_hosts = switch.list_switchmachines_hosts( + self.user_object + ) + self.assertIsNotNone(list_hosts) + + +class TestGetSwitchMachine(BaseTest): + """Test get a switch machines.""" + + def setUp(self): + super(TestGetSwitchMachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestGetSwitchMachine, self).tearDown() + + def test_get_switch_machine(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + get_switch_machine = switch.get_switch_machine( + self.user_object, + 2, + 1 + ) + self.assertIsNotNone(get_switch_machine) + + +class TestGetSwitchmachine(BaseTest): + """Test get a switch machine.""" + + def setUp(self): + super(TestGetSwitchmachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestGetSwitchmachine, self).tearDown() + + def test_get_switchmachine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + get_switchmachine = switch.get_switchmachine( + self.user_object, + 1 + ) + self.assertIsNotNone(get_switchmachine) + + +class TestUpdateSwitchMachine(BaseTest): + """Test update switch machine.""" + + def setUp(self): + super(TestUpdateSwitchMachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSwitchMachine, self).tearDown() + + def test_update_switch_machine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.update_switch_machine( + self.user_object, + 1, + 1, + tag='test_tag' + ) + update_switch_machine = switch.list_switch_machines( + self.user_object, + 1 + ) + expected = {'tag': 'test_tag'} + self.assertTrue( + item in update_switch_machine[0].items for item in expected.items() + ) + + +class TestUpdateSwitchmachine(BaseTest): + """Test update switch machine.""" + + def setUp(self): + super(TestUpdateSwitchmachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSwitchmachine, self).tearDown() + + def test_update_switchmachine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.update_switchmachine( + self.user_object, + 1, + location='test_location' + ) + update_switchmachine = switch.list_switchmachines( + self.user_object, + ) + expected = {'location': 'test_location'} + self.assertTrue( + item in update_switchmachine[0].items() + for item in expected.items() + ) + + +class TestPatchSwitchMachine(BaseTest): + """Test patch switch machine.""" + + def setUp(self): + super(TestPatchSwitchMachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestPatchSwitchMachine, self).tearDown() + + def test_pathc_switch_machine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.patch_switch_machine( + self.user_object, + 1, + 1, + patched_tag={ + 'patched_tag': 'test_patched_tag' + } + ) + switch_patch_switch_machine = switch.list_switch_machines( + self.user_object, + 1 + ) + expected = {'patched_tag': 'test_patched_tag'} + self.assertTrue( + item in switch_patch_switch_machine[0].items() + for item in expected.items() + ) + + +class TestPatchSwitchmachine(BaseTest): + """Test patch switch machine.""" + + def setUp(self): + super(TestPatchSwitchmachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestPatchSwitchmachine, self).tearDown() + + def test_patch_switchmachine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.patch_switchmachine( + self.user_object, + 1, + patched_location={ + 'patched_location': 'test_location' + } + ) + patch_switchmachine = switch.list_switchmachines( + self.user_object + ) + expected = {'patched_location': 'test_location'} + self.assertTrue( + item in patch_switchmachine[0].items() for item in expected.items() + ) + + +class TestDelSwitchMachine(BaseTest): + """Test delete switch machines.""" + + def setUp(self): + super(TestDelSwitchMachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestDelSwitchMachine, self).tearDown() + + def test_del_switch_machine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.del_switch_machine( + self.user_object, + 1, + 1 + ) + del_switch_machine = switch.list_switch_machines( + self.user_object, + 1 + ) + self.assertEqual([], del_switch_machine) + + +class TestDelSwitchmachine(BaseTest): + """Test delete switch machines.""" + + def setUp(self): + super(TestDelSwitchmachine, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestDelSwitchmachine, self).tearDown() + + def test_switchmachine(self): + switch.add_switch_machine( + self.user_object, + 1, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.del_switchmachine( + self.user_object, + 1 + ) + del_switchmachine = switch.list_switchmachines( + self.user_object + ) + self.assertEqual([], del_switchmachine) + + +class TestUpdateSwitchMachines(BaseTest): + """Test update switch machines.""" + + def setUp(self): + super(TestUpdateSwitchMachines, self).setUp() + logsetting.init() + + def tearDown(self): + super(TestUpdateSwitchMachines, self).tearDown() + + def test_update_switch_machines_remove(self): + switch.add_switch( + self.user_object, + ip='2887583784' + ) + switch.add_switch_machine( + self.user_object, + 2, + mac='28:6e:d4:46:c4:25', + port='1' + ) + switch.update_switch_machines( + self.user_object, + 2, + remove_machines=1 + ) + update_remove = switch.list_switch_machines( + self.user_object, + 2 + ) + self.assertEqual([], update_remove) + + +if __name__ == '__main__': + flags.init() + logsetting.init() + unittest2.main() diff --git a/compass/tests/db/api/test_user.py b/compass/tests/db/api/test_user.py index 513cf80e..9cf978bb 100644 --- a/compass/tests/db/api/test_user.py +++ b/compass/tests/db/api/test_user.py @@ -17,6 +17,7 @@ import logging import os import unittest2 +from base import BaseTest from compass.db.api import database from compass.db.api import user as user_api from compass.db import exception @@ -24,30 +25,7 @@ from compass.utils import flags from compass.utils import logsetting from compass.utils import setting_wrapper as setting os.environ['COMPASS_IGNORE_SETTING'] = 'true' - - -class BaseTest(unittest2.TestCase): - """Base Class for unit test.""" - - def setUp(self): - super(BaseTest, self).setUp() - reload(setting) - setting.CONFIG_DIR = os.path.join( - os.path.dirname(os.path.abspath(__file__)), - 'data' - ) - database.init('sqlite://') - database.create_db() - self.user_object = ( - user_api.get_user_object( - setting.COMPASS_ADMIN_EMAIL - ) - ) - - def tearDown(self): - database.drop_db() - reload(setting) - super(BaseTest, self).tearDown() +reload(setting) class TestGetUserObject(unittest2.TestCase): @@ -64,9 +42,8 @@ class TestGetUserObject(unittest2.TestCase): database.create_db() def tearDown(self): - database.drop_db() reload(setting) - super(TestGetUserObject, self).tearDown() + database.drop_db() def test_get_user_object(self): user_object = user_api.get_user_object(setting.COMPASS_ADMIN_EMAIL) @@ -138,11 +115,9 @@ class TestListUsers(BaseTest): def setUp(self): super(TestListUsers, self).setUp() - logsetting.init() def tearDown(self): super(TestListUsers, self).tearDown() - database.drop_db() def test_list_users(self): user = user_api.list_users(self.user_object) diff --git a/compass/tests/db/api/test_user_log.py b/compass/tests/db/api/test_user_log.py index 4b63198d..4e8c69a8 100644 --- a/compass/tests/db/api/test_user_log.py +++ b/compass/tests/db/api/test_user_log.py @@ -20,6 +20,7 @@ import unittest2 os.environ['COMPASS_IGNORE_SETTING'] = 'true' +from base import BaseTest from compass.utils import setting_wrapper as setting reload(setting) @@ -32,30 +33,6 @@ from compass.utils import flags from compass.utils import logsetting -class BaseTest(unittest2.TestCase): - """Base Class for unit test.""" - - def setUp(self): - super(BaseTest, self).setUp() - reload(setting) - setting.CONFIG_DIR = os.path.join( - os.path.dirname(os.path.abspath(__file__)), - 'data' - ) - database.init('sqlite://') - database.create_db() - self.user_object = ( - user_api.get_user_object( - setting.COMPASS_ADMIN_EMAIL, - ) - ) - - def tearDown(self): - database.drop_db() - reload(setting) - super(BaseTest, self).tearDown() - - class TestListUserActions(BaseTest): """Test user actions.""" diff --git a/requirements.txt b/requirements.txt index d06f6fd2..c37909df 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,6 +16,7 @@ MySQL-python netaddr paramiko==1.7.5 PyChef +sqlalchemy>=0.9.0 simplejson requests redis