From 523e61c9f7c5bdfbbcad65d07607e04d0a4a70cf Mon Sep 17 00:00:00 2001 From: Clemens Schwaighofer Date: Thu, 18 Dec 2025 17:20:57 +0900 Subject: [PATCH] Add SQL Main class as general wrapper for SQL DB handling --- src/corelibs/db_handling/sql_main.py | 76 ++++ test-run/db_handling/sql_main.py | 139 +++++++ test-run/db_handling/sqlite_io.py | 4 +- tests/unit/db_handling/test_sql_main.py | 461 ++++++++++++++++++++++++ 4 files changed, 677 insertions(+), 3 deletions(-) create mode 100644 src/corelibs/db_handling/sql_main.py create mode 100644 test-run/db_handling/sql_main.py create mode 100644 tests/unit/db_handling/test_sql_main.py diff --git a/src/corelibs/db_handling/sql_main.py b/src/corelibs/db_handling/sql_main.py new file mode 100644 index 0000000..a8d5a25 --- /dev/null +++ b/src/corelibs/db_handling/sql_main.py @@ -0,0 +1,76 @@ +""" +Main SQL base for any SQL calls +This is a wrapper for SQLiteIO or other future DB Interfaces +[Note: at the moment only SQLiteIO is implemented] +- on class creation connection with ValueError on fail +- connect method checks if already connected and warns +- connection class fails with ValueError if not valid target is selected (SQL wrapper type) +- connected check class method +- a process class that returns data as list or False if end or error + +TODO: adapt more CoreLibs DB IO class flow here +""" + +from typing import TYPE_CHECKING, Any, Literal +from corelibs.debug_handling.debug_helpers import call_stack +from corelibs.db_handling.sqlite_io import SQLiteIO +if TYPE_CHECKING: + from corelibs.logging_handling.log import Logger + + +IDENT_SPLIT_CHARACTER: str = ':' + + +class SQLMain: + """Main SQL interface class""" + def __init__(self, log: 'Logger', db_ident: str): + self.log = log + self.dbh: SQLiteIO | None = None + self.db_target: str | None = None + self.connect(db_ident) + if not self.connected(): + raise ValueError(f'Failed to connect to database [{call_stack()}]') + + def connect(self, db_ident: str): + """setup basic connection""" + if self.dbh is not None and self.dbh.conn is not None: + self.log.warning(f"A database connection already exists for: {self.db_target} [{call_stack()}]") + return + self.db_target, db_dsn = db_ident.split(IDENT_SPLIT_CHARACTER) + match self.db_target: + case 'sqlite': + # this is a Path only at the moment + self.dbh = SQLiteIO(self.log, db_dsn, row_factory='Dict') + case _: + raise ValueError(f'SQL interface for {self.db_target} is not implemented [{call_stack()}]') + if not self.dbh.db_connected(): + raise ValueError(f"DB Connection failed for: {self.db_target} [{call_stack()}]") + + def close(self): + """close connection""" + if self.dbh is None or not self.connected(): + return + # self.log.info(f"Close DB Connection: {self.db_target} [{call_stack()}]") + self.dbh.db_close() + + def connected(self) -> bool: + """check connectuon""" + if self.dbh is None or not self.dbh.db_connected(): + self.log.warning(f"No connection [{call_stack()}]") + return False + return True + + def process_query( + self, query: str, params: tuple[Any, ...] | None = None + ) -> list[tuple[Any, ...]] | list[dict[str, Any]] | Literal[False]: + """mini wrapper for execute query""" + if self.dbh is not None: + result = self.dbh.execute_query(query, params) + if result is False: + return False + else: + self.log.error(f"Problem connecting to db: {self.db_target} [{call_stack()}]") + return False + return result + +# __END__ diff --git a/test-run/db_handling/sql_main.py b/test-run/db_handling/sql_main.py new file mode 100644 index 0000000..747da3d --- /dev/null +++ b/test-run/db_handling/sql_main.py @@ -0,0 +1,139 @@ +""" +SQL Main wrapper test +""" + +from pathlib import Path +from uuid import uuid4 +import json +from corelibs.debug_handling.dump_data import dump_data +from corelibs.logging_handling.log import Log, Logger +from corelibs.db_handling.sql_main import SQLMain + +SCRIPT_PATH: Path = Path(__file__).resolve().parent +ROOT_PATH: Path = SCRIPT_PATH +DATABASE_DIR: Path = Path("database") +LOG_DIR: Path = Path("log") + + +def main() -> None: + """ + Comment + """ + log = Log( + log_path=ROOT_PATH.joinpath(LOG_DIR, 'sqlite_main.log'), + log_name="SQLite Main", + log_settings={ + "log_level_console": 'DEBUG', + "log_level_file": 'DEBUG', + } + ) + sql_main = SQLMain( + log=Logger(log.get_logger_settings()), + db_ident=f"sqlite:{ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_main.db')}" + ) + if sql_main.connected(): + log.info("SQL Main connected successfully") + else: + log.error('SQL Main connection failed') + if sql_main.dbh is None: + log.error('SQL Main DBH instance is None') + return + + if sql_main.dbh.trigger_exists('trg_test_a_set_date_updated_on_update'): + log.info("Trigger trg_test_a_set_date_updated_on_update exists") + if sql_main.dbh.table_exists('test_a'): + log.info("Table test_a exists, dropping for clean test") + sql_main.dbh.execute_query("DROP TABLE test_a;") + # create a dummy table + table_sql = """ + CREATE TABLE IF NOT EXISTS test_a ( + test_a_id INTEGER PRIMARY KEY, + date_created TEXT DEFAULT (strftime('%Y-%m-%d %H:%M:%f', 'now')), + date_updated TEXT, + uid TEXT NOT NULL UNIQUE, + set_current_timestamp TEXT DEFAULT CURRENT_TIMESTAMP, + text_a TEXT, + content, + int_a INTEGER, + float_a REAL + ); + """ + + result = sql_main.dbh.execute_query(table_sql) + log.debug(f"Create table result: {result}") + trigger_sql = """ + CREATE TRIGGER trg_test_a_set_date_updated_on_update + AFTER UPDATE ON test_a + FOR EACH ROW + WHEN OLD.date_updated IS NULL OR NEW.date_updated = OLD.date_updated + BEGIN + UPDATE test_a + SET date_updated = (strftime('%Y-%m-%d %H:%M:%f', 'now')) + WHERE test_a_id = NEW.test_a_id; + END; + """ + result = sql_main.dbh.execute_query(trigger_sql) + log.debug(f"Create trigger result: {result}") + result = sql_main.dbh.meta_data_detail('test_a') + log.debug(f"Table meta data detail: {dump_data(result)}") + # INSERT DATA + sql = """ + INSERT INTO test_a (uid, text_a, content, int_a, float_a) + VALUES (?, ?, ?, ?, ?) + RETURNING test_a_id, uid; + """ + result = sql_main.dbh.execute_query( + sql, + ( + str(uuid4()), + 'Some text A', + json.dumps({'foo': 'bar', 'number': 42}), + 123, + 123.456, + ) + ) + log.debug(f"[1] Insert data result: {dump_data(result)}") + __uid: str = '' + if result is not False: + # first one only of interest + result = dict(result[0]) + __uid = str(result.get('uid', '')) + # second insert + result = sql_main.dbh.execute_query( + sql, + ( + str(uuid4()), + 'Some text A', + json.dumps({'foo': 'bar', 'number': 42}), + 123, + 123.456, + ) + ) + log.debug(f"[2] Insert data result: {dump_data(result)}") + result = sql_main.dbh.execute_query("SELECT * FROM test_a;") + log.debug(f"Select data result: {dump_data(result)}") + result = sql_main.dbh.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,)) + log.debug(f"Fetch row result: {dump_data(result)}") + sql = """ + UPDATE test_a + SET text_a = ? + WHERE uid = ?; + """ + result = sql_main.dbh.execute_query( + sql, + ( + 'Some updated text A', + __uid, + ) + ) + log.debug(f"Update data result: {dump_data(result)}") + result = sql_main.dbh.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,)) + log.debug(f"Fetch row after update result: {dump_data(result)}") + + sql_main.close() + + +if __name__ == "__main__": + main() + +# __END__ diff --git a/test-run/db_handling/sqlite_io.py b/test-run/db_handling/sqlite_io.py index 1afb581..51f538f 100644 --- a/test-run/db_handling/sqlite_io.py +++ b/test-run/db_handling/sqlite_io.py @@ -1,7 +1,5 @@ -#!/usr/bin/env python3 - """ -Main comment +SQLite IO test """ from pathlib import Path diff --git a/tests/unit/db_handling/test_sql_main.py b/tests/unit/db_handling/test_sql_main.py new file mode 100644 index 0000000..31cd3e9 --- /dev/null +++ b/tests/unit/db_handling/test_sql_main.py @@ -0,0 +1,461 @@ +""" +PyTest: db_handling/sql_main +Tests for SQLMain class - Main SQL interface wrapper + +Note: Pylance warnings about "Redefining name from outer scope" in fixtures are expected. +This is standard pytest fixture behavior where fixture parameters shadow fixture definitions. +""" +# pylint: disable=redefined-outer-name,too-many-public-methods,protected-access +# pyright: reportUnknownParameterType=false, reportUnknownArgumentType=false +# pyright: reportMissingParameterType=false, reportUnknownVariableType=false +# pyright: reportArgumentType=false, reportGeneralTypeIssues=false + +from pathlib import Path +from typing import Generator +from unittest.mock import MagicMock, patch +import pytest +from corelibs.db_handling.sql_main import SQLMain, IDENT_SPLIT_CHARACTER +from corelibs.db_handling.sqlite_io import SQLiteIO + + +# Test fixtures +@pytest.fixture +def mock_logger() -> MagicMock: + """Create a mock logger for testing""" + logger = MagicMock() + logger.debug = MagicMock() + logger.info = MagicMock() + logger.warning = MagicMock() + logger.error = MagicMock() + return logger + + +@pytest.fixture +def temp_db_path(tmp_path: Path) -> Path: + """Create a temporary database file path""" + return tmp_path / "test_database.db" + + +@pytest.fixture +def mock_sqlite_io() -> Generator[MagicMock, None, None]: + """Create a mock SQLiteIO instance""" + mock_io = MagicMock(spec=SQLiteIO) + mock_io.conn = MagicMock() + mock_io.db_connected = MagicMock(return_value=True) + mock_io.db_close = MagicMock() + mock_io.execute_query = MagicMock(return_value=[]) + yield mock_io + + +# Test constant +class TestConstants: + """Tests for module-level constants""" + + def test_ident_split_character(self): + """Test that IDENT_SPLIT_CHARACTER is defined correctly""" + assert IDENT_SPLIT_CHARACTER == ':' + + +# Test SQLMain class initialization +class TestSQLMainInit: + """Tests for SQLMain.__init__""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_successful_initialization_sqlite( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test successful initialization with SQLite""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + assert sql_main.log == mock_logger + assert sql_main.dbh == mock_sqlite_instance + assert sql_main.db_target == 'sqlite' + mock_sqlite_class.assert_called_once_with(mock_logger, str(temp_db_path), row_factory='Dict') + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_initialization_connection_failure(self, mock_sqlite_class: MagicMock, mock_logger: MagicMock): + """Test initialization fails when connection cannot be established""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = None + mock_sqlite_instance.db_connected = MagicMock(return_value=False) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = 'sqlite:/path/to/db.db' + with pytest.raises(ValueError, match='DB Connection failed for: sqlite'): + SQLMain(mock_logger, db_ident) + + def test_initialization_invalid_db_target(self, mock_logger: MagicMock): + """Test initialization with unsupported database target""" + db_ident = 'postgresql:/path/to/db' + with pytest.raises(ValueError, match='SQL interface for postgresql is not implemented'): + SQLMain(mock_logger, db_ident) + + def test_initialization_malformed_db_ident(self, mock_logger: MagicMock): + """Test initialization with malformed db_ident string""" + db_ident = 'sqlite_no_colon' + with pytest.raises(ValueError): + SQLMain(mock_logger, db_ident) + + +# Test SQLMain.connect method +class TestSQLMainConnect: + """Tests for SQLMain.connect""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_connect_when_already_connected( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test connect warns when already connected""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + # Reset mock to check second call + mock_logger.warning.reset_mock() + + # Try to connect again + sql_main.connect(f'sqlite:{temp_db_path}') + + # Should have warned about existing connection + mock_logger.warning.assert_called_once() + assert 'already exists' in str(mock_logger.warning.call_args) + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_connect_sqlite_success( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test successful SQLite connection""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + sql_main = SQLMain.__new__(SQLMain) + sql_main.log = mock_logger + sql_main.dbh = None + sql_main.db_target = None + + db_ident = f'sqlite:{temp_db_path}' + sql_main.connect(db_ident) + + assert sql_main.db_target == 'sqlite' + assert sql_main.dbh == mock_sqlite_instance + mock_sqlite_class.assert_called_once_with(mock_logger, str(temp_db_path), row_factory='Dict') + + def test_connect_unsupported_database(self, mock_logger: MagicMock): + """Test connect with unsupported database type""" + sql_main = SQLMain.__new__(SQLMain) + sql_main.log = mock_logger + sql_main.dbh = None + sql_main.db_target = None + + db_ident = 'mysql:/path/to/db' + with pytest.raises(ValueError, match='SQL interface for mysql is not implemented'): + sql_main.connect(db_ident) + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_connect_db_connection_failed( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test connect raises error when DB connection fails""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=False) + mock_sqlite_class.return_value = mock_sqlite_instance + + sql_main = SQLMain.__new__(SQLMain) + sql_main.log = mock_logger + sql_main.dbh = None + sql_main.db_target = None + + db_ident = f'sqlite:{temp_db_path}' + with pytest.raises(ValueError, match='DB Connection failed for: sqlite'): + sql_main.connect(db_ident) + + +# Test SQLMain.close method +class TestSQLMainClose: + """Tests for SQLMain.close""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_close_successful( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test successful database close""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.db_close = MagicMock() + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + sql_main.close() + + mock_sqlite_instance.db_close.assert_called_once() + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_close_when_not_connected( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test close when not connected does nothing""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.db_close = MagicMock() + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + # Change db_connected to return False to simulate disconnection + mock_sqlite_instance.db_connected = MagicMock(return_value=False) + + sql_main.close() + + # Should not raise error and should exit early + assert mock_sqlite_instance.db_close.call_count == 0 + + def test_close_when_dbh_is_none(self, mock_logger: MagicMock): + """Test close when dbh is None""" + sql_main = SQLMain.__new__(SQLMain) + sql_main.log = mock_logger + sql_main.dbh = None + sql_main.db_target = 'sqlite' + + # Should not raise error + sql_main.close() + + +# Test SQLMain.connected method +class TestSQLMainConnected: + """Tests for SQLMain.connected""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_connected_returns_true( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test connected returns True when connected""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + assert sql_main.connected() is True + mock_logger.warning.assert_not_called() + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_connected_returns_false_when_not_connected( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test connected returns False and warns when not connected""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + # Reset warning calls from init + mock_logger.warning.reset_mock() + + # Change db_connected to return False to simulate disconnection + mock_sqlite_instance.db_connected = MagicMock(return_value=False) + + assert sql_main.connected() is False + mock_logger.warning.assert_called_once() + assert 'No connection' in str(mock_logger.warning.call_args) + + def test_connected_returns_false_when_dbh_is_none(self, mock_logger: MagicMock): + """Test connected returns False when dbh is None""" + sql_main = SQLMain.__new__(SQLMain) + sql_main.log = mock_logger + sql_main.dbh = None + sql_main.db_target = 'sqlite' + + assert sql_main.connected() is False + mock_logger.warning.assert_called_once() + + +# Test SQLMain.process_query method +class TestSQLMainProcessQuery: + """Tests for SQLMain.process_query""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_process_query_success_no_params( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test successful query execution without parameters""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + expected_result = [{'id': 1, 'name': 'test'}] + mock_sqlite_instance.execute_query = MagicMock(return_value=expected_result) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + query = "SELECT * FROM test" + result = sql_main.process_query(query) + + assert result == expected_result + mock_sqlite_instance.execute_query.assert_called_once_with(query, None) + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_process_query_success_with_params( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test successful query execution with parameters""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + expected_result = [{'id': 1, 'name': 'test'}] + mock_sqlite_instance.execute_query = MagicMock(return_value=expected_result) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + query = "SELECT * FROM test WHERE id = ?" + params = (1,) + result = sql_main.process_query(query, params) + + assert result == expected_result + mock_sqlite_instance.execute_query.assert_called_once_with(query, params) + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_process_query_returns_false_on_error( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test query returns False when execute_query fails""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.execute_query = MagicMock(return_value=False) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + query = "SELECT * FROM nonexistent" + result = sql_main.process_query(query) + + assert result is False + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_process_query_dbh_is_none( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test query returns False when dbh is None""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + # Manually set dbh to None + sql_main.dbh = None + + query = "SELECT * FROM test" + result = sql_main.process_query(query) + + assert result is False + mock_logger.error.assert_called_once() + assert 'Problem connecting to db' in str(mock_logger.error.call_args) + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_process_query_returns_empty_list( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test query returns empty list when no results""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.execute_query = MagicMock(return_value=[]) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + query = "SELECT * FROM test WHERE 1=0" + result = sql_main.process_query(query) + + assert result == [] + + +# Integration-like tests +class TestSQLMainIntegration: + """Integration-like tests for complete workflows""" + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_full_workflow_connect_query_close( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test complete workflow: connect, query, close""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.execute_query = MagicMock(return_value=[{'count': 5}]) + mock_sqlite_instance.db_close = MagicMock() + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + # Execute query + result = sql_main.process_query("SELECT COUNT(*) as count FROM test") + assert result == [{'count': 5}] + + # Check connected + assert sql_main.connected() is True + + # Close connection + sql_main.close() + mock_sqlite_instance.db_close.assert_called_once() + + @patch('corelibs.db_handling.sql_main.SQLiteIO') + def test_multiple_queries_same_connection( + self, mock_sqlite_class: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ): + """Test multiple queries on the same connection""" + mock_sqlite_instance = MagicMock() + mock_sqlite_instance.conn = MagicMock() + mock_sqlite_instance.db_connected = MagicMock(return_value=True) + mock_sqlite_instance.execute_query = MagicMock(side_effect=[ + [{'id': 1}], + [{'id': 2}], + [{'id': 3}] + ]) + mock_sqlite_class.return_value = mock_sqlite_instance + + db_ident = f'sqlite:{temp_db_path}' + sql_main = SQLMain(mock_logger, db_ident) + + result1 = sql_main.process_query("SELECT * FROM test WHERE id = 1") + result2 = sql_main.process_query("SELECT * FROM test WHERE id = 2") + result3 = sql_main.process_query("SELECT * FROM test WHERE id = 3") + + assert result1 == [{'id': 1}] + assert result2 == [{'id': 2}] + assert result3 == [{'id': 3}] + assert mock_sqlite_instance.execute_query.call_count == 3 + + +# __END__