From 142702dbc52497eee91066892bd1c4fb90012246 Mon Sep 17 00:00:00 2001 From: Nancy Wei Date: Thu, 25 Jun 2026 16:33:08 +1000 Subject: [PATCH 1/4] 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 8fdebc8f8..7aa8a5eb5 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 @@ -690,10 +691,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 2a04fe822742f88538362c1666de854b8ee7911a Mon Sep 17 00:00:00 2001 From: Nancy Wei Date: Thu, 25 Jun 2026 23:34:36 +1000 Subject: [PATCH 2/4] 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 7aa8a5eb5..9621419ac 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 @@ -741,6 +743,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) @@ -818,9 +848,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): @@ -893,6 +934,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() @@ -908,7 +953,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() @@ -1075,9 +1119,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") From 4b8efe6ffcfb886f5191f1d7450a7dc774a445e2 Mon Sep 17 00:00:00 2001 From: ROHAN PANDEY <95585299+rohan-pandeyy@users.noreply.github.com> Date: Sat, 25 Jul 2026 01:23:25 +0530 Subject: [PATCH 3/4] test(backend): use the real folders schema and align with conventions The folder unit tests built the folders table by hand, and that copy had swapped columns (parent_folder_id / folder_path) and no FOREIGN KEY, so ON DELETE CASCADE never existed under test. test_db_get_all_folder_details relied on it: its positional INSERT binds the path to parent_folder_id on the real schema, so the test only passed against the divergent copy. - Build the schema with db_create_folders_table()/db_create_images_table() in the fixture instead of a hand-written CREATE, and name the columns in that INSERT - Rebuild the test_db fixture on tempfile with try/finally cleanup, and patch settings/images/yolo_mapping too so the whole FK chain resolves when db_delete_folder turns foreign keys on - Give the two fixture-less unit tests test_db, so a change to their short-circuit can't point them at the user's real database - Restructure test_connection.py onto a shared fixture with section banners and classes; also assert defer_foreign_keys and that the connection is closed on exit - Reformat both files with black (the gate had been bypassed) Note: test_db_get_all_folders needs the db_get_all_folders connection leak fixed to tear down on Windows; that fix lands separately on main. --- backend/tests/test_connection.py | 107 ++++++++----- backend/tests/test_folders.py | 255 +++++++++++++++---------------- 2 files changed, 194 insertions(+), 168 deletions(-) diff --git a/backend/tests/test_connection.py b/backend/tests/test_connection.py index 7b1d286a0..374f9646f 100644 --- a/backend/tests/test_connection.py +++ b/backend/tests/test_connection.py @@ -1,61 +1,88 @@ +import os import sqlite3 +import tempfile +from typing import Iterator + import pytest -from unittest.mock import patch from app.database.connection import get_db_connection +# ############################## +# Pytest Fixtures +# ############################## -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",)) +@pytest.fixture(scope="function") +def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: + """Point the connection helper at a fresh tempfile database.""" + db_fd, db_path = tempfile.mkstemp() + os.close(db_fd) + try: + monkeypatch.setattr("app.config.settings.DATABASE_PATH", db_path) + monkeypatch.setattr("app.database.connection.DATABASE_PATH", db_path) + + yield db_path + finally: + os.unlink(db_path) - # 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 read_names(db_path: str) -> list: + """Read every row from the scratch table on a separate connection.""" + conn = sqlite3.connect(db_path) + rows = conn.execute("SELECT name FROM test").fetchall() + conn.close() + return rows -def test_get_db_connection_rolls_back_on_exception(tmp_path): - db_path = tmp_path / "test.db" +# ############################## +# Transaction handling +# ############################## - with patch("app.database.connection.DATABASE_PATH", str(db_path)): - # Create table first and commit successfully + +class TestGetDbConnection: + def test_commits_on_success(self, test_db): with get_db_connection() as conn: - conn.execute( - "CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)" - ) + conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)") + conn.execute("INSERT INTO test (name) VALUES (?)", ("Alice",)) + + # Visible from a fresh connection, so the commit really happened + assert read_names(test_db) == [("Alice",)] + + def test_rolls_back_on_exception(self, test_db): + 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",), - ) + 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 - -def test_get_db_connection_enables_pragmas(tmp_path): - db_path = tmp_path / "test.db" + assert read_names(test_db) == [] - with patch("app.database.connection.DATABASE_PATH", str(db_path)): + def test_closes_the_connection_on_the_way_out(self, test_db): 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] + conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)") + + # The handle is released, so using it again is an error + with pytest.raises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") - assert foreign_keys == 1 - assert ignore_check == 0 - assert recursive_triggers == 1 \ No newline at end of file + +# ############################## +# Pragmas +# ############################## + + +class TestConnectionPragmas: + @pytest.mark.parametrize( + "pragma, expected", + [ + ("foreign_keys", 1), + ("ignore_check_constraints", 0), + ("recursive_triggers", 1), + ("defer_foreign_keys", 0), + ], + ) + def test_enforcement_pragmas_are_set(self, test_db, pragma, expected): + with get_db_connection() as conn: + assert conn.execute(f"PRAGMA {pragma}").fetchone()[0] == expected diff --git a/backend/tests/test_folders.py b/backend/tests/test_folders.py index 9621419ac..e097c4264 100644 --- a/backend/tests/test_folders.py +++ b/backend/tests/test_folders.py @@ -13,26 +13,57 @@ 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 - +from app.database.folders import ( + db_create_folders_table, + 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, +) +from app.database.images import db_create_images_table +from app.database.yolo_mapping import db_create_YOLO_classes_table # ############################## # Pytest Fixtures # ############################## -@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(scope="function") +def test_db(monkeypatch): + """Point the folder DB modules at a fresh tempfile database.""" + db_fd, db_path = tempfile.mkstemp() + os.close(db_fd) + try: + monkeypatch.setattr("app.config.settings.DATABASE_PATH", db_path) + monkeypatch.setattr("app.database.folders.DATABASE_PATH", db_path) + monkeypatch.setattr("app.database.images.DATABASE_PATH", db_path) + monkeypatch.setattr("app.database.yolo_mapping.DATABASE_PATH", db_path) + + # Build the real schema rather than a hand-written copy: a divergent + # CREATE silently reorders columns and drops the ON DELETE CASCADE. + # db_delete_folder turns foreign keys on, so the whole FK chain + # (folders <- images <- image_classes -> mappings) has to resolve. + db_create_YOLO_classes_table() + db_create_folders_table() + db_create_images_table() # db_get_all_folder_details LEFT JOINs it + + yield db_path + finally: + os.unlink(db_path) @pytest.fixture @@ -693,52 +724,26 @@ def test_disable_ai_tagging_no_folders_updated(self, mock_disable_batch, client) assert data["success"] is True assert data["data"]["updated_count"] == 0 - # ============================================================================ # 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), - ]) + 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() + rows = conn.execute( + "SELECT folder_id FROM folders ORDER BY folder_id" + ).fetchall() conn.close() assert rows == [("folder-id-1",), ("folder-id-2",)] @@ -751,28 +756,31 @@ def test_db_insert_folders_batch_error(self, mock_connect): mock_connect.return_value = mock_conn mock_conn.cursor.return_value = mock_cursor - mock_cursor.executemany.side_effect = sqlite3.Error("Database connection failed") + 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) - ]) + 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): + 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) + with patch("app.database.folders.DATABASE_PATH", test_db), patch( + "app.database.folders.uuid.uuid4", return_value=fake_uuid + ): 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) folder = tmp_path / "photos" folder.mkdir() @@ -783,7 +791,6 @@ def test_db_insert_folder_success(self, test_db, tmp_path): 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() @@ -795,13 +802,11 @@ def test_db_insert_folder_existing_returns_existing_id(self, test_db, tmp_path): 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() @@ -813,7 +818,6 @@ def test_db_get_folder_id_from_path(self, test_db, tmp_path): 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() @@ -825,24 +829,26 @@ def test_db_get_folder_path_from_id(self, test_db, tmp_path): 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), - ]) + 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), - ]) + 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() @@ -865,12 +871,13 @@ def test_db_delete_folders_batch_empty_list(self, test_db): 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), - ]) + 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"]) @@ -879,7 +886,6 @@ def test_db_delete_folders_batch_success(self, test_db): 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() @@ -890,7 +896,6 @@ def test_db_delete_folder_success(self, test_db, tmp_path): 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() @@ -899,12 +904,13 @@ def test_db_delete_folder_not_exists(self, test_db, tmp_path): 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_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", @@ -924,7 +930,6 @@ def test_db_update_parent_ids_for_subtree(self, test_db): 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() @@ -934,17 +939,18 @@ 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): + def test_db_find_parent_folder_id_root_returns_none(self, test_db): + # Takes test_db even though the root check short-circuits, so a future + # change can't silently point this at the real user database. 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() - create_folders_table_for_test(test_db) cursor.execute( "INSERT INTO folders (folder_id, folder_path) VALUES (?, ?)", - ("parent-id", "/tmp/photos") + ("parent-id", "/tmp/photos"), ) conn.commit() conn.close() @@ -953,7 +959,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) - create_folders_table_for_test(test_db) conn.commit() conn.close() result = db_find_parent_folder_id("/tmp/photos/2024") @@ -962,10 +967,9 @@ def test_db_find_parent_folder_id_not_found(self, test_db): 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) + ("tmp", "/tmp", False), ) conn.commit() conn.close() @@ -1004,10 +1008,9 @@ def test_db_disable_ai_tagging_batch(self, mock_update_batch): assert result == 1 mock_update_batch.assert_called_once_with(["tmp"], False) - def test_db_get_folder_ids_by_path_prefix(self,test_db): + 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 (?, ?)", [ @@ -1024,9 +1027,8 @@ def test_db_get_folder_ids_by_path_prefix(self,test_db): ("folder-id-2", "/tmp/photos/2024"), ] - def test_db_get_folder_ids_by_paths(self,test_db): + 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( @@ -1039,44 +1041,40 @@ def test_db_get_folder_ids_by_paths(self,test_db): ) conn.commit() conn.close() - result = db_get_folder_ids_by_paths([ - "test_folder_1", - "test_folder_2", - ]) + 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): + def test_db_get_folder_ids_by_paths_empty(self, test_db): 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",)) + # Name the columns: a positional INSERT binds to the wrong ones, since + # parent_folder_id precedes folder_path in the real schema. + conn.execute( + """ + INSERT INTO folders (folder_id, folder_path, parent_folder_id, + last_modified_time, AI_Tagging, taggingCompleted) + VALUES (?, ?, ?, ?, ?, ?) + """, + ("folder-id-1", "/home/user/photos", None, 1693526400, True, False), + ) + conn.executemany( + "INSERT INTO images (id, path, folder_id) VALUES (?, ?, ?)", + [ + ("img-1", "/home/user/photos/a.jpg", "folder-id-1"), + ("img-2", "/home/user/photos/b.jpg", "folder-id-1"), + ], + ) conn.commit() conn.close() result = db_get_all_folder_details() @@ -1086,15 +1084,15 @@ def test_db_get_all_folder_details(self, test_db): "/home/user/photos", None, 1693526400, - 1, - 0, - 2, + 1, # AI_Tagging + 0, # taggingCompleted + "not_started", # indexing_status (schema default) + 2, # image_count ) ] - def test_db_get_direct_child_folders(self,test_db): + 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), @@ -1119,6 +1117,7 @@ def test_db_get_direct_child_folders(self,test_db): ("child_2", "/root/child2"), } + # ============================================================================ # Integration & Workflow Tests # ============================================================================ From c595fa3dd06d5adb63dab138dea6395392b2fddf Mon Sep 17 00:00:00 2001 From: ROHAN PANDEY <95585299+rohan-pandeyy@users.noreply.github.com> Date: Sat, 25 Jul 2026 02:23:56 +0530 Subject: [PATCH 4/4] test(backend): address review nitpicks on folder/connection tests - read_names: annotate as List[Tuple[str]] and close via try/finally - test_db fixture: add the type hints test_connection.py already carries - Give the mock-only tests the test_db fixture, so a drifted patch target can't write to the developer's real database - Drop the redundant DATABASE_PATH patch (the fixture already sets it) and the no-op connect in test_db_find_parent_folder_id_not_found - path-prefix query has no ORDER BY: compare as a set, not an ordered list - update_ai_tagging_batch: read the row back to assert the persisted value, not just rowcount; split the empty-list case into its own test - Add a test that update_parent_ids_for_subtree leaves a non-NULL parent untouched, pinning its NULL-only guard --- backend/tests/test_connection.py | 11 ++--- backend/tests/test_folders.py | 74 +++++++++++++++++++++++--------- 2 files changed, 60 insertions(+), 25 deletions(-) diff --git a/backend/tests/test_connection.py b/backend/tests/test_connection.py index 374f9646f..1b1b22f53 100644 --- a/backend/tests/test_connection.py +++ b/backend/tests/test_connection.py @@ -1,7 +1,7 @@ import os import sqlite3 import tempfile -from typing import Iterator +from typing import Iterator, List, Tuple import pytest @@ -26,12 +26,13 @@ def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: os.unlink(db_path) -def read_names(db_path: str) -> list: +def read_names(db_path: str) -> List[Tuple[str]]: """Read every row from the scratch table on a separate connection.""" conn = sqlite3.connect(db_path) - rows = conn.execute("SELECT name FROM test").fetchall() - conn.close() - return rows + try: + return conn.execute("SELECT name FROM test").fetchall() + finally: + conn.close() # ############################## diff --git a/backend/tests/test_folders.py b/backend/tests/test_folders.py index e097c4264..2ed2281ca 100644 --- a/backend/tests/test_folders.py +++ b/backend/tests/test_folders.py @@ -1,5 +1,6 @@ import sqlite3 import uuid +from typing import Iterator import pytest from fastapi import FastAPI @@ -43,7 +44,7 @@ @pytest.fixture(scope="function") -def test_db(monkeypatch): +def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: """Point the folder DB modules at a fresh tempfile database.""" db_fd, db_path = tempfile.mkstemp() os.close(db_fd) @@ -749,7 +750,7 @@ 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): + def test_db_insert_folders_batch_error(self, mock_connect, test_db): mock_conn = MagicMock() mock_cursor = MagicMock() @@ -774,9 +775,8 @@ def test_db_insert_folder_generates_uuid_when_folder_id_none( 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 - ): + # test_db already patches DATABASE_PATH, so only uuid4 needs stubbing + with patch("app.database.folders.uuid.uuid4", return_value=fake_uuid): result = db_insert_folder(str(folder), folder_id=None) assert result == str(fake_uuid) @@ -855,7 +855,7 @@ 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): + def test_db_delete_folders_batch_error(self, mock_connect, test_db): mock_conn = MagicMock() mock_cursor = MagicMock() mock_connect.return_value = mock_conn @@ -929,6 +929,31 @@ def test_db_update_parent_ids_for_subtree(self, test_db): assert parent_id == "root-id" + def test_db_update_parent_ids_for_subtree_preserves_existing_parent(self, test_db): + # The update only rewrites rows whose parent_folder_id is still NULL + db_insert_folders_batch( + [ + ("root-id", "/tmp/root", None, 1693526400, True, False), + ("child-id", "/tmp/root/child", "old-parent", 1693526500, True, False), + ] + ) + + db_update_parent_ids_for_subtree( + "/tmp/root", + {"/tmp/root/child": ("child-id", "root-id")}, + ) + + conn = sqlite3.connect(test_db) + try: + parent_id = conn.execute( + "SELECT parent_folder_id FROM folders WHERE folder_id = ?", + ("child-id",), + ).fetchone()[0] + finally: + conn.close() + + assert parent_id == "old-parent" + def test_db_folder_exists_true_false(self, test_db, tmp_path): folder = tmp_path / "photos" @@ -958,28 +983,36 @@ def test_db_find_parent_folder_id_found(self, test_db): assert result == "parent-id" def test_db_find_parent_folder_id_not_found(self, test_db): - conn = sqlite3.connect(test_db) - conn.commit() - conn.close() + # The fixture already yields an empty schema; nothing to seed here 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() - cursor.execute( + conn.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 + + assert db_update_ai_tagging_batch(["tmp"], True) == 1 + + # rowcount alone can't tell right column/value from wrong: read it back + conn = sqlite3.connect(test_db) + try: + ai_tagging = conn.execute( + "SELECT AI_Tagging FROM folders WHERE folder_id = ?", ("tmp",) + ).fetchone()[0] + finally: + conn.close() + assert ai_tagging == 1 + + def test_db_update_ai_tagging_batch_empty_list(self, test_db): + assert db_update_ai_tagging_batch([], True) == 0 @patch("app.database.folders.sqlite3.connect") - def test_db_update_ai_tagging_batch_sqlite_error(self, mock_connect): + def test_db_update_ai_tagging_batch_sqlite_error(self, mock_connect, test_db): mock_conn = MagicMock() mock_cursor = MagicMock() @@ -995,14 +1028,14 @@ def test_db_update_ai_tagging_batch_sqlite_error(self, mock_connect): 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): + def test_db_enable_ai_tagging_batch(self, mock_update_batch, test_db): 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): + def test_db_disable_ai_tagging_batch(self, mock_update_batch, test_db): mock_update_batch.return_value = 1 result = db_disable_ai_tagging_batch(["tmp"]) assert result == 1 @@ -1022,10 +1055,11 @@ def test_db_get_folder_ids_by_path_prefix(self, test_db): conn.commit() conn.close() result = db_get_folder_ids_by_path_prefix("/tmp") - assert result == [ + # The query has no ORDER BY, so row order isn't part of the contract + assert set(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)