diff --git a/pyproject.toml b/pyproject.toml index 229003c..1e1cea3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,3 +63,7 @@ ignore = [ [tool.pylint.MASTER] # this is for the tests/etc folders init-hook='import sys; sys.path.append("src/")' +[tool.pytest.ini_options] +testpaths = [ + "tests", +] diff --git a/src/corelibs/db_handling/__init__.py b/src/corelibs/db_handling/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/corelibs/db_handling/sqlite_io.py b/src/corelibs/db_handling/sqlite_io.py new file mode 100644 index 0000000..3c11ef2 --- /dev/null +++ b/src/corelibs/db_handling/sqlite_io.py @@ -0,0 +1,214 @@ +""" +SQLite DB::IO +Will be moved to the CoreLibs +also method names are subject to change +""" + +# import gc +from pathlib import Path +from typing import Any, Literal, TYPE_CHECKING +import sqlite3 +from corelibs.debug_handling.debug_helpers import call_stack +if TYPE_CHECKING: + from corelibs.logging_handling.log import Logger + + +class SQLiteIO(): + """Mini SQLite interface""" + + def __init__( + self, + log: 'Logger', + db_name: str | Path, + autocommit: bool = False, + enable_fkey: bool = True, + row_factory: str | None = None + ): + self.log = log + self.db_name = db_name + self.autocommit = autocommit + self.enable_fkey = enable_fkey + self.row_factory = row_factory + self.conn: sqlite3.Connection | None = self.db_connect() + + # def __del__(self): + # self.db_close() + + def db_connect(self) -> sqlite3.Connection | None: + """ + Connect to SQLite database, create if it doesn't exist + """ + try: + # Connect to database (creates if doesn't exist) + self.conn = sqlite3.connect(self.db_name, autocommit=self.autocommit) + self.conn.setconfig(sqlite3.SQLITE_DBCONFIG_ENABLE_FKEY, True) + # self.conn.execute("PRAGMA journal_mode=WAL") + # self.log.debug(f"Connected to database: {self.db_name}") + + def dict_factory(cursor: sqlite3.Cursor, row: list[Any]): + fields = [column[0] for column in cursor.description] + return dict(zip(fields, row)) + + match self.row_factory: + case 'Row': + self.conn.row_factory = sqlite3.Row + case 'Dict': + self.conn.row_factory = dict_factory + case _: + self.conn.row_factory = None + + return self.conn + except (sqlite3.Error, sqlite3.OperationalError) as e: + self.log.error(f"Error connecting to database [{type(e).__name__}] [{self.db_name}]: {e} [{call_stack()}]") + self.log.error(f"Error code: {e.sqlite_errorcode if hasattr(e, 'sqlite_errorcode') else 'N/A'}") + self.log.error(f"Error name: {e.sqlite_errorname if hasattr(e, 'sqlite_errorname') else 'N/A'}") + return None + + def db_close(self): + """close connection""" + if self.conn is not None: + self.conn.close() + self.conn = None + + def db_connected(self) -> bool: + """ + Return True if db connection is not none + + Returns: + bool -- _description_ + """ + return True if self.conn else False + + def __content_exists(self, content_name: str, sql_type: str) -> bool: + """ + Check if some content name for a certain type exists + + Arguments: + content_name {str} -- _description_ + sql_type {str} -- _description_ + + Returns: + bool -- _description_ + """ + if self.conn is None: + return False + try: + cursor = self.conn.cursor() + cursor.execute(""" + SELECT name + FROM sqlite_master + WHERE type = ? AND name = ? + """, (sql_type, content_name,)) + return cursor.fetchone() is not None + except sqlite3.Error as e: + self.log.error(f"Error checking table [{content_name}/{sql_type}] existence: {e} [{call_stack()}]") + return False + + def table_exists(self, table_name: str) -> bool: + """ + Check if a table exists in the database + """ + return self.__content_exists(table_name, 'table') + + def trigger_exists(self, trigger_name: str) -> bool: + """ + Check if a triggere exits + """ + return self.__content_exists(trigger_name, 'trigger') + + def index_exists(self, index_name: str) -> bool: + """ + Check if a triggere exits + """ + return self.__content_exists(index_name, 'index') + + def meta_data_detail(self, table_name: str) -> list[tuple[Any, ...]] | list[dict[str, Any]] | Literal[False]: + """table detail""" + query_show_table = """ + SELECT + ti.cid, ti.name, ti.type, ti.'notnull', ti.dflt_value, ti.pk, + il_ii.idx_name, il_ii.idx_unique, il_ii.idx_origin, il_ii.idx_partial + FROM + sqlite_schema AS m, + pragma_table_info(m.name) AS ti + LEFT JOIN ( + SELECT + il.name AS idx_name, il.'unique' AS idx_unique, il.origin AS idx_origin, il.partial AS idx_partial, + ii.cid AS tbl_cid + FROM + sqlite_schema AS m, + pragma_index_list(m.name) AS il, + pragma_index_info(il.name) AS ii + WHERE m.name = ?1 + ) AS il_ii ON (ti.cid = il_ii.tbl_cid) + WHERE + m.name = ?1 + """ + return self.execute_query(query_show_table, (table_name,)) + + def execute_cursor( + self, query: str, params: tuple[Any, ...] | None = None + ) -> sqlite3.Cursor | Literal[False]: + """execute a cursor, used in execute query or return one and for fetch_row""" + if self.conn is None: + self.log.warning(f"No connection [{call_stack()}]") + return False + try: + cursor = self.conn.cursor() + if params: + cursor.execute(query, params) + else: + cursor.execute(query) + return cursor + except sqlite3.Error as e: + self.log.error(f"Error during executing cursor [{query}:{params}]: {e} [{call_stack()}]") + return False + + def execute_query( + self, query: str, params: tuple[Any, ...] | None = None + ) -> list[tuple[Any, ...]] | list[dict[str, Any]] | Literal[False]: + """query execute with or without params, returns result""" + if self.conn is None: + self.log.warning(f"No connection [{call_stack()}]") + return False + try: + if (cursor := self.execute_cursor(query, params)) is False: + return False + # fetch before commit because we need to get the RETURN before + result = cursor.fetchall() + # this is for INSERT/UPDATE/CREATE only + self.conn.commit() + return result + except sqlite3.Error as e: + self.log.error(f"Error during executing query [{query}:{params}]: {e} [{call_stack()}]") + return False + + def return_one( + self, query: str, params: tuple[Any, ...] | None = None + ) -> tuple[Any, ...] | dict[str, Any] | Literal[False] | None: + """return one row, only for SELECT""" + if self.conn is None: + self.log.warning(f"No connection [{call_stack()}]") + return False + try: + if (cursor := self.execute_cursor(query, params)) is False: + return False + return cursor.fetchone() + except sqlite3.Error as e: + self.log.error(f"Error during return one: {e} [{call_stack()}]") + return False + + def fetch_row( + self, cursor: sqlite3.Cursor | Literal[False] + ) -> tuple[Any, ...] | dict[str, Any] | Literal[False] | None: + """read from cursor""" + if self.conn is None or cursor is False: + self.log.warning(f"No connection [{call_stack()}]") + return False + try: + return cursor.fetchone() + except sqlite3.Error as e: + self.log.error(f"Error during fetch row: {e} [{call_stack()}]") + return False + +# __END__ diff --git a/test-run/config_handling/settings_loader.py b/test-run/config_handling/settings_loader.py index 59e24f5..3e24fad 100644 --- a/test-run/config_handling/settings_loader.py +++ b/test-run/config_handling/settings_loader.py @@ -12,6 +12,7 @@ from corelibs.config_handling.settings_loader_handling.settings_loader_check imp SCRIPT_PATH: Path = Path(__file__).resolve().parent ROOT_PATH: Path = SCRIPT_PATH CONFIG_DIR: Path = Path("config") +LOG_DIR: Path = Path("log") CONFIG_FILE: str = "settings.ini" @@ -26,9 +27,8 @@ def main(): print(f"regex {regex_c} check against {value} -> {result}") # for log testing - script_path: Path = Path(__file__).resolve().parent log = Log( - log_path=script_path.joinpath('log', 'settings_loader.log'), + log_path=ROOT_PATH.joinpath(LOG_DIR, 'settings_loader.log'), log_name="Settings Loader", log_settings={ "log_level_console": 'DEBUG', diff --git a/test-run/db_handling/database/.gitignore b/test-run/db_handling/database/.gitignore new file mode 100644 index 0000000..d6b7ef3 --- /dev/null +++ b/test-run/db_handling/database/.gitignore @@ -0,0 +1,2 @@ +* +!.gitignore diff --git a/test-run/db_handling/log/.gitignore b/test-run/db_handling/log/.gitignore new file mode 100644 index 0000000..d6b7ef3 --- /dev/null +++ b/test-run/db_handling/log/.gitignore @@ -0,0 +1,2 @@ +* +!.gitignore diff --git a/test-run/db_handling/sqlite_io.py b/test-run/db_handling/sqlite_io.py new file mode 100644 index 0000000..1afb581 --- /dev/null +++ b/test-run/db_handling/sqlite_io.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 + +""" +Main comment +""" + +from pathlib import Path +from uuid import uuid4 +import json +import sqlite3 +from corelibs.debug_handling.dump_data import dump_data +from corelibs.logging_handling.log import Log, Logger +from corelibs.db_handling.sqlite_io import SQLiteIO + +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_io.log'), + log_name="SQLite IO", + log_settings={ + "log_level_console": 'DEBUG', + "log_level_file": 'DEBUG', + } + ) + db = SQLiteIO( + log=Logger(log.get_logger_settings()), + db_name=ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_io.db'), + row_factory='Dict' + ) + if db.db_connected(): + log.info(f"Connected to DB: {db.db_name}") + if db.trigger_exists('trg_test_a_set_date_updated_on_update'): + log.info("Trigger trg_test_a_set_date_updated_on_update exists") + if db.table_exists('test_a'): + log.info("Table test_a exists, dropping for clean test") + db.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 = db.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 = db.execute_query(trigger_sql) + log.debug(f"Create trigger result: {result}") + result = db.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 = db.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 = db.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 = db.execute_query("SELECT * FROM test_a;") + log.debug(f"Select data result: {dump_data(result)}") + result = db.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 = db.execute_query( + sql, + ( + 'Some updated text A', + __uid, + ) + ) + log.debug(f"Update data result: {dump_data(result)}") + result = db.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,)) + log.debug(f"Fetch row after update result: {dump_data(result)}") + + db.db_close() + + db = SQLiteIO( + log=Logger(log.get_logger_settings()), + db_name=ROOT_PATH.joinpath(DATABASE_DIR, 'test_sqlite_io.db'), + row_factory='Row' + ) + result = db.return_one("SELECT * FROM test_a WHERE uid = ?;", (__uid,)) + if result is not None and result is not False: + log.debug(f"Fetch row result: {dump_data(result)} -> {dict(result)} -> {result.keys()}") + log.debug(f"Access via index: {result[5]} -> {result['text_a']}") + if isinstance(result, sqlite3.Row): + log.debug('Result is sqlite3.Row as expected') + + +if __name__ == "__main__": + main() + +# __END__ diff --git a/tests/unit/db_handling/__init__.py b/tests/unit/db_handling/__init__.py new file mode 100644 index 0000000..3f4b12d --- /dev/null +++ b/tests/unit/db_handling/__init__.py @@ -0,0 +1,3 @@ +""" +db_handling tests +""" diff --git a/tests/unit/db_handling/test_sqlite_io.py b/tests/unit/db_handling/test_sqlite_io.py new file mode 100644 index 0000000..81d9b10 --- /dev/null +++ b/tests/unit/db_handling/test_sqlite_io.py @@ -0,0 +1,1133 @@ +""" +PyTest: db_handling/sqlite_io +Tests for SQLiteIO class - Mini SQLite interface + +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 +# pyright: reportIndexIssue=false + +import sqlite3 +from pathlib import Path +from typing import Generator +from unittest.mock import MagicMock, patch +import pytest +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 sqlite_io_instance(mock_logger: MagicMock, temp_db_path: Path) -> Generator[SQLiteIO, None, None]: + """Create a SQLiteIO instance with temporary database""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + autocommit=False, + enable_fkey=True, + row_factory=None + ) + yield instance + # Cleanup + if instance.conn: + instance.db_close() + + +@pytest.fixture +def sqlite_io_with_row_factory(mock_logger: MagicMock, temp_db_path: Path) -> Generator[SQLiteIO, None, None]: + """Create a SQLiteIO instance with Row factory""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + autocommit=False, + enable_fkey=True, + row_factory='Row' + ) + yield instance + if instance.conn: + instance.db_close() + + +@pytest.fixture +def sqlite_io_with_dict_factory(mock_logger: MagicMock, temp_db_path: Path) -> Generator[SQLiteIO, None, None]: + """Create a SQLiteIO instance with Dict factory""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + autocommit=False, + enable_fkey=True, + row_factory='Dict' + ) + yield instance + if instance.conn: + instance.db_close() + + +@pytest.fixture +def populated_db(sqlite_io_instance: SQLiteIO) -> SQLiteIO: + """Create a populated test database with sample data""" + # Create a test table + create_table_query = """ + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + email TEXT UNIQUE, + age INTEGER + ) + """ + sqlite_io_instance.execute_query(create_table_query) + + # Insert test data + test_users = [ + ("Alice", "alice@example.com", 30), + ("Bob", "bob@example.com", 25), + ("Charlie", "charlie@example.com", 35), + ] + + for name, email, age in test_users: + sqlite_io_instance.execute_query( + "INSERT INTO users (name, email, age) VALUES (?, ?, ?)", + (name, email, age) + ) + + return sqlite_io_instance + + +# MARK: Initialization Tests +class TestSQLiteIOInitialization: + """Test SQLiteIO initialization and connection""" + + def test_init_with_default_params(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test initialization with default parameters""" + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert instance.log == mock_logger + assert instance.db_name == temp_db_path + assert instance.autocommit is False + assert instance.enable_fkey is True + assert instance.row_factory is None + assert instance.conn is not None + + instance.db_close() + + def test_init_with_custom_params(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test initialization with custom parameters""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + autocommit=True, + enable_fkey=False, + row_factory='Row' + ) + + assert instance.autocommit is True + assert instance.enable_fkey is False + assert instance.row_factory == 'Row' + assert instance.conn is not None + + instance.db_close() + + def test_init_with_string_path(self, mock_logger: MagicMock, tmp_path: Path) -> None: + """Test initialization with string database path""" + db_path = str(tmp_path / "string_path.db") + instance = SQLiteIO(log=mock_logger, db_name=db_path) + + assert instance.db_name == db_path + assert instance.conn is not None + + instance.db_close() + + def test_init_with_pathlib_path(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test initialization with Path object""" + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert instance.db_name == temp_db_path + assert instance.conn is not None + + instance.db_close() + + def test_init_creates_database_file(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test that initialization creates the database file""" + assert not temp_db_path.exists() + + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert temp_db_path.exists() + + instance.db_close() + + def test_init_with_row_factory(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test initialization with Row factory""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + row_factory='Row' + ) + + assert instance.conn is not None + assert instance.conn.row_factory == sqlite3.Row + + instance.db_close() + + def test_init_with_dict_factory(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test initialization with Dict factory""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + row_factory='Dict' + ) + + assert instance.conn is not None + assert instance.conn.row_factory is not None + # Test that it returns dict-like results + instance.execute_query("CREATE TABLE test (id INTEGER, name TEXT)") + instance.execute_query("INSERT INTO test VALUES (1, 'test')") + result = instance.return_one("SELECT * FROM test") + assert isinstance(result, dict) + + instance.db_close() + + +# MARK: Connection Tests +class TestSQLiteIOConnection: + """Test connection management""" + + def test_db_connect_success(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test successful database connection""" + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert instance.conn is not None + assert isinstance(instance.conn, sqlite3.Connection) + + instance.db_close() + + def test_db_connect_with_invalid_path(self, mock_logger: MagicMock) -> None: + """Test connection with invalid path (should still work as SQLite creates)""" + # SQLite will create the database even with unusual paths + instance = SQLiteIO(log=mock_logger, db_name="/tmp/test_invalid_path.db") + + assert instance.conn is not None + + instance.db_close() + + @patch('sqlite3.connect') + def test_db_connect_error_handling( + self, mock_connect: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ) -> None: + """Test error handling during connection""" + # Simulate connection error + mock_connect.side_effect = sqlite3.Error("Connection failed") + + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert instance.conn is None + mock_logger.error.assert_called() + + @patch('sqlite3.connect') + def test_db_connect_operational_error( + self, mock_connect: MagicMock, mock_logger: MagicMock, temp_db_path: Path + ) -> None: + """Test operational error handling during connection""" + # Simulate operational error + error = sqlite3.OperationalError("Operational error") + mock_connect.side_effect = error + + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + assert instance.conn is None + mock_logger.error.assert_called() + + def test_db_close(self, sqlite_io_instance: SQLiteIO) -> None: + """Test closing database connection""" + assert sqlite_io_instance.conn is not None + + sqlite_io_instance.db_close() + + assert sqlite_io_instance.conn is None + + def test_db_close_already_closed(self, sqlite_io_instance: SQLiteIO) -> None: + """Test closing already closed connection""" + sqlite_io_instance.db_close() + assert sqlite_io_instance.conn is None + + # Should not raise error + sqlite_io_instance.db_close() + assert sqlite_io_instance.conn is None + + def test_db_connected_true(self, sqlite_io_instance: SQLiteIO) -> None: + """Test db_connected returns True when connected""" + assert sqlite_io_instance.db_connected() is True + + def test_db_connected_false(self, sqlite_io_instance: SQLiteIO) -> None: + """Test db_connected returns False when not connected""" + sqlite_io_instance.db_close() + + assert sqlite_io_instance.db_connected() is False + + +# MARK: Table/Trigger/Index Existence Tests +class TestContentExistence: + """Test checking existence of database objects""" + + def test_table_exists_true(self, sqlite_io_instance: SQLiteIO) -> None: + """Test table_exists returns True for existing table""" + sqlite_io_instance.execute_query( + "CREATE TABLE test_table (id INTEGER PRIMARY KEY)" + ) + + assert sqlite_io_instance.table_exists("test_table") is True + + def test_table_exists_false(self, sqlite_io_instance: SQLiteIO) -> None: + """Test table_exists returns False for non-existing table""" + assert sqlite_io_instance.table_exists("non_existing_table") is False + + def test_table_exists_no_connection(self, sqlite_io_instance: SQLiteIO) -> None: + """Test table_exists returns False when not connected""" + sqlite_io_instance.db_close() + + assert sqlite_io_instance.table_exists("test_table") is False + + def test_trigger_exists_true(self, sqlite_io_instance: SQLiteIO) -> None: + """Test trigger_exists returns True for existing trigger""" + # Create table first + sqlite_io_instance.execute_query( + "CREATE TABLE test_table (id INTEGER, updated_at TEXT)" + ) + + # Create trigger + sqlite_io_instance.execute_query(""" + CREATE TRIGGER test_trigger + AFTER UPDATE ON test_table + BEGIN + UPDATE test_table SET updated_at = datetime('now') WHERE id = NEW.id; + END + """) + + assert sqlite_io_instance.trigger_exists("test_trigger") is True + + def test_trigger_exists_false(self, sqlite_io_instance: SQLiteIO) -> None: + """Test trigger_exists returns False for non-existing trigger""" + assert sqlite_io_instance.trigger_exists("non_existing_trigger") is False + + def test_index_exists_true(self, sqlite_io_instance: SQLiteIO) -> None: + """Test index_exists returns True for existing index""" + # Create table first + sqlite_io_instance.execute_query( + "CREATE TABLE test_table (id INTEGER, name TEXT)" + ) + + # Create index + sqlite_io_instance.execute_query( + "CREATE INDEX test_index ON test_table (name)" + ) + + assert sqlite_io_instance.index_exists("test_index") is True + + def test_index_exists_false(self, sqlite_io_instance: SQLiteIO) -> None: + """Test index_exists returns False for non-existing index""" + assert sqlite_io_instance.index_exists("non_existing_index") is False + + def test_content_exists_error_handling(self, sqlite_io_instance: SQLiteIO) -> None: + """Test error handling in content existence check""" + # Close connection to trigger error + sqlite_io_instance.db_close() + + result = sqlite_io_instance.table_exists("test") + + assert result is False + + +# MARK: Query Execution Tests +class TestQueryExecution: + """Test query execution methods""" + + def test_execute_query_create_table(self, sqlite_io_instance: SQLiteIO) -> None: + """Test executing CREATE TABLE query""" + result = sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)" + ) + + assert result == [] + assert sqlite_io_instance.table_exists("test") is True + + def test_execute_query_insert(self, sqlite_io_instance: SQLiteIO) -> None: + """Test executing INSERT query""" + sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + + result = sqlite_io_instance.execute_query( + "INSERT INTO test (id, name) VALUES (?, ?)", + (1, "Alice") + ) + + assert result == [] + + def test_execute_query_select(self, populated_db: SQLiteIO) -> None: + """Test executing SELECT query""" + result = populated_db.execute_query("SELECT * FROM users ORDER BY id") + + assert isinstance(result, list) + assert len(result) == 3 + assert result[0][1] == "Alice" + + def test_execute_query_select_with_params(self, populated_db: SQLiteIO) -> None: + """Test SELECT query with parameters""" + result = populated_db.execute_query( + "SELECT * FROM users WHERE name = ?", + ("Bob",) + ) + + assert len(result) == 1 + assert result[0][1] == "Bob" + + def test_execute_query_update(self, populated_db: SQLiteIO) -> None: + """Test executing UPDATE query""" + populated_db.execute_query( + "UPDATE users SET age = ? WHERE name = ?", + (26, "Bob") + ) + + # Verify update + check = populated_db.execute_query( + "SELECT age FROM users WHERE name = ?", + ("Bob",) + ) + assert check[0][0] == 26 + + def test_execute_query_delete(self, populated_db: SQLiteIO) -> None: + """Test executing DELETE query""" + populated_db.execute_query( + "DELETE FROM users WHERE name = ?", + ("Charlie",) + ) + + # Verify deletion + check = populated_db.execute_query("SELECT COUNT(*) FROM users") + assert check[0][0] == 2 + + def test_execute_query_no_connection(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test execute_query with no connection""" + sqlite_io_instance.db_close() + + result = sqlite_io_instance.execute_query("SELECT 1") + + assert result is False + mock_logger.warning.assert_called() + + def test_execute_query_syntax_error(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test execute_query with syntax error""" + result = sqlite_io_instance.execute_query("INVALID SQL SYNTAX") + + assert result is False + mock_logger.error.assert_called() + + def test_execute_query_with_none_params(self, sqlite_io_instance: SQLiteIO) -> None: + """Test execute_query with None params""" + sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER)" + ) + result = sqlite_io_instance.execute_query( + "INSERT INTO test VALUES (1)", + None + ) + + assert result == [] + + +# MARK: Cursor Execution Tests +class TestCursorExecution: + """Test cursor-based operations""" + + def test_execute_cursor_success(self, sqlite_io_instance: SQLiteIO) -> None: + """Test successful cursor execution""" + sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + sqlite_io_instance.execute_query( + "INSERT INTO test VALUES (1, 'Alice')" + ) + + cursor = sqlite_io_instance.execute_cursor("SELECT * FROM test") + + assert cursor is not False + assert isinstance(cursor, sqlite3.Cursor) + + def test_execute_cursor_with_params(self, sqlite_io_instance: SQLiteIO) -> None: + """Test cursor execution with parameters""" + sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + + cursor = sqlite_io_instance.execute_cursor( + "SELECT * FROM test WHERE id = ?", + (1,) + ) + + assert cursor is not False + + def test_execute_cursor_no_connection(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test cursor execution with no connection""" + sqlite_io_instance.db_close() + + cursor = sqlite_io_instance.execute_cursor("SELECT 1") + + assert cursor is False + mock_logger.warning.assert_called() + + def test_execute_cursor_error(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test cursor execution error handling""" + cursor = sqlite_io_instance.execute_cursor("INVALID SQL") + + assert cursor is False + mock_logger.error.assert_called() + + +# MARK: Return One Tests +class TestReturnOne: + """Test return_one method""" + + def test_return_one_success(self, populated_db: SQLiteIO) -> None: + """Test return_one with successful query""" + result = populated_db.return_one( + "SELECT * FROM users WHERE name = ?", + ("Alice",) + ) + + assert result is not None + assert result is not False + assert result[1] == "Alice" + + def test_return_one_no_result(self, populated_db: SQLiteIO) -> None: + """Test return_one when no results found""" + result = populated_db.return_one( + "SELECT * FROM users WHERE name = ?", + ("NonExistent",) + ) + + assert result is None + + def test_return_one_no_params(self, populated_db: SQLiteIO) -> None: + """Test return_one without parameters""" + result = populated_db.return_one("SELECT * FROM users LIMIT 1") + + assert result is not None + assert result is not False + + def test_return_one_no_connection(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test return_one with no connection""" + sqlite_io_instance.db_close() + + result = sqlite_io_instance.return_one("SELECT 1") + + assert result is False + mock_logger.warning.assert_called() + + def test_return_one_error(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test return_one error handling""" + result = sqlite_io_instance.return_one("INVALID SQL") + + assert result is False + mock_logger.error.assert_called() + + def test_return_one_with_row_factory(self, sqlite_io_with_row_factory: SQLiteIO) -> None: + """Test return_one with Row factory""" + sqlite_io_with_row_factory.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + sqlite_io_with_row_factory.execute_query( + "INSERT INTO test VALUES (1, 'Alice')" + ) + + result = sqlite_io_with_row_factory.return_one("SELECT * FROM test") + + assert result is not None + assert isinstance(result, sqlite3.Row) + assert not isinstance(result, dict) + assert result['name'] == 'Alice' + + def test_return_one_with_dict_factory(self, sqlite_io_with_dict_factory: SQLiteIO) -> None: + """Test return_one with Dict factory""" + sqlite_io_with_dict_factory.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + sqlite_io_with_dict_factory.execute_query( + "INSERT INTO test VALUES (1, 'Alice')" + ) + + result = sqlite_io_with_dict_factory.return_one("SELECT * FROM test") + + assert result is not None + assert isinstance(result, dict) + assert result['name'] == 'Alice' + + +# MARK: Fetch Row Tests +class TestFetchRow: + """Test fetch_row method""" + + def test_fetch_row_success(self, populated_db: SQLiteIO) -> None: + """Test fetch_row with valid cursor""" + cursor = populated_db.execute_cursor("SELECT * FROM users ORDER BY id") + + result = populated_db.fetch_row(cursor) + + assert result is not None + assert result is not False + assert result[1] == "Alice" + + def test_fetch_row_multiple_calls(self, populated_db: SQLiteIO) -> None: + """Test multiple fetch_row calls""" + cursor = populated_db.execute_cursor("SELECT * FROM users ORDER BY id") + + result1 = populated_db.fetch_row(cursor) + result2 = populated_db.fetch_row(cursor) + result3 = populated_db.fetch_row(cursor) + + assert result1 is not False + assert result2 is not False + assert result3 is not False + assert result1 is not None + assert result2 is not None + assert result3 is not None + assert result1[1] == "Alice" + assert result2[1] == "Bob" + assert result3[1] == "Charlie" + + def test_fetch_row_exhausted_cursor(self, populated_db: SQLiteIO) -> None: + """Test fetch_row when cursor is exhausted""" + cursor = populated_db.execute_cursor("SELECT * FROM users LIMIT 1") + + populated_db.fetch_row(cursor) # First row + result = populated_db.fetch_row(cursor) # No more rows + + assert result is None + + def test_fetch_row_no_connection(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test fetch_row with no connection""" + sqlite_io_instance.db_close() + + result = sqlite_io_instance.fetch_row(False) + + assert result is False + mock_logger.warning.assert_called() + + def test_fetch_row_false_cursor(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test fetch_row with False cursor""" + result = sqlite_io_instance.fetch_row(False) + + assert result is False + mock_logger.warning.assert_called() + + def test_fetch_row_error(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test fetch_row error handling""" + # Create a closed cursor to trigger error + cursor = sqlite_io_instance.execute_cursor("SELECT 1") + if cursor is not False: + cursor.close() + + result = sqlite_io_instance.fetch_row(cursor) + + assert result is False + mock_logger.error.assert_called() + + +# MARK: Metadata Tests +class TestMetadata: + """Test metadata retrieval""" + + def test_meta_data_detail_simple_table(self, sqlite_io_instance: SQLiteIO) -> None: + """Test meta_data_detail for simple table""" + sqlite_io_instance.execute_query(""" + CREATE TABLE test_table ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + email TEXT UNIQUE, + age INTEGER DEFAULT 0 + ) + """) + + result = sqlite_io_instance.meta_data_detail("test_table") + + assert result is not False + assert isinstance(result, list) + assert len(result) > 0 + + def test_meta_data_detail_with_index(self, sqlite_io_instance: SQLiteIO) -> None: + """Test meta_data_detail for table with index""" + sqlite_io_instance.execute_query(""" + CREATE TABLE test_table ( + id INTEGER PRIMARY KEY, + name TEXT + ) + """) + sqlite_io_instance.execute_query( + "CREATE INDEX idx_name ON test_table (name)" + ) + + result = sqlite_io_instance.meta_data_detail("test_table") + + assert result is not False + assert isinstance(result, list) + + def test_meta_data_detail_nonexistent_table(self, sqlite_io_instance: SQLiteIO) -> None: + """Test meta_data_detail for non-existent table""" + result = sqlite_io_instance.meta_data_detail("nonexistent_table") + + assert result == [] or result is not False + + +# MARK: Row Factory Tests +class TestRowFactory: + """Test different row factory configurations""" + + def test_row_factory_none(self, mock_logger: MagicMock, temp_db_path) -> None: + """Test with no row factory (default tuple results)""" + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path, row_factory=None) + instance.execute_query("CREATE TABLE test (id INTEGER, name TEXT)") + instance.execute_query("INSERT INTO test VALUES (1, 'Alice')") + + result = instance.return_one("SELECT * FROM test") + + assert isinstance(result, tuple) + assert result[0] == 1 + assert result[1] == 'Alice' + + instance.db_close() + + def test_row_factory_row(self, sqlite_io_with_row_factory: SQLiteIO) -> None: + """Test with Row factory""" + sqlite_io_with_row_factory.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + sqlite_io_with_row_factory.execute_query( + "INSERT INTO test VALUES (1, 'Alice')" + ) + + result = sqlite_io_with_row_factory.execute_query("SELECT * FROM test") + + assert len(result) == 1 + assert isinstance(result[0], sqlite3.Row) + assert result[0]['id'] == 1 + assert result[0]['name'] == 'Alice' + + def test_row_factory_dict(self, sqlite_io_with_dict_factory: SQLiteIO) -> None: + """Test with Dict factory""" + sqlite_io_with_dict_factory.execute_query( + "CREATE TABLE test (id INTEGER, name TEXT)" + ) + sqlite_io_with_dict_factory.execute_query( + "INSERT INTO test VALUES (1, 'Alice')" + ) + + result = sqlite_io_with_dict_factory.execute_query("SELECT * FROM test") + + assert len(result) == 1 + assert isinstance(result[0], dict) + assert result[0]['id'] == 1 + assert result[0]['name'] == 'Alice' + + def test_row_factory_invalid(self, mock_logger: MagicMock, temp_db_path: Path) -> None: + """Test with invalid row factory string""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + row_factory='InvalidFactory' + ) + + # Should default to None + instance.execute_query("CREATE TABLE test (id INTEGER)") + instance.execute_query("INSERT INTO test VALUES (1)") + result = instance.return_one("SELECT * FROM test") + + assert isinstance(result, tuple) + + instance.db_close() + + +# MARK: Integration Tests +class TestIntegration: + """Integration tests combining multiple operations""" + + def test_full_crud_cycle(self, sqlite_io_instance: SQLiteIO) -> None: + """Test complete CRUD cycle""" + # Create + sqlite_io_instance.execute_query(""" + CREATE TABLE products ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + price REAL, + stock INTEGER DEFAULT 0 + ) + """) + + # Insert + sqlite_io_instance.execute_query( + "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)", + ("Widget", 19.99, 100) + ) + + # Read + result = sqlite_io_instance.return_one( + "SELECT * FROM products WHERE name = ?", + ("Widget",) + ) + assert result is not None + assert result[1] == "Widget" + + # Update + sqlite_io_instance.execute_query( + "UPDATE products SET price = ? WHERE name = ?", + (24.99, "Widget") + ) + + updated = sqlite_io_instance.return_one( + "SELECT price FROM products WHERE name = ?", + ("Widget",) + ) + assert updated is not None + assert updated is not False + assert updated[0] == 24.99 + + # Delete + sqlite_io_instance.execute_query( + "DELETE FROM products WHERE name = ?", + ("Widget",) + ) + + deleted = sqlite_io_instance.return_one( + "SELECT * FROM products WHERE name = ?", + ("Widget",) + ) + assert deleted is None + + def test_transaction_workflow(self, sqlite_io_instance: SQLiteIO) -> None: + """Test transaction workflow""" + sqlite_io_instance.execute_query( + "CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance REAL)" + ) + sqlite_io_instance.execute_query( + "INSERT INTO accounts VALUES (1, 1000.0), (2, 500.0)" + ) + + # Transfer money + sqlite_io_instance.execute_query( + "UPDATE accounts SET balance = balance - 100 WHERE id = 1" + ) + sqlite_io_instance.execute_query( + "UPDATE accounts SET balance = balance + 100 WHERE id = 2" + ) + + # Verify + account1 = sqlite_io_instance.return_one( + "SELECT balance FROM accounts WHERE id = 1" + ) + account2 = sqlite_io_instance.return_one( + "SELECT balance FROM accounts WHERE id = 2" + ) + + assert account1 is not False + assert account2 is not False + assert account1 is not None + assert account2 is not None + assert account1[0] == 900.0 + assert account2[0] == 600.0 + + def test_cursor_iteration(self, populated_db: SQLiteIO) -> None: + """Test iterating through cursor results""" + cursor = populated_db.execute_cursor( + "SELECT name FROM users ORDER BY name" + ) + + names: list[str] = [] + while True: + row = populated_db.fetch_row(cursor) + if row is None: + break + names.append(row[0]) + + assert names == ["Alice", "Bob", "Charlie"] + + def test_foreign_key_enforcement(self, sqlite_io_instance: SQLiteIO) -> None: + """Test foreign key constraints are enabled""" + sqlite_io_instance.execute_query(""" + CREATE TABLE parent ( + id INTEGER PRIMARY KEY + ) + """) + + sqlite_io_instance.execute_query(""" + CREATE TABLE child ( + id INTEGER PRIMARY KEY, + parent_id INTEGER, + FOREIGN KEY (parent_id) REFERENCES parent(id) + ) + """) + + # Insert valid parent + sqlite_io_instance.execute_query("INSERT INTO parent VALUES (1)") + + # This should work + result1 = sqlite_io_instance.execute_query( + "INSERT INTO child VALUES (1, 1)" + ) + assert result1 == [] + + +# MARK: Parametrized Tests +class TestParametrized: + """Parametrized tests for comprehensive coverage""" + + @pytest.mark.parametrize("row_factory,expected_type", [ + (None, tuple), + ('Row', sqlite3.Row), + ('Dict', dict), + ]) + def test_row_factory_types( + self, + mock_logger: MagicMock, temp_db_path: Path, + row_factory: str | None, + expected_type + ) -> None: + """Test different row factory configurations""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + row_factory=row_factory + ) + + instance.execute_query("CREATE TABLE test (id INTEGER, name TEXT)") + instance.execute_query("INSERT INTO test VALUES (1, 'Test')") + + result = instance.return_one("SELECT * FROM test") + + assert isinstance(result, expected_type) + + instance.db_close() + + @pytest.mark.parametrize("autocommit", [True, False]) + def test_autocommit_modes(self, mock_logger: MagicMock, temp_db_path: Path, autocommit: bool) -> None: + """Test different autocommit modes""" + instance = SQLiteIO( + log=mock_logger, + db_name=temp_db_path, + autocommit=autocommit + ) + + assert instance.autocommit == autocommit + assert instance.conn is not None + + instance.db_close() + + @pytest.mark.parametrize("table_name,should_exist", [ + ("existing_table", True), + ("nonexistent_table", False), + ]) + def test_table_existence(self, sqlite_io_instance: SQLiteIO, table_name: str, should_exist: bool) -> None: + """Test table existence checks""" + if should_exist: + sqlite_io_instance.execute_query( + f"CREATE TABLE {table_name} (id INTEGER)" + ) + + result = sqlite_io_instance.table_exists(table_name) + + assert result == should_exist + + +# MARK: Edge Cases +class TestEdgeCases: + """Test edge cases and boundary conditions""" + + def test_empty_database(self, sqlite_io_instance: SQLiteIO) -> None: + """Test operations on empty database""" + result = sqlite_io_instance.execute_query( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + + assert isinstance(result, list) + + def test_large_insert(self, sqlite_io_instance: SQLiteIO) -> None: + """Test inserting many rows""" + sqlite_io_instance.execute_query( + "CREATE TABLE large_table (id INTEGER, value TEXT)" + ) + + # Insert 1000 rows + for i in range(1000): + sqlite_io_instance.execute_query( + "INSERT INTO large_table VALUES (?, ?)", + (i, f"value_{i}") + ) + + count = sqlite_io_instance.return_one( + "SELECT COUNT(*) FROM large_table" + ) + + assert count is not False + assert count is not None + assert count[0] == 1000 + + def test_unicode_data(self, sqlite_io_instance: SQLiteIO) -> None: + """Test handling Unicode data""" + sqlite_io_instance.execute_query( + "CREATE TABLE unicode_test (text TEXT)" + ) + + unicode_strings = ["Hello 世界", "Привет", "مرحبا", "🚀🌟"] + + for text in unicode_strings: + sqlite_io_instance.execute_query( + "INSERT INTO unicode_test VALUES (?)", + (text,) + ) + + results = sqlite_io_instance.execute_query( + "SELECT * FROM unicode_test" + ) + + assert len(results) == len(unicode_strings) + + def test_null_values(self, sqlite_io_instance: SQLiteIO) -> None: + """Test handling NULL values""" + sqlite_io_instance.execute_query( + "CREATE TABLE null_test (id INTEGER, value TEXT)" + ) + + sqlite_io_instance.execute_query( + "INSERT INTO null_test VALUES (1, NULL)" + ) + + result = sqlite_io_instance.return_one( + "SELECT value FROM null_test WHERE id = 1" + ) + + assert result is not False + assert result is not None + assert result[0] is None + + def test_empty_string_values(self, sqlite_io_instance: SQLiteIO) -> None: + """Test handling empty string values""" + sqlite_io_instance.execute_query( + "CREATE TABLE empty_test (value TEXT)" + ) + + sqlite_io_instance.execute_query( + "INSERT INTO empty_test VALUES (?)", + ("",) + ) + + result = sqlite_io_instance.return_one("SELECT value FROM empty_test") + + assert result is not False + assert result is not None + assert result[0] == "" + + def test_very_long_string(self, sqlite_io_instance: SQLiteIO) -> None: + """Test handling very long strings""" + sqlite_io_instance.execute_query( + "CREATE TABLE long_test (value TEXT)" + ) + + long_string = "x" * 10000 + + sqlite_io_instance.execute_query( + "INSERT INTO long_test VALUES (?)", + (long_string,) + ) + + result = sqlite_io_instance.return_one("SELECT value FROM long_test") + + assert result is not False + assert result is not None + assert len(result[0]) == 10000 + + def test_special_characters_in_table_name(self, sqlite_io_instance: SQLiteIO) -> None: + """Test table names with special characters""" + # SQLite allows various characters in quoted identifiers + sqlite_io_instance.execute_query( + 'CREATE TABLE "test-table" (id INTEGER)' + ) + + assert sqlite_io_instance.table_exists("test-table") is True + + def test_reconnection_after_close(self, mock_logger: MagicMock, temp_db_path) -> None: + """Test reconnecting after closing""" + instance = SQLiteIO(log=mock_logger, db_name=temp_db_path) + + instance.db_close() + assert instance.conn is None + + # Reconnect + instance.conn = instance.db_connect() + assert instance.conn is not None + + instance.db_close() + + +# MARK: Error Scenarios +class TestErrorScenarios: + """Test various error scenarios""" + + def test_syntax_error_in_query(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test handling of SQL syntax errors""" + result = sqlite_io_instance.execute_query("SELECT * FORM invalid_syntax") + + assert result is False + mock_logger.error.assert_called() + + def test_constraint_violation(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test handling of constraint violations""" + sqlite_io_instance.execute_query( + "CREATE TABLE test (id INTEGER PRIMARY KEY)" + ) + sqlite_io_instance.execute_query("INSERT INTO test VALUES (1)") + + # Try to insert duplicate primary key + result = sqlite_io_instance.execute_query("INSERT INTO test VALUES (1)") + + assert result is False + mock_logger.error.assert_called() + + def test_invalid_column_name(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test querying invalid column""" + sqlite_io_instance.execute_query("CREATE TABLE test (id INTEGER)") + + result = sqlite_io_instance.execute_query( + "SELECT invalid_column FROM test" + ) + + assert result is False + mock_logger.error.assert_called() + + def test_missing_table(self, sqlite_io_instance: SQLiteIO, mock_logger: MagicMock) -> None: + """Test querying non-existent table""" + result = sqlite_io_instance.execute_query( + "SELECT * FROM nonexistent_table" + ) + + assert result is False + mock_logger.error.assert_called() + + +# __END__