diff --git a/api/tests/test_containers_integration_tests/models/test_workflow_node_execution_model.py b/api/tests/test_containers_integration_tests/models/test_workflow_node_execution_model.py new file mode 100644 index 0000000000..14c2263110 --- /dev/null +++ b/api/tests/test_containers_integration_tests/models/test_workflow_node_execution_model.py @@ -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 diff --git a/api/tests/unit_tests/models/test_workflow_trigger_log.py b/api/tests/unit_tests/models/test_workflow_trigger_log.py deleted file mode 100644 index 7fdad92fb6..0000000000 --- a/api/tests/unit_tests/models/test_workflow_trigger_log.py +++ /dev/null @@ -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