opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-27 06:39:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use sqlite3 session in test_base_trace_instance (#38744)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-24 12:55:32 +09:00 committed by GitHub
parent 34e2205efa
commit bbb133cd58
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 84 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

152
api/tests/unit_tests/core/ops/test_base_trace_instance.py
View File

@ -1,12 +1,18 @@
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from sqlalchemy import Engine
from sqlalchemy.orm import Session
from core.ops.base_trace_instance import BaseTraceInstance
from core.ops.entities.config_entity import BaseTracingConfig
from core.ops.entities.trace_entity import BaseTraceInfo
from models import Account, App, TenantAccountJoin
from models import Account, App, Tenant, TenantAccountJoin
from models.account import TenantAccountRole
from models.model import AppMode
TABLES = (App, Account, Tenant, TenantAccountJoin)
class ConcreteTraceInstance(BaseTraceInstance):
@ -17,91 +23,101 @@ class ConcreteTraceInstance(BaseTraceInstance):
super().trace(trace_info)
@pytest.fixture
def mock_db_session(monkeypatch: pytest.MonkeyPatch):
mock_session = MagicMock(spec=Session)
mock_session.__enter__.return_value = mock_session
mock_session.__exit__.return_value = None
mock_session_class = MagicMock(return_value=mock_session)
monkeypatch.setattr("core.ops.base_trace_instance.Session", mock_session_class)
monkeypatch.setattr("core.ops.base_trace_instance.db", MagicMock())
return mock_session
@pytest.fixture(autouse=True)
def _bind_production_sessions(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
"""Bind both service-owned ORM sessions to the test's SQLite engine."""
database = SimpleNamespace(engine=sqlite_engine)
monkeypatch.setattr("core.ops.base_trace_instance.db", database)
monkeypatch.setattr("models.account.db", database)
def test_get_service_account_with_tenant_app_not_found(mock_db_session):
mock_db_session.scalar.return_value = None
def _persist_app(session: Session, *, created_by: str | None) -> App:
app = App(
id="some_app_id",
tenant_id="tenant_id",
name="Trace App",
description="",
mode=AppMode.CHAT,
icon_type=None,
icon=None,
icon_background=None,
app_model_config_id=None,
workflow_id=None,
enable_site=True,
enable_api=True,
max_active_requests=None,
created_by=created_by,
)
session.add(app)
session.commit()
return app
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
def _persist_account(session: Session) -> Account:
account = Account(name="Creator", email="creator@example.com")
account.id = "creator_id"
session.add(account)
session.commit()
return account
def _trace_instance() -> ConcreteTraceInstance:
# Tracing configuration is a domain collaborator, not an ORM session.
return ConcreteTraceInstance(MagicMock(spec=BaseTracingConfig))
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_app_not_found(sqlite_session: Session):
with pytest.raises(ValueError, match="App with id some_app_id not found"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_no_creator(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = None
mock_db_session.scalar.return_value = mock_app
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_no_creator(sqlite_session: Session):
_persist_app(sqlite_session, created_by=None)
with pytest.raises(ValueError, match="App with id some_app_id has no creator"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_creator_not_found(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
# First call to scalar returns app, second returns None (for account)
mock_db_session.scalar.side_effect = [mock_app, None]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_creator_not_found(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
with pytest.raises(ValueError, match="Creator account with id creator_id not found for app some_app_id"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_tenant_not_found(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
mock_account = MagicMock(spec=Account)
mock_account.id = "creator_id"
mock_db_session.scalar.side_effect = [mock_app, mock_account, None]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_tenant_not_found(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
_persist_account(sqlite_session)
with pytest.raises(ValueError, match="Current tenant not found for account creator_id"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_success(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_success(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
_persist_account(sqlite_session)
tenant = Tenant(name="Workspace")
tenant.id = "tenant_id"
sqlite_session.add_all(
[
tenant,
TenantAccountJoin(
tenant_id=tenant.id,
account_id="creator_id",
current=True,
role=TenantAccountRole.OWNER,
),
]
)
sqlite_session.commit()
mock_account = MagicMock(spec=Account)
mock_account.id = "creator_id"
result = _trace_instance().get_service_account_with_tenant("some_app_id")
mock_tenant_join = MagicMock(spec=TenantAccountJoin)
mock_tenant_join.tenant_id = "tenant_id"
mock_db_session.scalar.side_effect = [mock_app, mock_account, mock_tenant_join]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
result = instance.get_service_account_with_tenant("some_app_id")
assert result == mock_account
mock_account.set_tenant_id_with_session.assert_called_once_with("tenant_id", session=mock_db_session)
assert result.id == "creator_id"
assert result.current_tenant_id == "tenant_id"
assert result.current_role == TenantAccountRole.OWNER