From 2e98a1eb2d22155c4e7a6a49bc19a7c312d2eaa8 Mon Sep 17 00:00:00 2001 From: anand k Date: Wed, 20 Mar 2024 17:42:53 +0530 Subject: [PATCH] Test cases for helper_sql & Added mock library dependency in requirements.txt --- requirements.txt | 1 + src/tests/test_helper_sql.py | 131 +++++++++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+) create mode 100644 src/tests/test_helper_sql.py diff --git a/requirements.txt b/requirements.txt index ecb021cd..6f4a22fb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ coverage psutil pycryptodome PyQt5;python_version>="3.7" +mock;python_version<="2.7" python_prctl;platform_system=="Linux" six xvfbwrapper;platform_system=="Linux" diff --git a/src/tests/test_helper_sql.py b/src/tests/test_helper_sql.py new file mode 100644 index 00000000..036bd2c9 --- /dev/null +++ b/src/tests/test_helper_sql.py @@ -0,0 +1,131 @@ +"""Test cases for helper_sql""" + +import unittest + +try: + # Python 3 + from unittest.mock import patch +except ImportError: + # Python 2 + from mock import patch + +import pybitmessage.helper_sql as helper_sql + + +class TestHelperSql(unittest.TestCase): + """Test class for helper_sql""" + + @classmethod + def setUpClass(cls): + helper_sql.sql_available = True + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlReturnQueue.get") + def test_sqlquery_no_args(self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put): + """Test sqlQuery with no additional arguments""" + mock_sqlreturnqueue_get.return_value = ("dummy_result", None) + result = helper_sql.sqlQuery( + "SELECT msgid FROM inbox where folder='inbox' ORDER BY received" + ) + self.assertEqual(mock_sqlsubmitqueue_put.call_count, 2) + self.assertEqual(result, "dummy_result") + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlReturnQueue.get") + def test_sqlquery_with_args(self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put): + """Test sqlQuery with additional arguments""" + mock_sqlreturnqueue_get.return_value = ("dummy_result", None) + result = helper_sql.sqlQuery( + "SELECT address FROM addressbook WHERE address=?", "PB-5yfds868gbkj" + ) + self.assertEqual(mock_sqlsubmitqueue_put.call_count, 2) + self.assertEqual(result, "dummy_result") + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlReturnQueue.get") + def test_sqlexecute(self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put): + """Test sqlExecute with valid arguments""" + mock_sqlreturnqueue_get.return_value = (None, 1) + rowcount = helper_sql.sqlExecute( + "UPDATE sent SET status = 'msgqueued'" + "WHERE ackdata = ? AND folder = 'sent'", + "1710652313", + ) + self.assertEqual(mock_sqlsubmitqueue_put.call_count, 3) + self.assertEqual(rowcount, 1) + + @patch("pybitmessage.helper_sql.SqlBulkExecute.execute") + def test_sqlexecute_script(self, mock_execute): + """Test sqlExecuteScript with a SQL script""" + helper_sql.sqlExecuteScript( + "CREATE TABLE test (id INTEGER); INSERT INTO test VALUES (1);" + ) + self.assertTrue(mock_execute.assert_called) + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch( + "pybitmessage.helper_sql.sqlReturnQueue.get", + ) + def test_sqlexecute_chunked(self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put): + """Test sqlExecuteChunked with valid arguments""" + # side_effect is list of return value (_, rowcount) + # of sqlReturnQueue.get for each chunk + CHUNK_COUNT = 6 + CHUNK_SIZE = 999 + ID_COUNT = CHUNK_COUNT * CHUNK_SIZE + CHUNKS_ROWCOUNT_LIST = [50, 29, 28, 18, 678, 900] + TOTAL_ROW_COUNT = sum(CHUNKS_ROWCOUNT_LIST) + mock_sqlreturnqueue_get.side_effect = [(None, rowcount) for rowcount in CHUNKS_ROWCOUNT_LIST] + args = [] + for i in range(0, ID_COUNT): + args.append("arg{}".format(i)) + total_row_count_return = helper_sql.sqlExecuteChunked( + "INSERT INTO table VALUES {}", ID_COUNT, *args + ) + self.assertEqual(TOTAL_ROW_COUNT, total_row_count_return) + self.assertTrue(mock_sqlsubmitqueue_put.called) + self.assertTrue(mock_sqlreturnqueue_get.called) + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlReturnQueue.get") + def test_sqlexecute_chunked_with_idcount_zero( + self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put + ): + """Test sqlExecuteChunked with id count 0""" + ID_COUNT = 0 + args = list() + for i in range(0, ID_COUNT): + args.append("arg{}".format(i)) + total_row_count = helper_sql.sqlExecuteChunked( + "INSERT INTO table VALUES {}", ID_COUNT, *args + ) + self.assertEqual(total_row_count, 0) + self.assertFalse(mock_sqlsubmitqueue_put.called) + self.assertFalse(mock_sqlreturnqueue_get.called) + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlReturnQueue.get") + def test_sqlexecute_chunked_with_args_less( + self, mock_sqlreturnqueue_get, mock_sqlsubmitqueue_put + ): + """Test sqlExecuteChunked with length of args less than idcount""" + ID_COUNT = 12 + args = ["args0", "arg1"] + total_row_count = helper_sql.sqlExecuteChunked( + "INSERT INTO table VALUES {}", ID_COUNT, *args + ) + self.assertEqual(total_row_count, 0) + self.assertFalse(mock_sqlsubmitqueue_put.called) + self.assertFalse(mock_sqlreturnqueue_get.called) + + @patch("pybitmessage.helper_sql.sqlSubmitQueue.put") + @patch("pybitmessage.helper_sql.sqlSubmitQueue.task_done") + def test_sqlstored_procedure(self, mock_task_done, mock_sqlsubmitqueue_put): + """Test sqlStoredProcedure with a stored procedure name""" + helper_sql.sqlStoredProcedure("exit") + self.assertTrue(mock_task_done.called_once) + mock_sqlsubmitqueue_put.assert_called_with("terminate") + + @classmethod + def tearDownClass(cls): + helper_sql.sql_available = False