Update db layer api tests

Change-Id: I10b594d95b148fb9233c040da31c1ba6d9beaf72

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(

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 datetime
-import copy
+import logging
-import mock
 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
+class BaseTest(unittest2.TestCase):
-from compass.db.api import adapter_holder as adapter_api
+    """Base Class for unit test."""
-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."""
 
     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()
+        database.drop_db()
-
-    def test_login(self):
-        pass
-
-
-if __name__ == '__main__':
-    flags.init()
-    logsetting.init()
-    unittest2.main()

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."""
 

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()

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()

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'
-
+reload(setting)
-
-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 TestListPermissions(BaseTest):

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()

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'
-
+reload(setting)
-
-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 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)

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."""
 

requirements.txt

@@ -16,6 +16,7 @@ MySQL-python
 netaddr
 paramiko==1.7.5
 PyChef
+sqlalchemy>=0.9.0
 simplejson
 requests
 redis