test: use SQLite sessions in service API fixtures (#38784)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/tests/unit_tests/conftest.py

@@ -37,6 +37,7 @@ os.environ.setdefault("STORAGE_TYPE", "opendal")
 
 from core.db.session_factory import configure_session_factory, session_factory
 from extensions import ext_redis
+from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
 from models.base import TypeBase
 
 
@@ -148,32 +149,45 @@ def _configure_session_factory(_unit_test_engine):
         configure_session_factory(_unit_test_engine, expire_on_commit=False)
 
 
-def setup_mock_tenant_owner_execute_result(mock_db, mock_tenant, mock_owner):
-    """
-    Helper to stub the tenant-owner execute result for service API app authentication.
+def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:
+    """Persist the owner identity resolved by service-API app authentication.
 
-    The validate_app_token decorator currently resolves the active tenant owner
-    via db.session.execute(select(Tenant, Account)...).one_or_none().
-
-    Args:
-        mock_db: The mocked db object
-        mock_tenant: Mock tenant object to return
-        mock_owner: Mock owner object to return from the execute result
+    The legacy name is retained temporarily for consumers on independent
+    conversion branches, but this helper no longer fabricates an execute result.
     """
+    membership = TenantAccountJoin(
+        tenant_id=tenant.id,
+        account_id=owner.id,
+        role=TenantAccountRole.OWNER,
+    )
+    owner._current_tenant = tenant
+    session.add_all([tenant, owner, membership])
+    session.commit()
+    return membership
+
+
+def persist_service_api_dataset_owner(
+    session: Session,
+    tenant: Tenant,
+    tenant_account_join: TenantAccountJoin,
+) -> None:
+    """Persist the tenant-owner mapping resolved by dataset-token authentication."""
+    session.add_all([tenant, tenant_account_join])
+    session.commit()
+
+
+def setup_mock_tenant_owner_execute_result(mock_db: MagicMock, mock_tenant: object, mock_owner: object) -> None:
+    """Stub the legacy owner query; SQLite-backed tests use ``persist_service_api_tenant_owner``."""
     mock_db.session.execute.return_value.one_or_none.return_value = (mock_tenant, mock_owner)
 
 
-def setup_mock_dataset_owner_execute_result(mock_db, mock_tenant, mock_tenant_account_join):
-    """
-    Helper to stub the tenant-owner execute result for dataset token authentication.
-
-    The validate_dataset_token decorator currently resolves the owner mapping via
-    db.session.execute(select(Tenant, TenantAccountJoin)...).one_or_none(), and
-    then loads the Account separately via db.session.get(...).
-
-    Args:
-        mock_db: The mocked db object
-        mock_tenant: Mock tenant object to return
-        mock_tenant_account_join: Mock tenant-account join object to return
-    """
-    mock_db.session.execute.return_value.one_or_none.return_value = (mock_tenant, mock_tenant_account_join)
+def setup_mock_dataset_owner_execute_result(
+    mock_db: MagicMock,
+    mock_tenant: object,
+    mock_tenant_account_join: object,
+) -> None:
+    """Stub the legacy dataset-owner query; SQLite tests use ``persist_service_api_dataset_owner``."""
+    mock_db.session.execute.return_value.one_or_none.return_value = (
+        mock_tenant,
+        mock_tenant_account_join,
+    )

api/tests/unit_tests/controllers/service_api/conftest.py

@@ -7,18 +7,57 @@ Service API controller tests.
 """
 
 import uuid
+from collections.abc import Iterator
+from dataclasses import dataclass
 from unittest.mock import Mock
 
 import pytest
 from flask import Flask
+from sqlalchemy import Engine
+from sqlalchemy.orm import Session
 
 from core.rag.index_processor.constant.index_type import IndexStructureType
-from models.account import TenantStatus
+from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole, TenantStatus
+from models.base import TypeBase
 from models.model import App, AppMode, EndUser
-from tests.unit_tests.conftest import (
-    setup_mock_dataset_owner_execute_result,
-    setup_mock_tenant_owner_execute_result,
-)
+
+
+@dataclass(frozen=True)
+class ServiceApiIdentity:
+    """Persisted owner identity for service-API authentication tests."""
+
+    session: Session
+    tenant: Tenant
+    account: Account
+    membership: TenantAccountJoin
+
+
+@pytest.fixture
+def service_api_identity(sqlite_engine: Engine) -> Iterator[ServiceApiIdentity]:
+    """Yield an isolated SQLite session with a real active tenant owner."""
+    TypeBase.metadata.create_all(
+        sqlite_engine,
+        tables=[Account.__table__, Tenant.__table__, TenantAccountJoin.__table__],
+    )
+    with Session(sqlite_engine, expire_on_commit=False) as session:
+        tenant = Tenant(name="Service API Workspace")
+        tenant.id = str(uuid.uuid4())
+        account = Account(name="Service API Owner", email=f"owner-{tenant.id}@example.com")
+        account.id = str(uuid.uuid4())
+        membership = TenantAccountJoin(
+            tenant_id=tenant.id,
+            account_id=account.id,
+            role=TenantAccountRole.OWNER,
+        )
+        account._current_tenant = tenant
+        session.add_all([tenant, account, membership])
+        session.commit()
+        yield ServiceApiIdentity(
+            session=session,
+            tenant=tenant,
+            account=account,
+            membership=membership,
+        )
 
 
 @pytest.fixture
@@ -110,40 +149,6 @@ def mock_dataset_api_token(mock_tenant_id):
     return token
 
 
-class AuthenticationMocker:
-    """
-    Helper class to set up common authentication mocking patterns.
-
-    Usage:
-        auth_mocker = AuthenticationMocker()
-        with auth_mocker.mock_app_auth(mock_api_token, mock_app_model, mock_tenant):
-            # Test code here
-    """
-
-    @staticmethod
-    def setup_db_queries(mock_db, mock_app, mock_tenant, mock_account=None):
-        """Configure mock_db to return app and tenant via session.get()."""
-        mock_db.session.get.side_effect = [mock_app, mock_tenant]
-
-        if mock_account:
-            setup_mock_tenant_owner_execute_result(mock_db, mock_tenant, mock_account)
-
-    @staticmethod
-    def setup_dataset_auth(mock_db, mock_tenant, mock_account):
-        """Configure mock_db for dataset token authentication."""
-        mock_ta = Mock()
-        mock_ta.account_id = mock_account.id
-
-        setup_mock_dataset_owner_execute_result(mock_db, mock_tenant, mock_ta)
-        mock_db.session.get.return_value = mock_account
-
-
-@pytest.fixture
-def auth_mocker():
-    """Provide an AuthenticationMocker instance."""
-    return AuthenticationMocker()
-
-
 @pytest.fixture
 def mock_dataset():
     """Create a mock Dataset model."""

api/tests/unit_tests/controllers/service_api/test_conftest.py

@@ -0,0 +1,55 @@
+"""State-based checks for shared service-API authentication fixtures."""
+
+from uuid import uuid4
+
+from sqlalchemy import select
+
+from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
+from tests.unit_tests.conftest import (
+    persist_service_api_dataset_owner,
+    persist_service_api_tenant_owner,
+)
+from tests.unit_tests.controllers.service_api.conftest import ServiceApiIdentity
+
+
+def test_service_api_identity_persists_tenant_scoped_owner(service_api_identity: ServiceApiIdentity) -> None:
+    identity = service_api_identity
+
+    owner_row = identity.session.execute(
+        select(Tenant, Account)
+        .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
+        .join(Account, TenantAccountJoin.account_id == Account.id)
+        .where(
+            Tenant.id == identity.tenant.id,
+            TenantAccountJoin.role == TenantAccountRole.OWNER,
+        )
+    ).one()
+
+    assert owner_row == (identity.tenant, identity.account)
+    assert identity.account.current_tenant is identity.tenant
+
+
+def test_shared_helpers_persist_real_app_and_dataset_owner_rows(service_api_identity: ServiceApiIdentity) -> None:
+    session = service_api_identity.session
+    app_tenant = Tenant(name="App Workspace")
+    app_tenant.id = str(uuid4())
+    app_owner = Account(name="App Owner", email=f"app-owner-{app_tenant.id}@example.com")
+    app_owner.id = str(uuid4())
+
+    app_membership = persist_service_api_tenant_owner(session, app_tenant, app_owner)
+
+    dataset_tenant = Tenant(name="Dataset Workspace")
+    dataset_tenant.id = str(uuid4())
+    dataset_membership = TenantAccountJoin(
+        tenant_id=dataset_tenant.id,
+        account_id=service_api_identity.account.id,
+        role=TenantAccountRole.OWNER,
+    )
+    persist_service_api_dataset_owner(session, dataset_tenant, dataset_membership)
+
+    assert session.get(TenantAccountJoin, app_membership.id) is app_membership
+    assert session.execute(
+        select(Tenant, TenantAccountJoin)
+        .join(TenantAccountJoin, Tenant.id == TenantAccountJoin.tenant_id)
+        .where(Tenant.id == dataset_tenant.id)
+    ).one() == (dataset_tenant, dataset_membership)