test: use sqlite3 session in test_credential_utils (#38747)

api/tests/unit_tests/core/helper/test_credential_utils.py

@@ -3,9 +3,48 @@ from typing import cast
 
 import pytest
 from pytest_mock import MockerFixture
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session, sessionmaker
 
+import core.db.session_factory as session_factory_module
 from core.entities import PluginCredentialType
 from core.helper.credential_utils import check_credential_policy_compliance, is_credential_exists
+from models.provider import ProviderCredential, ProviderModelCredential
+from models.tools import BuiltinToolProvider
+
+SQLITE_MODELS = (ProviderCredential, ProviderModelCredential, BuiltinToolProvider)
+pytestmark = [
+    pytest.mark.usefixtures("sqlite_session"),
+    pytest.mark.parametrize("sqlite_session", [SQLITE_MODELS], indirect=True),
+]
+
+
+@pytest.fixture(autouse=True)
+def bind_session_factory(sqlite_engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Bind service-owned sessions to the current test's SQLite engine."""
+
+    maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
+    monkeypatch.setattr(session_factory_module, "create_session", maker)
+
+
+def _persist_credential(session: Session, credential_type: PluginCredentialType) -> None:
+    if credential_type == PluginCredentialType.MODEL:
+        credential = ProviderCredential(
+            tenant_id="tenant-1",
+            provider_name="openai",
+            credential_name="Default",
+            encrypted_config="{}",
+        )
+    else:
+        credential = BuiltinToolProvider(
+            tenant_id="tenant-1",
+            user_id="user-1",
+            provider="openai",
+            name="Default",
+        )
+    credential.id = "cred-1"
+    session.add(credential)
+    session.commit()
 
 
 def test_check_credential_policy_compliance_returns_when_feature_disabled(
@@ -109,53 +148,37 @@ def test_check_credential_policy_compliance_returns_when_credential_id_empty(
     ],
 )
 def test_is_credential_exists_by_type(
-    mocker: MockerFixture,
+    sqlite_session: Session,
     credential_type: PluginCredentialType,
     scalar_result: str | None,
     expected: bool,
 ) -> None:
-    session = mocker.MagicMock()
-    session.scalar.return_value = scalar_result
-    session_context = mocker.MagicMock()
-    session_context.__enter__.return_value = session
-    mocker.patch("core.db.session_factory.create_session", return_value=session_context)
+    if scalar_result is not None:
+        _persist_credential(sqlite_session, credential_type)
 
     result = is_credential_exists("cred-1", credential_type)
 
     assert result is expected
-    session.scalar.assert_called_once()
 
 
-def test_is_credential_exists_returns_false_for_unknown_type(
-    mocker: MockerFixture,
-) -> None:
-    session = mocker.MagicMock()
-    session_context = mocker.MagicMock()
-    session_context.__enter__.return_value = session
-    mocker.patch("core.db.session_factory.create_session", return_value=session_context)
-
+def test_is_credential_exists_returns_false_for_unknown_type() -> None:
     result = is_credential_exists("cred-1", cast(PluginCredentialType, "unknown"))
 
     assert result is False
-    session.scalar.assert_not_called()
 
 
 def test_is_credential_exists_uses_configured_session_factory_without_flask_app_context(
     mocker: MockerFixture,
+    sqlite_session: Session,
 ) -> None:
     class RaisingDB:
         @property
         def engine(self):
             raise RuntimeError("Working outside of application context.")
 
-    session = mocker.MagicMock()
-    session.scalar.return_value = "model-credential"
-    session_context = mocker.MagicMock()
-    session_context.__enter__.return_value = session
-    create_session = mocker.patch("core.db.session_factory.create_session", return_value=session_context)
+    _persist_credential(sqlite_session, PluginCredentialType.MODEL)
     mocker.patch("extensions.ext_database.db", new=RaisingDB())
 
     result = is_credential_exists("cred-1", PluginCredentialType.MODEL)
 
     assert result is True
-    create_session.assert_called_once_with()