diff options
author | 2017-07-27 09:02:54 +0300 | |
---|---|---|
committer | 2017-07-27 14:56:25 +0300 | |
commit | 7e83d0876ddb84a45e130eeba28bc40ef53c074b (patch) | |
tree | 47d76239ae7658d87c66abd142df92709427e7dd /app/test/fetch/db_fetch/test_db_access.py | |
parent | 378ecbd8947589b9cbb39013a0c2e2aa201e03bd (diff) |
Calipso initial release for OPNFV
Change-Id: I7210c244b0c10fa80bfa8c77cb86c9d6ddf8bc88
Signed-off-by: Yaron Yogev <yaronyogev@gmail.com>
Diffstat (limited to 'app/test/fetch/db_fetch/test_db_access.py')
-rw-r--r-- | app/test/fetch/db_fetch/test_db_access.py | 108 |
1 files changed, 108 insertions, 0 deletions
diff --git a/app/test/fetch/db_fetch/test_db_access.py b/app/test/fetch/db_fetch/test_db_access.py new file mode 100644 index 0000000..4ef3e74 --- /dev/null +++ b/app/test/fetch/db_fetch/test_db_access.py @@ -0,0 +1,108 @@ +############################################################################### +# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) # +# and others # +# # +# All rights reserved. This program and the accompanying materials # +# are made available under the terms of the Apache License, Version 2.0 # +# which accompanies this distribution, and is available at # +# http://www.apache.org/licenses/LICENSE-2.0 # +############################################################################### +from discover.fetchers.db.db_access import DbAccess +from test.fetch.test_fetch import TestFetch +from test.fetch.db_fetch.test_data.db_access import * +from unittest.mock import MagicMock, patch +from test.fetch.db_fetch.mock_cursor import MockCursor + + +class TestDbAccess(TestFetch): + + def setUp(self): + self.configure_environment() + self.fetcher = DbAccess() + + @patch("mysql.connector.connect") + def test_db_connect(self, db_connect): + DbAccess.conn = None + db_conn = MagicMock() + db_conn.ping = MagicMock() + db_connect.return_value = db_conn + + self.fetcher.db_connect(DB_CONFIG['host'], DB_CONFIG['port'], + DB_CONFIG['user'], DB_CONFIG['password'], + DB_CONFIG['schema']) + + self.assertEqual(True, db_connect.called, "connect method has't been called") + db_conn.ping.assert_called_once_with(True) + + def test_connect_to_db(self): + DbAccess.conn = None + self.fetcher.db_connect = MagicMock() + self.fetcher.connect_to_db() + + self.assertEqual(True, self.fetcher.db_connect.called) + + def test_connect_to_db_with_force(self): + DbAccess.conn = MagicMock() + self.fetcher.db_connect = MagicMock() + self.fetcher.connect_to_db(force=True) + + self.assertEqual(True, self.fetcher.db_connect.called) + + def test_connect_to_db_without_force(self): + DbAccess.conn = MagicMock() + self.fetcher.db_connect = MagicMock() + self.fetcher.connect_to_db() + + self.assertEqual(False, self.fetcher.db_connect.called) + + def test_get_objects_list_for_id_with_id(self): + # mock the db cursor + mock_cursor = MockCursor(OBJECTS_LIST) + mock_cursor.execute = MagicMock() + + self.fetcher.connect_to_db = MagicMock() + DbAccess.conn.cursor = MagicMock(return_value=mock_cursor) + + result = self.fetcher.get_objects_list_for_id(QUERY_WITH_ID, OBJECT_TYPE, ID) + + mock_cursor.execute.assert_called_once_with(QUERY_WITH_ID, [ID]) + self.assertEqual(result, OBJECTS_LIST, "Can't get objects list") + + def test_get_objects_list_for_id_without_id(self): + mock_cursor = MockCursor(OBJECTS_LIST) + + self.fetcher.connect_to_db = MagicMock() + DbAccess.conn.cursor = MagicMock(return_value=mock_cursor) + mock_cursor.execute = MagicMock() + + result = self.fetcher.get_objects_list_for_id(QUERY_WITHOUT_ID, OBJECT_TYPE, None) + + mock_cursor.execute.assert_called_once_with(QUERY_WITHOUT_ID) + self.assertEqual(result, OBJECTS_LIST, "Can't get objects list") + + def test_get_objects_list_for_id_with_id_with_exception(self): + mock_cursor = MockCursor(OBJECTS_LIST) + self.fetcher.connect_to_db = MagicMock() + # mock exception + DbAccess.conn.cursor = MagicMock(return_value=mock_cursor) + mock_cursor.execute = MagicMock(side_effect=[AttributeError, ""]) + + result = self.fetcher.get_objects_list_for_id(QUERY_WITH_ID, OBJECT_TYPE, ID) + + self.assertEqual(mock_cursor.execute.call_count, 2, "Can't invoke execute method " + + "twice when error occurs") + self.assertEqual(result, OBJECTS_LIST, "Can't get objects list") + + def test_get_objects_list_for_id_without_id_with_exception(self): + mock_cursor = MockCursor(OBJECTS_LIST) + self.fetcher.connect_to_db = MagicMock() + DbAccess.conn.cursor = MagicMock(return_value=mock_cursor) + mock_cursor.execute = MagicMock(side_effect=[AttributeError, ""]) + + result = self.fetcher.get_objects_list_for_id(QUERY_WITHOUT_ID, + OBJECT_TYPE, + None) + + self.assertEqual(mock_cursor.execute.call_count, 2, "Can't invoke execute method " + + "twice when error occurs") + self.assertEqual(result, OBJECTS_LIST, "Can't get objects list") |