mirror of
https://github.com/langgenius/dify.git
synced 2026-04-16 02:16:57 +08:00
test: migrate WorkflowNodeExecutionModel creator property SQL tests to Testcontainers (#34958)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
wdeveloper16
GitHub
parent
169184ac9b
commit
50206ae8a7
@ -0,0 +1,170 @@
|
||||
"""
|
||||
Integration tests for WorkflowNodeExecutionModel.created_by_account and .created_by_end_user.
|
||||
|
||||
Migrated from unit_tests/models/test_workflow_trigger_log.py, replacing
|
||||
monkeypatch.setattr(db.session, "scalar", ...) with real Account/EndUser rows
|
||||
persisted in PostgreSQL so the db.session.get() call executes against the DB.
|
||||
"""
|
||||
|
||||
from collections.abc import Generator
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models.account import Account
|
||||
from models.enums import CreatorUserRole
|
||||
from models.model import App, AppMode, EndUser
|
||||
from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom
|
||||
|
||||
|
||||
class TestWorkflowNodeExecutionModelCreatedBy:
|
||||
"""Integration tests for WorkflowNodeExecutionModel creator lookup properties."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _auto_rollback(self, db_session_with_containers: Session) -> Generator[None, None, None]:
|
||||
"""Automatically rollback session changes after each test."""
|
||||
yield
|
||||
db_session_with_containers.rollback()
|
||||
|
||||
def _create_account(self, db_session: Session) -> Account:
|
||||
account = Account(
|
||||
name="Test Account",
|
||||
email=f"test_{uuid4()}@example.com",
|
||||
password="hashed-password",
|
||||
password_salt="salt",
|
||||
interface_language="en-US",
|
||||
timezone="UTC",
|
||||
)
|
||||
db_session.add(account)
|
||||
db_session.flush()
|
||||
return account
|
||||
|
||||
def _create_end_user(self, db_session: Session, tenant_id: str, app_id: str) -> EndUser:
|
||||
end_user = EndUser(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
type="service_api",
|
||||
external_user_id=f"ext-{uuid4()}",
|
||||
name="End User",
|
||||
session_id=f"session-{uuid4()}",
|
||||
)
|
||||
end_user.is_anonymous = False
|
||||
db_session.add(end_user)
|
||||
db_session.flush()
|
||||
return end_user
|
||||
|
||||
def _create_app(self, db_session: Session, tenant_id: str, created_by: str) -> App:
|
||||
app = App(
|
||||
tenant_id=tenant_id,
|
||||
name=f"App {uuid4()}",
|
||||
mode=AppMode.WORKFLOW,
|
||||
enable_site=False,
|
||||
enable_api=True,
|
||||
is_demo=False,
|
||||
is_public=False,
|
||||
is_universal=False,
|
||||
created_by=created_by,
|
||||
updated_by=created_by,
|
||||
)
|
||||
db_session.add(app)
|
||||
db_session.flush()
|
||||
return app
|
||||
|
||||
def _make_execution(
|
||||
self, tenant_id: str, app_id: str, created_by_role: str, created_by: str
|
||||
) -> WorkflowNodeExecutionModel:
|
||||
return WorkflowNodeExecutionModel(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app_id,
|
||||
workflow_id=str(uuid4()),
|
||||
triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
|
||||
workflow_run_id=None,
|
||||
index=1,
|
||||
predecessor_node_id=None,
|
||||
node_execution_id=None,
|
||||
node_id="n1",
|
||||
node_type="start",
|
||||
title="Start",
|
||||
inputs=None,
|
||||
process_data=None,
|
||||
outputs=None,
|
||||
status="succeeded",
|
||||
error=None,
|
||||
elapsed_time=0.0,
|
||||
execution_metadata=None,
|
||||
created_by_role=created_by_role,
|
||||
created_by=created_by,
|
||||
)
|
||||
|
||||
def test_created_by_account_returns_account_when_role_is_account(self, db_session_with_containers: Session) -> None:
|
||||
"""created_by_account returns the Account row when role is ACCOUNT."""
|
||||
account = self._create_account(db_session_with_containers)
|
||||
app = self._create_app(db_session_with_containers, str(uuid4()), account.id)
|
||||
|
||||
execution = self._make_execution(
|
||||
tenant_id=app.tenant_id,
|
||||
app_id=app.id,
|
||||
created_by_role=CreatorUserRole.ACCOUNT.value,
|
||||
created_by=account.id,
|
||||
)
|
||||
|
||||
result = execution.created_by_account
|
||||
|
||||
assert result is not None
|
||||
assert result.id == account.id
|
||||
|
||||
def test_created_by_account_returns_none_when_role_is_end_user(self, db_session_with_containers: Session) -> None:
|
||||
"""created_by_account returns None when role is END_USER, even if an Account exists."""
|
||||
account = self._create_account(db_session_with_containers)
|
||||
app = self._create_app(db_session_with_containers, str(uuid4()), account.id)
|
||||
|
||||
execution = self._make_execution(
|
||||
tenant_id=app.tenant_id,
|
||||
app_id=app.id,
|
||||
created_by_role=CreatorUserRole.END_USER.value,
|
||||
created_by=account.id,
|
||||
)
|
||||
|
||||
result = execution.created_by_account
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_created_by_end_user_returns_end_user_when_role_is_end_user(
|
||||
self, db_session_with_containers: Session
|
||||
) -> None:
|
||||
"""created_by_end_user returns the EndUser row when role is END_USER."""
|
||||
account = self._create_account(db_session_with_containers)
|
||||
tenant_id = str(uuid4())
|
||||
app = self._create_app(db_session_with_containers, tenant_id, account.id)
|
||||
end_user = self._create_end_user(db_session_with_containers, tenant_id, app.id)
|
||||
|
||||
execution = self._make_execution(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app.id,
|
||||
created_by_role=CreatorUserRole.END_USER.value,
|
||||
created_by=end_user.id,
|
||||
)
|
||||
|
||||
result = execution.created_by_end_user
|
||||
|
||||
assert result is not None
|
||||
assert result.id == end_user.id
|
||||
|
||||
def test_created_by_end_user_returns_none_when_role_is_account(self, db_session_with_containers: Session) -> None:
|
||||
"""created_by_end_user returns None when role is ACCOUNT, even if an EndUser exists."""
|
||||
account = self._create_account(db_session_with_containers)
|
||||
tenant_id = str(uuid4())
|
||||
app = self._create_app(db_session_with_containers, tenant_id, account.id)
|
||||
end_user = self._create_end_user(db_session_with_containers, tenant_id, app.id)
|
||||
|
||||
execution = self._make_execution(
|
||||
tenant_id=tenant_id,
|
||||
app_id=app.id,
|
||||
created_by_role=CreatorUserRole.ACCOUNT.value,
|
||||
created_by=end_user.id,
|
||||
)
|
||||
|
||||
result = execution.created_by_end_user
|
||||
|
||||
assert result is None
|
||||
@ -1,188 +0,0 @@
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
from models.engine import db
|
||||
from models.enums import CreatorUserRole
|
||||
from models.workflow import WorkflowNodeExecutionModel
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_db_scalar(monkeypatch):
|
||||
"""Provide a controllable fake for db.session.scalar (SQLAlchemy 2.0 style)."""
|
||||
calls = []
|
||||
|
||||
def _install(side_effect):
|
||||
def _fake_scalar(statement):
|
||||
calls.append(statement)
|
||||
return side_effect(statement)
|
||||
|
||||
# Patch the modern API used by the model implementation
|
||||
monkeypatch.setattr(db.session, "scalar", _fake_scalar)
|
||||
|
||||
# Backward-compatibility: if the implementation still uses db.session.get,
|
||||
# make it delegate to the same side_effect so tests remain valid on older code.
|
||||
if hasattr(db.session, "get"):
|
||||
|
||||
def _fake_get(*_args, **_kwargs):
|
||||
return side_effect(None)
|
||||
|
||||
monkeypatch.setattr(db.session, "get", _fake_get)
|
||||
|
||||
return calls
|
||||
|
||||
return _install
|
||||
|
||||
|
||||
def make_account(id_: str = "acc-1"):
|
||||
# Use a simple object to avoid constructing a full SQLAlchemy model instance
|
||||
# Python 3.12 forbids reassigning __class__ for SimpleNamespace; not needed here.
|
||||
obj = types.SimpleNamespace()
|
||||
obj.id = id_
|
||||
return obj
|
||||
|
||||
|
||||
def make_end_user(id_: str = "user-1"):
|
||||
# Lightweight stand-in object; no need to spoof class identity.
|
||||
obj = types.SimpleNamespace()
|
||||
obj.id = id_
|
||||
return obj
|
||||
|
||||
|
||||
def test_created_by_account_returns_account_when_role_account(fake_db_scalar):
|
||||
account = make_account("acc-1")
|
||||
|
||||
# The implementation uses db.session.scalar(select(Account)...). We only need to
|
||||
# return the expected object when called; the exact SQL is irrelevant for this unit test.
|
||||
def side_effect(_statement):
|
||||
return account
|
||||
|
||||
fake_db_scalar(side_effect)
|
||||
|
||||
log = WorkflowNodeExecutionModel(
|
||||
tenant_id="t1",
|
||||
app_id="a1",
|
||||
workflow_id="w1",
|
||||
triggered_from="workflow-run",
|
||||
workflow_run_id=None,
|
||||
index=1,
|
||||
predecessor_node_id=None,
|
||||
node_execution_id=None,
|
||||
node_id="n1",
|
||||
node_type="start",
|
||||
title="Start",
|
||||
inputs=None,
|
||||
process_data=None,
|
||||
outputs=None,
|
||||
status="succeeded",
|
||||
error=None,
|
||||
elapsed_time=0.0,
|
||||
execution_metadata=None,
|
||||
created_by_role=CreatorUserRole.ACCOUNT.value,
|
||||
created_by="acc-1",
|
||||
)
|
||||
|
||||
assert log.created_by_account is account
|
||||
|
||||
|
||||
def test_created_by_account_returns_none_when_role_not_account(fake_db_scalar):
|
||||
# Even if an Account with matching id exists, property should return None when role is END_USER
|
||||
account = make_account("acc-1")
|
||||
|
||||
def side_effect(_statement):
|
||||
return account
|
||||
|
||||
fake_db_scalar(side_effect)
|
||||
|
||||
log = WorkflowNodeExecutionModel(
|
||||
tenant_id="t1",
|
||||
app_id="a1",
|
||||
workflow_id="w1",
|
||||
triggered_from="workflow-run",
|
||||
workflow_run_id=None,
|
||||
index=1,
|
||||
predecessor_node_id=None,
|
||||
node_execution_id=None,
|
||||
node_id="n1",
|
||||
node_type="start",
|
||||
title="Start",
|
||||
inputs=None,
|
||||
process_data=None,
|
||||
outputs=None,
|
||||
status="succeeded",
|
||||
error=None,
|
||||
elapsed_time=0.0,
|
||||
execution_metadata=None,
|
||||
created_by_role=CreatorUserRole.END_USER.value,
|
||||
created_by="acc-1",
|
||||
)
|
||||
|
||||
assert log.created_by_account is None
|
||||
|
||||
|
||||
def test_created_by_end_user_returns_end_user_when_role_end_user(fake_db_scalar):
|
||||
end_user = make_end_user("user-1")
|
||||
|
||||
def side_effect(_statement):
|
||||
return end_user
|
||||
|
||||
fake_db_scalar(side_effect)
|
||||
|
||||
log = WorkflowNodeExecutionModel(
|
||||
tenant_id="t1",
|
||||
app_id="a1",
|
||||
workflow_id="w1",
|
||||
triggered_from="workflow-run",
|
||||
workflow_run_id=None,
|
||||
index=1,
|
||||
predecessor_node_id=None,
|
||||
node_execution_id=None,
|
||||
node_id="n1",
|
||||
node_type="start",
|
||||
title="Start",
|
||||
inputs=None,
|
||||
process_data=None,
|
||||
outputs=None,
|
||||
status="succeeded",
|
||||
error=None,
|
||||
elapsed_time=0.0,
|
||||
execution_metadata=None,
|
||||
created_by_role=CreatorUserRole.END_USER.value,
|
||||
created_by="user-1",
|
||||
)
|
||||
|
||||
assert log.created_by_end_user is end_user
|
||||
|
||||
|
||||
def test_created_by_end_user_returns_none_when_role_not_end_user(fake_db_scalar):
|
||||
end_user = make_end_user("user-1")
|
||||
|
||||
def side_effect(_statement):
|
||||
return end_user
|
||||
|
||||
fake_db_scalar(side_effect)
|
||||
|
||||
log = WorkflowNodeExecutionModel(
|
||||
tenant_id="t1",
|
||||
app_id="a1",
|
||||
workflow_id="w1",
|
||||
triggered_from="workflow-run",
|
||||
workflow_run_id=None,
|
||||
index=1,
|
||||
predecessor_node_id=None,
|
||||
node_execution_id=None,
|
||||
node_id="n1",
|
||||
node_type="start",
|
||||
title="Start",
|
||||
inputs=None,
|
||||
process_data=None,
|
||||
outputs=None,
|
||||
status="succeeded",
|
||||
error=None,
|
||||
elapsed_time=0.0,
|
||||
execution_metadata=None,
|
||||
created_by_role=CreatorUserRole.ACCOUNT.value,
|
||||
created_by="user-1",
|
||||
)
|
||||
|
||||
assert log.created_by_end_user is None
|
||||
Loading…
Reference in New Issue
Block a user