From 17ab49b38fd12230774d83d0363f2eb00328c3af Mon Sep 17 00:00:00 2001 From: Nancy Wei Date: Thu, 25 Jun 2026 16:33:08 +1000 Subject: [PATCH 1/2] add test --- backend/tests/test_connection.py | 65 +++++ backend/tests/test_folders.py | 417 +++++++++++++++++++++++++++++-- 2 files changed, 466 insertions(+), 16 deletions(-) create mode 100644 backend/tests/test_connection.py diff --git a/backend/tests/test_connection.py b/backend/tests/test_connection.py new file mode 100644 index 000000000..cc912e07f --- /dev/null +++ b/backend/tests/test_connection.py @@ -0,0 +1,65 @@ +import sqlite3 +import pytest +from unittest.mock import patch + +from app.database.connection import get_db_connection + + +def test_get_db_connection_commits_on_success(tmp_path): + db_path = tmp_path / "test.db" + + with patch("app.database.connection.DATABASE_PATH", str(db_path)): + with get_db_connection() as conn: + conn.execute("DROP TABLE IF EXISTS test") + conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)") + conn.execute("INSERT INTO test (name) VALUES (?)", ("Alice",)) + + # reopen DB to check commit happened + conn = sqlite3.connect(db_path) + result = conn.execute("SELECT name FROM test").fetchone() + conn.close() + + assert result[0] == "Alice" + + +def test_get_db_connection_rolls_back_on_exception(tmp_path): + db_path = tmp_path / "test.db" + + with patch("app.database.connection.DATABASE_PATH", str(db_path)): + # Create table first and commit successfully + with get_db_connection() as conn: + conn.execute( + "CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)" + ) + + # Insert data, then raise error + with pytest.raises(ValueError): + with get_db_connection() as conn: + conn.execute( + "INSERT INTO test (name) VALUES (?)", + ("Bob",), + ) + raise ValueError("fail") + + # Check Bob was rolled back + conn = sqlite3.connect(db_path) + result = conn.execute("SELECT name FROM test").fetchone() + conn.close() + + assert result is None + # CREATE TABLE was rolled back too + assert result is None + +def test_get_db_connection_enables_pragmas(tmp_path): + db_path = tmp_path / "test.db" + + with patch("app.database.connection.DATABASE_PATH", str(db_path)): + with get_db_connection() as conn: + foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0] + ignore_check = conn.execute("PRAGMA ignore_check_constraints").fetchone()[0] + recursive_triggers = conn.execute("PRAGMA recursive_triggers").fetchone()[0] + case_sensitive_like = conn.execute("PRAGMA case_sensitive_like").fetchone() + + assert foreign_keys == 1 + assert ignore_check == 0 + assert recursive_triggers == 1 \ No newline at end of file diff --git a/backend/tests/test_folders.py b/backend/tests/test_folders.py index 31f6b1cd4..addcc287e 100644 --- a/backend/tests/test_folders.py +++ b/backend/tests/test_folders.py @@ -1,3 +1,4 @@ +import sqlite3 import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -10,26 +11,26 @@ from app.routes.folders import router as folders_router +from app.database.folders import db_disable_ai_tagging_batch, db_enable_ai_tagging_batch, db_update_ai_tagging_batch, \ + db_get_folder_ids_by_path_prefix, db_get_folder_ids_by_paths, db_get_all_folder_details, \ + db_get_direct_child_folders, db_get_folder_path_from_id, db_insert_folders_batch, db_insert_folder, \ + db_get_folder_id_from_path, db_delete_folder, db_update_parent_ids_for_subtree, db_folder_exists, \ + db_delete_folders_batch, db_get_all_folder_ids, db_get_all_folders, db_find_parent_folder_id + + # ############################## # Pytest Fixtures # ############################## -@pytest.fixture(scope="function") -def test_db(): - """Create a temporary test database for each test.""" - db_fd, db_path = tempfile.mkstemp() - - import app.config.settings - - original_db_path = app.config.settings.DATABASE_PATH - app.config.settings.DATABASE_PATH = db_path - - yield db_path - - app.config.settings.DATABASE_PATH = original_db_path - os.close(db_fd) - os.unlink(db_path) +@pytest.fixture +def test_db(tmp_path, monkeypatch): + db_path = tmp_path / "test.db" + monkeypatch.setattr( + "app.database.folders.DATABASE_PATH", + str(db_path) + ) + return str(db_path) @pytest.fixture @@ -688,10 +689,394 @@ def test_disable_ai_tagging_no_folders_updated(self, mock_disable_batch, client) assert data["success"] is True assert data["data"]["updated_count"] == 0 + # ============================================================================ - # Integration & Workflow Tests + # Unit Tests # ============================================================================ + +def create_folders_table_for_test(db_path): + conn = sqlite3.connect(db_path) + cursor = conn.cursor() + conn.execute("DROP TABLE IF EXISTS folders") + cursor.execute(""" + CREATE TABLE IF NOT EXISTS folders + ( + folder_id + TEXT + PRIMARY + KEY, + folder_path + TEXT + UNIQUE, + parent_folder_id + TEXT, + last_modified_time + INTEGER, + AI_Tagging + BOOLEAN, + taggingCompleted + BOOLEAN + ) + """) + conn.commit() + conn.close() + + +class TestFoldersUnit: + + def test_db_insert_folders_batch(self, test_db): + create_folders_table_for_test(test_db) + + db_insert_folders_batch([ + ("folder-id-1", "/tmp/photos", None, 1693526400, True, False), + ("folder-id-2", "/tmp/docs", None, 1693526500, False, True), + ]) + + conn = sqlite3.connect(test_db) + rows = conn.execute("SELECT folder_id FROM folders ORDER BY folder_id").fetchall() + conn.close() + + assert rows == [("folder-id-1",), ("folder-id-2",)] + + def test_db_insert_folder_success(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "photos" + folder.mkdir() + + result = db_insert_folder(str(folder), folder_id="folder-id-1") + + assert result == "folder-id-1" + assert db_folder_exists(str(folder)) is True + + def test_db_insert_folder_existing_returns_existing_id(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "photos" + folder.mkdir() + + first_id = db_insert_folder(str(folder), folder_id="folder-id-1") + second_id = db_insert_folder(str(folder), folder_id="folder-id-2") + + assert first_id == "folder-id-1" + assert second_id == "folder-id-1" + + def test_db_insert_folder_invalid_directory(self, test_db): + create_folders_table_for_test(test_db) + + with pytest.raises(ValueError): + db_insert_folder("/path/does/not/exist") + + def test_db_get_folder_id_from_path(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "docs" + folder.mkdir() + + db_insert_folder(str(folder), folder_id="folder-id-1") + + result = db_get_folder_id_from_path(str(folder)) + + assert result == "folder-id-1" + + def test_db_get_folder_path_from_id(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "docs" + folder.mkdir() + + db_insert_folder(str(folder), folder_id="folder-id-1") + + result = db_get_folder_path_from_id("folder-id-1") + + assert result == os.path.abspath(str(folder)) + + def test_db_get_all_folders(self, test_db): + create_folders_table_for_test(test_db) + + db_insert_folders_batch([ + ("folder-id-1", "/tmp/photos", None, 1693526400, True, False), + ("folder-id-2", "/tmp/docs", None, 1693526500, False, True), + ]) + + result = db_get_all_folders() + + assert set(result) == {"/tmp/photos", "/tmp/docs"} + + def test_db_get_all_folder_ids(self, test_db): + create_folders_table_for_test(test_db) + + db_insert_folders_batch([ + ("folder-id-1", "/tmp/photos", None, 1693526400, True, False), + ("folder-id-2", "/tmp/docs", None, 1693526500, False, True), + ]) + + result = db_get_all_folder_ids() + + assert set(result) == {"folder-id-1", "folder-id-2"} + + def test_db_delete_folders_batch_empty_list(self, test_db): + result = db_delete_folders_batch([]) + + assert result == 0 + + def test_db_delete_folders_batch_success(self, test_db): + create_folders_table_for_test(test_db) + + db_insert_folders_batch([ + ("folder-id-1", "/tmp/photos", None, 1693526400, True, False), + ("folder-id-2", "/tmp/docs", None, 1693526500, False, True), + ]) + + result = db_delete_folders_batch(["folder-id-1"]) + + assert result == 1 + assert db_get_folder_path_from_id("folder-id-1") is None + assert db_get_folder_path_from_id("folder-id-2") == "/tmp/docs" + + def test_db_delete_folder_success(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "photos" + folder.mkdir() + + db_insert_folder(str(folder), folder_id="folder-id-1") + db_delete_folder(str(folder)) + + assert db_folder_exists(str(folder)) is False + + def test_db_delete_folder_not_exists(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "missing" + folder.mkdir() + + with pytest.raises(ValueError): + db_delete_folder(str(folder)) + + def test_db_update_parent_ids_for_subtree(self, test_db): + create_folders_table_for_test(test_db) + + db_insert_folders_batch([ + ("root-id", "/tmp/root", None, 1693526400, True, False), + ("child-id", "/tmp/root/child", None, 1693526500, True, False), + ]) + + db_update_parent_ids_for_subtree( + "/tmp/root", + { + "/tmp/root": ("root-id", None), + "/tmp/root/child": ("child-id", "root-id"), + }, + ) + + conn = sqlite3.connect(test_db) + parent_id = conn.execute( + "SELECT parent_folder_id FROM folders WHERE folder_id = ?", + ("child-id",), + ).fetchone()[0] + conn.close() + + assert parent_id == "root-id" + + def test_db_folder_exists_true_false(self, test_db, tmp_path): + create_folders_table_for_test(test_db) + + folder = tmp_path / "photos" + folder.mkdir() + + db_insert_folder(str(folder), folder_id="folder-id-1") + + assert db_folder_exists(str(folder)) is True + assert db_folder_exists(str(tmp_path / "missing")) is False + + def test_db_find_parent_folder_id_found(self, test_db): + conn = sqlite3.connect(test_db) + cursor = conn.cursor() + create_folders_table_for_test(test_db) + cursor.execute( + "INSERT INTO folders (folder_id, folder_path) VALUES (?, ?)", + ("parent-id", "/tmp/photos") + ) + conn.commit() + conn.close() + result = db_find_parent_folder_id("/tmp/photos/2024") + assert result == "parent-id" + + def test_db_find_parent_folder_id_not_found(self, test_db): + conn = sqlite3.connect(test_db) + cursor = conn.cursor() + create_folders_table_for_test(test_db) + conn.commit() + conn.close() + result = db_find_parent_folder_id("/tmp/photos/2024") + assert result is None + + def test_db_update_ai_tagging_batch(self, test_db): + conn = sqlite3.connect(test_db) + cursor = conn.cursor() + create_folders_table_for_test(test_db) + cursor.execute( + "INSERT INTO folders (folder_id, folder_path, AI_Tagging) VALUES (?, ?, ?)", + ("tmp", "/tmp", False) + ) + conn.commit() + conn.close() + result = db_update_ai_tagging_batch(["tmp"], True) + assert result == 1 + result = db_update_ai_tagging_batch([], True) + assert result == 0 + + @patch("app.database.folders.sqlite3.connect") + def test_db_update_ai_tagging_batch_sqlite_error(self, mock_connect): + mock_conn = MagicMock() + mock_cursor = MagicMock() + + mock_connect.return_value = mock_conn + mock_conn.cursor.return_value = mock_cursor + + mock_cursor.execute.side_effect = sqlite3.Error("fake db error") + + with pytest.raises(sqlite3.Error): + db_update_ai_tagging_batch(["folder-id-1"], False) + + mock_conn.rollback.assert_called_once() + mock_conn.close.assert_called_once() + + @patch("app.database.folders.db_update_ai_tagging_batch") + def test_db_enable_ai_tagging_batch(self, mock_update_batch): + mock_update_batch.return_value = 1 + result = db_enable_ai_tagging_batch(["tmp"]) + assert result == 1 + mock_update_batch.assert_called_once_with(["tmp"], True) + + @patch("app.database.folders.db_update_ai_tagging_batch") + def test_db_disable_ai_tagging_batch(self, mock_update_batch): + mock_update_batch.return_value = 1 + result = db_disable_ai_tagging_batch(["tmp"]) + assert result == 1 + mock_update_batch.assert_called_once_with(["tmp"], False) + + def test_db_get_folder_ids_by_path_prefix(self,test_db): + conn = sqlite3.connect(test_db) + cursor = conn.cursor() + create_folders_table_for_test(test_db) + cursor.executemany( + "INSERT INTO folders (folder_id, folder_path) VALUES (?, ?)", + [ + ("folder-id-1", "/tmp/photos"), + ("folder-id-2", "/tmp/photos/2024"), + ("folder-id-3", "/other/documents"), + ], + ) + conn.commit() + conn.close() + result = db_get_folder_ids_by_path_prefix("/tmp") + assert result == [ + ("folder-id-1", "/tmp/photos"), + ("folder-id-2", "/tmp/photos/2024"), + ] + + def test_db_get_folder_ids_by_paths(self,test_db): + conn = sqlite3.connect(test_db) + create_folders_table_for_test(test_db) + folder1 = os.path.abspath("test_folder_1") + folder2 = os.path.abspath("test_folder_2") + conn.execute( + "INSERT INTO folders (folder_id, folder_path) VALUES (?, ?)", + ("id_1", folder1), + ) + conn.execute( + "INSERT INTO folders (folder_id, folder_path) VALUES (?, ?)", + ("id_2", folder2), + ) + conn.commit() + conn.close() + result = db_get_folder_ids_by_paths([ + "test_folder_1", + "test_folder_2", + ]) + assert result == { + folder1: "id_1", + folder2: "id_2", + } + + def test_db_get_folder_ids_by_paths_empty(self): + result = db_get_folder_ids_by_paths([]) + assert result == {} + + def test_db_get_all_folder_details(self, test_db): + conn = sqlite3.connect(test_db) + cursor = conn.cursor() + create_folders_table_for_test(test_db) + cursor.execute(""" + CREATE TABLE images + ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + folder_id TEXT + ) + """) + cursor.execute(""" + INSERT INTO folders VALUES (?, ?, ?, ?, ?, ?) + """, ( + "folder-id-1","/home/user/photos",None,1693526400,True,False, + )) + cursor.execute(""" + INSERT INTO images (folder_id) + VALUES (?) + """, ("folder-id-1",)) + + cursor.execute(""" + INSERT INTO images (folder_id) + VALUES (?) + """, ("folder-id-1",)) + conn.commit() + conn.close() + result = db_get_all_folder_details() + assert result == [ + ( + "folder-id-1", + "/home/user/photos", + None, + 1693526400, + 1, + 0, + 2, + ) + ] + + def test_db_get_direct_child_folders(self,test_db): + conn = sqlite3.connect(test_db) + create_folders_table_for_test(test_db) + conn.execute( + "INSERT INTO folders (folder_id, folder_path, parent_folder_id) VALUES (?, ?, ?)", + ("root", "/root", None), + ) + conn.execute( + "INSERT INTO folders (folder_id, folder_path, parent_folder_id) VALUES (?, ?, ?)", + ("child_1", "/root/child1", "root"), + ) + conn.execute( + "INSERT INTO folders (folder_id, folder_path, parent_folder_id) VALUES (?, ?, ?)", + ("child_2", "/root/child2", "root"), + ) + conn.execute( + "INSERT INTO folders (folder_id, folder_path, parent_folder_id) VALUES (?, ?, ?)", + ("grandchild", "/root/child1/grandchild", "child_1"), + ) + conn.commit() + conn.close() + result = db_get_direct_child_folders("root") + assert set(result) == { + ("child_1", "/root/child1"), + ("child_2", "/root/child2"), + } + + # ============================================================================ + # Integration & Workflow Tests + # ============================================================================ +class TestFoldersIntegration: @patch("app.routes.folders.folder_util_add_folder_tree") @patch("app.routes.folders.db_update_parent_ids_for_subtree") @patch("app.routes.folders.db_find_parent_folder_id") From ad31a4fda0fc08eb48dd079ffa180633d8a5575a Mon Sep 17 00:00:00 2001 From: Nancy Wei Date: Thu, 25 Jun 2026 23:34:36 +1000 Subject: [PATCH 2/2] fix --- backend/tests/test_connection.py | 4 --- backend/tests/test_folders.py | 54 +++++++++++++++++++++++++++++--- 2 files changed, 49 insertions(+), 9 deletions(-) diff --git a/backend/tests/test_connection.py b/backend/tests/test_connection.py index cc912e07f..7b1d286a0 100644 --- a/backend/tests/test_connection.py +++ b/backend/tests/test_connection.py @@ -45,9 +45,6 @@ def test_get_db_connection_rolls_back_on_exception(tmp_path): conn = sqlite3.connect(db_path) result = conn.execute("SELECT name FROM test").fetchone() conn.close() - - assert result is None - # CREATE TABLE was rolled back too assert result is None def test_get_db_connection_enables_pragmas(tmp_path): @@ -58,7 +55,6 @@ def test_get_db_connection_enables_pragmas(tmp_path): foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0] ignore_check = conn.execute("PRAGMA ignore_check_constraints").fetchone()[0] recursive_triggers = conn.execute("PRAGMA recursive_triggers").fetchone()[0] - case_sensitive_like = conn.execute("PRAGMA case_sensitive_like").fetchone() assert foreign_keys == 1 assert ignore_check == 0 diff --git a/backend/tests/test_folders.py b/backend/tests/test_folders.py index addcc287e..459b3bf61 100644 --- a/backend/tests/test_folders.py +++ b/backend/tests/test_folders.py @@ -1,4 +1,6 @@ import sqlite3 +import uuid + import pytest from fastapi import FastAPI from fastapi.testclient import TestClient @@ -739,6 +741,34 @@ def test_db_insert_folders_batch(self, test_db): assert rows == [("folder-id-1",), ("folder-id-2",)] + @patch("app.database.folders.sqlite3.connect") + def test_db_insert_folders_batch_error(self, mock_connect): + mock_conn = MagicMock() + mock_cursor = MagicMock() + + mock_connect.return_value = mock_conn + mock_conn.cursor.return_value = mock_cursor + + mock_cursor.executemany.side_effect = sqlite3.Error("Database connection failed") + + with pytest.raises(sqlite3.Error): + db_insert_folders_batch([ + ("folder-id-1", "/tmp/photos", None, 1693526400, True, False) + ]) + + mock_conn.rollback.assert_called_once() + mock_conn.close.assert_called_once() + + def test_db_insert_folder_generates_uuid_when_folder_id_none(self, test_db, tmp_path): + folder = tmp_path / "photos" + folder.mkdir() + fake_uuid = uuid.UUID("12345678-1234-5678-1234-567812345678") + with patch("app.database.folders.DATABASE_PATH", test_db), \ + patch("app.database.folders.uuid.uuid4", return_value=fake_uuid): + create_folders_table_for_test(test_db) + result = db_insert_folder(str(folder), folder_id=None) + assert result == str(fake_uuid) + def test_db_insert_folder_success(self, test_db, tmp_path): create_folders_table_for_test(test_db) @@ -816,9 +846,20 @@ def test_db_get_all_folder_ids(self, test_db): assert set(result) == {"folder-id-1", "folder-id-2"} + @patch("app.database.folders.sqlite3.connect") + def test_db_delete_folders_batch_error(self, mock_connect): + mock_conn = MagicMock() + mock_cursor = MagicMock() + mock_connect.return_value = mock_conn + mock_conn.cursor.return_value = mock_cursor + mock_cursor.execute.side_effect = sqlite3.Error("Database connection failed") + with pytest.raises(sqlite3.Error): + db_delete_folders_batch(["folder-id-1"]) + mock_conn.rollback.assert_called_once() + mock_conn.close.assert_called_once() + def test_db_delete_folders_batch_empty_list(self, test_db): result = db_delete_folders_batch([]) - assert result == 0 def test_db_delete_folders_batch_success(self, test_db): @@ -891,6 +932,10 @@ def test_db_folder_exists_true_false(self, test_db, tmp_path): assert db_folder_exists(str(folder)) is True assert db_folder_exists(str(tmp_path / "missing")) is False + def test_db_find_parent_folder_id_root_returns_none(self): + result = db_find_parent_folder_id("/") + assert result is None + def test_db_find_parent_folder_id_found(self, test_db): conn = sqlite3.connect(test_db) cursor = conn.cursor() @@ -906,7 +951,6 @@ def test_db_find_parent_folder_id_found(self, test_db): def test_db_find_parent_folder_id_not_found(self, test_db): conn = sqlite3.connect(test_db) - cursor = conn.cursor() create_folders_table_for_test(test_db) conn.commit() conn.close() @@ -1073,9 +1117,9 @@ def test_db_get_direct_child_folders(self,test_db): ("child_2", "/root/child2"), } - # ============================================================================ - # Integration & Workflow Tests - # ============================================================================ +# ============================================================================ +# Integration & Workflow Tests +# ============================================================================ class TestFoldersIntegration: @patch("app.routes.folders.folder_util_add_folder_tree") @patch("app.routes.folders.db_update_parent_ids_for_subtree")