mirror of
https://github.com/langgenius/dify.git
synced 2026-09-24 23:22:26 +08:00
feat: collaboration (#30781)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Co-authored-by: yyh <yuanyouhuilyz@gmail.com> Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
This commit is contained in:
非法操作
co-authored by
autofix-ci[bot]
gemini-code-assist[bot]
yyh
yyh
autofix-ci[bot]
gemini-code-assist[bot]
yyh
yyh
parent
cf4d7afb9c
commit
53a22aa41b
@@ -48,7 +48,7 @@ os.environ["OPENDAL_FS_ROOT"] = "/tmp/dify-storage"
|
||||
os.environ.setdefault("STORAGE_TYPE", "opendal")
|
||||
os.environ.setdefault("OPENDAL_SCHEME", "fs")
|
||||
|
||||
_CACHED_APP = create_app()
|
||||
_SIO_APP, _CACHED_APP = create_app()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
|
||||
@@ -369,7 +369,7 @@ def _create_app_with_containers() -> Flask:
|
||||
|
||||
# Create and configure the Flask application
|
||||
logger.info("Initializing Flask application...")
|
||||
app = create_app()
|
||||
sio_app, app = create_app()
|
||||
logger.info("Flask application created successfully")
|
||||
|
||||
# Initialize database schema
|
||||
|
||||
@@ -274,6 +274,7 @@ class TestFeatureService:
|
||||
mock_config.ENABLE_EMAIL_CODE_LOGIN = True
|
||||
mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
|
||||
mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = False
|
||||
mock_config.ENABLE_COLLABORATION_MODE = True
|
||||
mock_config.ALLOW_REGISTER = False
|
||||
mock_config.ALLOW_CREATE_WORKSPACE = False
|
||||
mock_config.MAIL_TYPE = "smtp"
|
||||
@@ -298,6 +299,7 @@ class TestFeatureService:
|
||||
# Verify authentication settings
|
||||
assert result.enable_email_code_login is True
|
||||
assert result.enable_email_password_login is False
|
||||
assert result.enable_collaboration_mode is True
|
||||
assert result.is_allow_register is False
|
||||
assert result.is_allow_create_workspace is False
|
||||
|
||||
@@ -401,6 +403,7 @@ class TestFeatureService:
|
||||
mock_config.ENABLE_EMAIL_CODE_LOGIN = True
|
||||
mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
|
||||
mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = False
|
||||
mock_config.ENABLE_COLLABORATION_MODE = False
|
||||
mock_config.ALLOW_REGISTER = True
|
||||
mock_config.ALLOW_CREATE_WORKSPACE = True
|
||||
mock_config.MAIL_TYPE = "smtp"
|
||||
@@ -422,6 +425,7 @@ class TestFeatureService:
|
||||
assert result.enable_email_code_login is True
|
||||
assert result.enable_email_password_login is True
|
||||
assert result.enable_social_oauth_login is False
|
||||
assert result.enable_collaboration_mode is False
|
||||
assert result.is_allow_register is True
|
||||
assert result.is_allow_create_workspace is True
|
||||
assert result.is_email_setup is True
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
@@ -347,3 +348,87 @@ def test_advanced_chat_run_conversation_not_exists(app, monkeypatch: pytest.Monk
|
||||
):
|
||||
with pytest.raises(NotFound):
|
||||
handler(api, app_model=SimpleNamespace(id="app"))
|
||||
|
||||
|
||||
def test_workflow_online_users_filters_inaccessible_workflow(app, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
app_id_1 = "11111111-1111-1111-1111-111111111111"
|
||||
app_id_2 = "22222222-2222-2222-2222-222222222222"
|
||||
signed_avatar_url = "https://files.example.com/signed/avatar-1"
|
||||
sign_avatar = Mock(return_value=signed_avatar_url)
|
||||
monkeypatch.setattr(workflow_module, "current_account_with_tenant", lambda: (SimpleNamespace(), "tenant-1"))
|
||||
monkeypatch.setattr(
|
||||
workflow_module,
|
||||
"WorkflowService",
|
||||
lambda: SimpleNamespace(get_accessible_app_ids=lambda app_ids, tenant_id: {app_id_1}),
|
||||
)
|
||||
monkeypatch.setattr(workflow_module.file_helpers, "get_signed_file_url", sign_avatar)
|
||||
|
||||
workflow_module.redis_client.hgetall.side_effect = lambda key: (
|
||||
{
|
||||
b"sid-1": json.dumps(
|
||||
{
|
||||
"user_id": "u-1",
|
||||
"username": "Alice",
|
||||
"avatar": "avatar-file-id",
|
||||
"sid": "sid-1",
|
||||
}
|
||||
)
|
||||
}
|
||||
if key == f"{workflow_module.WORKFLOW_ONLINE_USERS_PREFIX}{app_id_1}"
|
||||
else {}
|
||||
)
|
||||
|
||||
api = workflow_module.WorkflowOnlineUsersApi()
|
||||
handler = _unwrap(api.get)
|
||||
|
||||
with app.test_request_context(
|
||||
f"/apps/workflows/online-users?app_ids={app_id_1},{app_id_2}",
|
||||
method="GET",
|
||||
):
|
||||
response = handler(api)
|
||||
|
||||
assert response == {
|
||||
"data": [
|
||||
{
|
||||
"app_id": app_id_1,
|
||||
"users": [
|
||||
{
|
||||
"user_id": "u-1",
|
||||
"username": "Alice",
|
||||
"avatar": signed_avatar_url,
|
||||
"sid": "sid-1",
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
}
|
||||
workflow_module.redis_client.hgetall.assert_called_once_with(
|
||||
f"{workflow_module.WORKFLOW_ONLINE_USERS_PREFIX}{app_id_1}"
|
||||
)
|
||||
sign_avatar.assert_called_once_with("avatar-file-id")
|
||||
|
||||
|
||||
def test_workflow_online_users_rejects_excessive_workflow_ids(app, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(workflow_module, "current_account_with_tenant", lambda: (SimpleNamespace(), "tenant-1"))
|
||||
accessible_app_ids = Mock(return_value=set())
|
||||
monkeypatch.setattr(
|
||||
workflow_module,
|
||||
"WorkflowService",
|
||||
lambda: SimpleNamespace(get_accessible_app_ids=accessible_app_ids),
|
||||
)
|
||||
|
||||
excessive_ids = ",".join(f"wf-{index}" for index in range(workflow_module.MAX_WORKFLOW_ONLINE_USERS_QUERY_IDS + 1))
|
||||
|
||||
api = workflow_module.WorkflowOnlineUsersApi()
|
||||
handler = _unwrap(api.get)
|
||||
|
||||
with app.test_request_context(
|
||||
f"/apps/workflows/online-users?app_ids={excessive_ids}",
|
||||
method="GET",
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
handler(api)
|
||||
|
||||
assert exc.value.code == 400
|
||||
assert "Maximum" in exc.value.description
|
||||
accessible_app_ids.assert_not_called()
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import nullcontext
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, PropertyMock, patch
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from werkzeug.exceptions import Forbidden
|
||||
|
||||
from controllers.console import console_ns
|
||||
from controllers.console import wraps as console_wraps
|
||||
from controllers.console.app import workflow_comment as workflow_comment_module
|
||||
from controllers.console.app import wraps as app_wraps
|
||||
from libs import login as login_lib
|
||||
from models.account import Account, AccountStatus, TenantAccountRole
|
||||
|
||||
|
||||
def _make_account(role: TenantAccountRole) -> Account:
|
||||
account = Account(name="tester", email="tester@example.com")
|
||||
account.status = AccountStatus.ACTIVE
|
||||
account.role = role
|
||||
account.id = "account-123" # type: ignore[assignment]
|
||||
account._current_tenant = SimpleNamespace(id="tenant-123") # type: ignore[attr-defined]
|
||||
account._get_current_object = lambda: account # type: ignore[attr-defined]
|
||||
return account
|
||||
|
||||
|
||||
def _make_app() -> SimpleNamespace:
|
||||
return SimpleNamespace(id="app-123", tenant_id="tenant-123", status="normal", mode="workflow")
|
||||
|
||||
|
||||
def _patch_console_guards(monkeypatch: pytest.MonkeyPatch, account: Account, app_model: SimpleNamespace) -> None:
|
||||
monkeypatch.setattr(login_lib.dify_config, "LOGIN_DISABLED", True)
|
||||
monkeypatch.setattr(login_lib, "current_user", account)
|
||||
monkeypatch.setattr(login_lib, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
|
||||
monkeypatch.setattr(login_lib, "check_csrf_token", lambda *_, **__: None)
|
||||
monkeypatch.setattr(console_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
|
||||
monkeypatch.setattr(console_wraps.dify_config, "EDITION", "CLOUD")
|
||||
monkeypatch.setattr(app_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
|
||||
monkeypatch.setattr(app_wraps, "_load_app_model", lambda _app_id: app_model)
|
||||
monkeypatch.setattr(workflow_comment_module, "current_user", account)
|
||||
|
||||
|
||||
def _patch_write_services(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
for method_name in (
|
||||
"create_comment",
|
||||
"update_comment",
|
||||
"delete_comment",
|
||||
"resolve_comment",
|
||||
"validate_comment_access",
|
||||
"create_reply",
|
||||
"update_reply",
|
||||
"delete_reply",
|
||||
):
|
||||
monkeypatch.setattr(workflow_comment_module.WorkflowCommentService, method_name, MagicMock())
|
||||
|
||||
|
||||
def _patch_payload(payload: dict[str, object] | None):
|
||||
if payload is None:
|
||||
return nullcontext()
|
||||
return patch.object(
|
||||
type(console_ns),
|
||||
"payload",
|
||||
new_callable=PropertyMock,
|
||||
return_value=payload,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class WriteCase:
|
||||
resource_cls: type
|
||||
method_name: str
|
||||
path: str
|
||||
kwargs: dict[str, str]
|
||||
payload: dict[str, object] | None = None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"case",
|
||||
[
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentListApi,
|
||||
method_name="post",
|
||||
path="/console/api/apps/app-123/workflow/comments",
|
||||
kwargs={"app_id": "app-123"},
|
||||
payload={"content": "hello", "position_x": 1.0, "position_y": 2.0, "mentioned_user_ids": []},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentDetailApi,
|
||||
method_name="put",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1"},
|
||||
payload={"content": "hello", "position_x": 1.0, "position_y": 2.0, "mentioned_user_ids": []},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentDetailApi,
|
||||
method_name="delete",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1"},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentResolveApi,
|
||||
method_name="post",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1/resolve",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1"},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentReplyApi,
|
||||
method_name="post",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1/replies",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1"},
|
||||
payload={"content": "reply", "mentioned_user_ids": []},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentReplyDetailApi,
|
||||
method_name="put",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1/replies/reply-1",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1", "reply_id": "reply-1"},
|
||||
payload={"content": "reply", "mentioned_user_ids": []},
|
||||
),
|
||||
WriteCase(
|
||||
resource_cls=workflow_comment_module.WorkflowCommentReplyDetailApi,
|
||||
method_name="delete",
|
||||
path="/console/api/apps/app-123/workflow/comments/comment-1/replies/reply-1",
|
||||
kwargs={"app_id": "app-123", "comment_id": "comment-1", "reply_id": "reply-1"},
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_write_endpoints_require_edit_permission(app: Flask, monkeypatch: pytest.MonkeyPatch, case: WriteCase) -> None:
|
||||
app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
|
||||
account = _make_account(TenantAccountRole.NORMAL)
|
||||
app_model = _make_app()
|
||||
_patch_console_guards(monkeypatch, account, app_model)
|
||||
_patch_write_services(monkeypatch)
|
||||
|
||||
with app.test_request_context(case.path, method=case.method_name.upper(), json=case.payload):
|
||||
with _patch_payload(case.payload):
|
||||
handler = getattr(case.resource_cls(), case.method_name)
|
||||
with pytest.raises(Forbidden):
|
||||
handler(**case.kwargs)
|
||||
|
||||
|
||||
def test_create_comment_allows_editor(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
|
||||
account = _make_account(TenantAccountRole.EDITOR)
|
||||
app_model = _make_app()
|
||||
_patch_console_guards(monkeypatch, account, app_model)
|
||||
|
||||
create_comment_mock = MagicMock(return_value={"id": "comment-1"})
|
||||
monkeypatch.setattr(workflow_comment_module.WorkflowCommentService, "create_comment", create_comment_mock)
|
||||
payload = {"content": "hello", "position_x": 1.0, "position_y": 2.0, "mentioned_user_ids": []}
|
||||
|
||||
with app.test_request_context("/console/api/apps/app-123/workflow/comments", method="POST", json=payload):
|
||||
with _patch_payload(payload):
|
||||
result = workflow_comment_module.WorkflowCommentListApi().post(app_id="app-123")
|
||||
|
||||
if isinstance(result, tuple):
|
||||
response = result[0]
|
||||
else:
|
||||
response = result
|
||||
assert response["id"] == "comment-1"
|
||||
create_comment_mock.assert_called_once_with(
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
created_by="account-123",
|
||||
content="hello",
|
||||
position_x=1.0,
|
||||
position_y=2.0,
|
||||
mentioned_user_ids=[],
|
||||
)
|
||||
|
||||
|
||||
def test_update_comment_omits_mentions_when_payload_does_not_include_them(
|
||||
app: Flask, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
|
||||
account = _make_account(TenantAccountRole.EDITOR)
|
||||
app_model = _make_app()
|
||||
_patch_console_guards(monkeypatch, account, app_model)
|
||||
|
||||
update_comment_mock = MagicMock(return_value={"id": "comment-1", "updated_at": datetime(2024, 1, 1, 12, 0, 0)})
|
||||
monkeypatch.setattr(workflow_comment_module.WorkflowCommentService, "update_comment", update_comment_mock)
|
||||
payload = {"content": "hello", "position_x": 10.0, "position_y": 20.0}
|
||||
|
||||
with app.test_request_context("/console/api/apps/app-123/workflow/comments/comment-1", method="PUT", json=payload):
|
||||
with _patch_payload(payload):
|
||||
workflow_comment_module.WorkflowCommentDetailApi().put(app_id="app-123", comment_id="comment-1")
|
||||
|
||||
update_comment_mock.assert_called_once_with(
|
||||
tenant_id="tenant-123",
|
||||
app_id="app-123",
|
||||
comment_id="comment-1",
|
||||
user_id="account-123",
|
||||
content="hello",
|
||||
position_x=10.0,
|
||||
position_y=20.0,
|
||||
mentioned_user_ids=None,
|
||||
)
|
||||
@@ -503,6 +503,7 @@ class TestEmailI18nIntegration:
|
||||
EmailType.ACCOUNT_DELETION_VERIFICATION,
|
||||
EmailType.QUEUE_MONITOR_ALERT,
|
||||
EmailType.DOCUMENT_CLEAN_NOTIFY,
|
||||
EmailType.WORKFLOW_COMMENT_MENTION,
|
||||
]
|
||||
|
||||
for email_type in expected_types:
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from models.comment import WorkflowComment, WorkflowCommentMention, WorkflowCommentReply
|
||||
|
||||
|
||||
def test_workflow_comment_account_properties_and_cache() -> None:
|
||||
comment = WorkflowComment(created_by="user-1", resolved_by="user-2", content="hello", position_x=1, position_y=2)
|
||||
created_account = Mock(id="user-1")
|
||||
resolved_account = Mock(id="user-2")
|
||||
|
||||
with patch("models.comment.db.session.get", side_effect=[created_account, resolved_account]) as get_mock:
|
||||
assert comment.created_by_account is created_account
|
||||
assert comment.resolved_by_account is resolved_account
|
||||
assert get_mock.call_count == 2
|
||||
|
||||
comment.cache_created_by_account(created_account)
|
||||
comment.cache_resolved_by_account(resolved_account)
|
||||
with patch("models.comment.db.session.get") as get_mock:
|
||||
assert comment.created_by_account is created_account
|
||||
assert comment.resolved_by_account is resolved_account
|
||||
get_mock.assert_not_called()
|
||||
|
||||
comment_without_resolver = WorkflowComment(
|
||||
created_by="user-1",
|
||||
resolved_by=None,
|
||||
content="hello",
|
||||
position_x=1,
|
||||
position_y=2,
|
||||
)
|
||||
with patch("models.comment.db.session.get") as get_mock:
|
||||
assert comment_without_resolver.resolved_by_account is None
|
||||
get_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_workflow_comment_counts_and_participants() -> None:
|
||||
reply_1 = WorkflowCommentReply(comment_id="comment-1", content="reply-1", created_by="user-2")
|
||||
reply_2 = WorkflowCommentReply(comment_id="comment-1", content="reply-2", created_by="user-2")
|
||||
mention_1 = WorkflowCommentMention(comment_id="comment-1", mentioned_user_id="user-3")
|
||||
mention_2 = WorkflowCommentMention(comment_id="comment-1", mentioned_user_id="user-4")
|
||||
comment = WorkflowComment(created_by="user-1", resolved_by=None, content="hello", position_x=1, position_y=2)
|
||||
comment.replies = [reply_1, reply_2]
|
||||
comment.mentions = [mention_1, mention_2]
|
||||
|
||||
account_1 = Mock(id="user-1")
|
||||
account_2 = Mock(id="user-2")
|
||||
account_3 = Mock(id="user-3")
|
||||
account_map = {
|
||||
"user-1": account_1,
|
||||
"user-2": account_2,
|
||||
"user-3": account_3,
|
||||
"user-4": None,
|
||||
}
|
||||
|
||||
with patch("models.comment.db.session.get", side_effect=lambda _model, user_id: account_map[user_id]) as get_mock:
|
||||
participants = comment.participants
|
||||
|
||||
assert comment.reply_count == 2
|
||||
assert comment.mention_count == 2
|
||||
assert set(participants) == {account_1, account_2, account_3}
|
||||
assert get_mock.call_count == 4
|
||||
|
||||
|
||||
def test_workflow_comment_participants_use_cached_accounts() -> None:
|
||||
reply = WorkflowCommentReply(comment_id="comment-1", content="reply-1", created_by="user-2")
|
||||
mention = WorkflowCommentMention(comment_id="comment-1", mentioned_user_id="user-3")
|
||||
comment = WorkflowComment(created_by="user-1", resolved_by=None, content="hello", position_x=1, position_y=2)
|
||||
comment.replies = [reply]
|
||||
comment.mentions = [mention]
|
||||
|
||||
account_1 = Mock(id="user-1")
|
||||
account_2 = Mock(id="user-2")
|
||||
account_3 = Mock(id="user-3")
|
||||
comment.cache_created_by_account(account_1)
|
||||
reply.cache_created_by_account(account_2)
|
||||
mention.cache_mentioned_user_account(account_3)
|
||||
|
||||
with patch("models.comment.db.session.get") as get_mock:
|
||||
participants = comment.participants
|
||||
|
||||
assert set(participants) == {account_1, account_2, account_3}
|
||||
get_mock.assert_not_called()
|
||||
|
||||
|
||||
def test_reply_and_mention_account_properties_and_cache() -> None:
|
||||
reply = WorkflowCommentReply(comment_id="comment-1", content="reply", created_by="user-1")
|
||||
mention = WorkflowCommentMention(comment_id="comment-1", mentioned_user_id="user-2")
|
||||
reply_account = Mock(id="user-1")
|
||||
mention_account = Mock(id="user-2")
|
||||
|
||||
with patch("models.comment.db.session.get", side_effect=[reply_account, mention_account]) as get_mock:
|
||||
assert reply.created_by_account is reply_account
|
||||
assert mention.mentioned_user_account is mention_account
|
||||
assert get_mock.call_count == 2
|
||||
|
||||
reply.cache_created_by_account(reply_account)
|
||||
mention.cache_mentioned_user_account(mention_account)
|
||||
with patch("models.comment.db.session.get") as get_mock:
|
||||
assert reply.created_by_account is reply_account
|
||||
assert mention.mentioned_user_account is mention_account
|
||||
get_mock.assert_not_called()
|
||||
@@ -0,0 +1,121 @@
|
||||
import json
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from repositories import workflow_collaboration_repository as repo_module
|
||||
from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
|
||||
|
||||
|
||||
class TestWorkflowCollaborationRepository:
|
||||
@pytest.fixture
|
||||
def mock_redis(self, monkeypatch: pytest.MonkeyPatch) -> Mock:
|
||||
mock_redis = Mock()
|
||||
monkeypatch.setattr(repo_module, "redis_client", mock_redis)
|
||||
return mock_redis
|
||||
|
||||
def test_get_sid_mapping_returns_mapping(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.get.return_value = b'{"workflow_id":"wf-1","user_id":"u-1"}'
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
result = repository.get_sid_mapping("sid-1")
|
||||
|
||||
# Assert
|
||||
assert result == {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
|
||||
def test_list_sessions_filters_invalid_entries(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.hgetall.return_value = {
|
||||
b"sid-1": b'{"user_id":"u-1","username":"Jane","sid":"sid-1","connected_at":2}',
|
||||
b"sid-2": b'{"username":"Missing","sid":"sid-2"}',
|
||||
b"sid-3": b"not-json",
|
||||
}
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
result = repository.list_sessions("wf-1")
|
||||
|
||||
# Assert
|
||||
assert result == [
|
||||
{
|
||||
"user_id": "u-1",
|
||||
"username": "Jane",
|
||||
"avatar": None,
|
||||
"sid": "sid-1",
|
||||
"connected_at": 2,
|
||||
}
|
||||
]
|
||||
|
||||
def test_set_session_info_persists_payload(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.exists.return_value = True
|
||||
repository = WorkflowCollaborationRepository()
|
||||
payload = {
|
||||
"user_id": "u-1",
|
||||
"username": "Jane",
|
||||
"avatar": None,
|
||||
"sid": "sid-1",
|
||||
"connected_at": 1,
|
||||
}
|
||||
|
||||
# Act
|
||||
repository.set_session_info("wf-1", payload)
|
||||
|
||||
# Assert
|
||||
assert mock_redis.hset.called
|
||||
workflow_key, sid, session_json = mock_redis.hset.call_args.args
|
||||
assert workflow_key == "workflow_online_users:wf-1"
|
||||
assert sid == "sid-1"
|
||||
assert json.loads(session_json)["user_id"] == "u-1"
|
||||
assert mock_redis.set.called
|
||||
|
||||
def test_refresh_session_state_expires_keys(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.exists.return_value = True
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
repository.refresh_session_state("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
assert mock_redis.expire.call_count == 2
|
||||
|
||||
def test_get_current_leader_decodes_bytes(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.get.return_value = b"sid-1"
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
result = repository.get_current_leader("wf-1")
|
||||
|
||||
# Assert
|
||||
assert result == "sid-1"
|
||||
|
||||
def test_set_leader_if_absent_uses_nx(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.set.return_value = True
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
result = repository.set_leader_if_absent("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
assert result is True
|
||||
_key, _value = mock_redis.set.call_args.args
|
||||
assert _key == "workflow_leader:wf-1"
|
||||
assert _value == "sid-1"
|
||||
assert mock_redis.set.call_args.kwargs["nx"] is True
|
||||
assert "ex" in mock_redis.set.call_args.kwargs
|
||||
|
||||
def test_get_session_sids_decodes(self, mock_redis: Mock) -> None:
|
||||
# Arrange
|
||||
mock_redis.hkeys.return_value = [b"sid-1", "sid-2"]
|
||||
repository = WorkflowCollaborationRepository()
|
||||
|
||||
# Act
|
||||
result = repository.get_session_sids("wf-1")
|
||||
|
||||
# Assert
|
||||
assert result == ["sid-1", "sid-2"]
|
||||
@@ -0,0 +1,608 @@
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository
|
||||
from services.workflow_collaboration_service import WorkflowCollaborationService
|
||||
|
||||
|
||||
class TestWorkflowCollaborationService:
|
||||
@pytest.fixture
|
||||
def service(self) -> tuple[WorkflowCollaborationService, Mock, Mock]:
|
||||
repository = Mock(spec=WorkflowCollaborationRepository)
|
||||
socketio = Mock()
|
||||
return WorkflowCollaborationService(repository, socketio), repository, socketio
|
||||
|
||||
def test_authorize_and_join_workflow_room_returns_leader_status(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, socketio = service
|
||||
socketio.get_session.return_value = {
|
||||
"user_id": "u-1",
|
||||
"username": "Jane",
|
||||
"avatar": None,
|
||||
"tenant_id": "t-1",
|
||||
}
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "_can_access_workflow", return_value=True),
|
||||
patch.object(collaboration_service, "get_or_set_leader", return_value="sid-1"),
|
||||
patch.object(collaboration_service, "broadcast_online_users"),
|
||||
):
|
||||
# Act
|
||||
result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
assert result == ("u-1", True)
|
||||
repository.set_session_info.assert_called_once()
|
||||
socketio.enter_room.assert_called_once_with("sid-1", "wf-1")
|
||||
socketio.emit.assert_called_once_with("status", {"isLeader": True}, room="sid-1")
|
||||
|
||||
def test_authorize_and_join_workflow_room_returns_none_when_missing_user(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, _repository, socketio = service
|
||||
socketio.get_session.return_value = {}
|
||||
|
||||
# Act
|
||||
result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
assert result is None
|
||||
|
||||
def test_authorize_and_join_workflow_room_returns_none_when_missing_tenant(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
socketio.get_session.return_value = {"user_id": "u-1", "username": "Jane", "avatar": None}
|
||||
|
||||
result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
|
||||
|
||||
assert result is None
|
||||
repository.set_session_info.assert_not_called()
|
||||
socketio.enter_room.assert_not_called()
|
||||
socketio.emit.assert_not_called()
|
||||
|
||||
def test_authorize_and_join_workflow_room_returns_none_when_workflow_is_not_accessible(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
socketio.get_session.return_value = {
|
||||
"user_id": "u-1",
|
||||
"username": "Jane",
|
||||
"avatar": None,
|
||||
"tenant_id": "t-1",
|
||||
}
|
||||
|
||||
with patch.object(collaboration_service, "_can_access_workflow", return_value=False):
|
||||
result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
|
||||
|
||||
assert result is None
|
||||
repository.set_session_info.assert_not_called()
|
||||
socketio.enter_room.assert_not_called()
|
||||
socketio.emit.assert_not_called()
|
||||
|
||||
def test_repr_and_save_socket_identity(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
collaboration_service, _repository, socketio = service
|
||||
user = Mock()
|
||||
user.id = "u-1"
|
||||
user.name = "Jane"
|
||||
user.avatar = "avatar.png"
|
||||
user.current_tenant_id = "t-1"
|
||||
|
||||
assert "WorkflowCollaborationService" in repr(collaboration_service)
|
||||
|
||||
collaboration_service.save_socket_identity("sid-1", user)
|
||||
|
||||
socketio.save_session.assert_called_once_with(
|
||||
"sid-1",
|
||||
{"user_id": "u-1", "username": "Jane", "avatar": "avatar.png", "tenant_id": "t-1"},
|
||||
)
|
||||
|
||||
def test_can_access_workflow_uses_session_factory(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, _repository, _socketio = service
|
||||
session = Mock()
|
||||
session.scalar.return_value = "wf-1"
|
||||
session_context = Mock()
|
||||
session_context.__enter__ = Mock(return_value=session)
|
||||
session_context.__exit__ = Mock(return_value=False)
|
||||
|
||||
with patch(
|
||||
"services.workflow_collaboration_service.session_factory.create_session",
|
||||
return_value=session_context,
|
||||
):
|
||||
result = collaboration_service._can_access_workflow("wf-1", "tenant-1")
|
||||
|
||||
assert result is True
|
||||
session.scalar.assert_called_once()
|
||||
|
||||
def test_relay_collaboration_event_unauthorized(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_sid_mapping.return_value = None
|
||||
|
||||
# Act
|
||||
result = collaboration_service.relay_collaboration_event("sid-1", {})
|
||||
|
||||
# Assert
|
||||
assert result == ({"msg": "unauthorized"}, 401)
|
||||
|
||||
def test_relay_collaboration_event_emits_update(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
payload = {"type": "mouse_move", "data": {"x": 1}, "timestamp": 123}
|
||||
|
||||
# Act
|
||||
result = collaboration_service.relay_collaboration_event("sid-1", payload)
|
||||
|
||||
# Assert
|
||||
assert result == ({"msg": "event_broadcasted"}, 200)
|
||||
socketio.emit.assert_called_once_with(
|
||||
"collaboration_update",
|
||||
{"type": "mouse_move", "userId": "u-1", "data": {"x": 1}, "timestamp": 123},
|
||||
room="wf-1",
|
||||
skip_sid="sid-1",
|
||||
)
|
||||
|
||||
def test_relay_collaboration_event_requires_event_type(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
|
||||
result = collaboration_service.relay_collaboration_event("sid-1", {"data": {"x": 1}})
|
||||
|
||||
assert result == ({"msg": "invalid event type"}, 400)
|
||||
|
||||
def test_relay_collaboration_event_sync_request_forwards_to_active_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
repository.get_current_leader.return_value = "sid-leader"
|
||||
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "refresh_session_state"),
|
||||
patch.object(collaboration_service, "is_session_active", return_value=True),
|
||||
):
|
||||
result = collaboration_service.relay_collaboration_event("sid-1", payload)
|
||||
|
||||
assert result == ({"msg": "sync_request_forwarded"}, 200)
|
||||
socketio.emit.assert_called_once_with(
|
||||
"collaboration_update",
|
||||
{"type": "sync_request", "userId": "u-1", "data": {"reason": "join"}, "timestamp": 123},
|
||||
room="sid-leader",
|
||||
)
|
||||
repository.set_leader.assert_not_called()
|
||||
|
||||
def test_relay_collaboration_event_sync_request_reelects_active_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
repository.get_current_leader.return_value = "sid-old"
|
||||
repository.list_sessions.return_value = [
|
||||
{
|
||||
"user_id": "u-2",
|
||||
"username": "B",
|
||||
"avatar": None,
|
||||
"sid": "sid-2",
|
||||
"connected_at": 1,
|
||||
"graph_active": True,
|
||||
},
|
||||
{
|
||||
"user_id": "u-3",
|
||||
"username": "C",
|
||||
"avatar": None,
|
||||
"sid": "sid-3",
|
||||
"connected_at": 2,
|
||||
"graph_active": True,
|
||||
},
|
||||
]
|
||||
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
|
||||
|
||||
def _is_session_active(_workflow_id: str, session_sid: str) -> bool:
|
||||
return session_sid != "sid-old"
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "refresh_session_state"),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
patch.object(collaboration_service, "is_session_active", side_effect=_is_session_active),
|
||||
):
|
||||
result = collaboration_service.relay_collaboration_event("sid-2", payload)
|
||||
|
||||
assert result == ({"msg": "sync_request_forwarded"}, 200)
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
|
||||
socketio.emit.assert_called_once_with(
|
||||
"collaboration_update",
|
||||
{"type": "sync_request", "userId": "u-1", "data": {"reason": "join"}, "timestamp": 123},
|
||||
room="sid-2",
|
||||
)
|
||||
|
||||
def test_relay_collaboration_event_sync_request_returns_when_no_active_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
repository.get_current_leader.return_value = "sid-old"
|
||||
repository.list_sessions.return_value = []
|
||||
payload = {"type": "sync_request", "data": {"reason": "join"}, "timestamp": 123}
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "refresh_session_state"),
|
||||
patch.object(collaboration_service, "is_session_active", return_value=False),
|
||||
):
|
||||
result = collaboration_service.relay_collaboration_event("sid-2", payload)
|
||||
|
||||
assert result == ({"msg": "no_active_leader"}, 200)
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
socketio.emit.assert_not_called()
|
||||
|
||||
def test_relay_graph_event_unauthorized(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_sid_mapping.return_value = None
|
||||
|
||||
# Act
|
||||
result = collaboration_service.relay_graph_event("sid-1", {"nodes": []})
|
||||
|
||||
# Assert
|
||||
assert result == ({"msg": "unauthorized"}, 401)
|
||||
|
||||
def test_disconnect_session_no_mapping(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_sid_mapping.return_value = None
|
||||
|
||||
# Act
|
||||
collaboration_service.disconnect_session("sid-1")
|
||||
|
||||
# Assert
|
||||
repository.delete_session.assert_not_called()
|
||||
|
||||
def test_disconnect_session_cleans_up(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "handle_leader_disconnect") as handle_leader_disconnect,
|
||||
patch.object(collaboration_service, "broadcast_online_users") as broadcast_online_users,
|
||||
):
|
||||
# Act
|
||||
collaboration_service.disconnect_session("sid-1")
|
||||
|
||||
# Assert
|
||||
repository.delete_session.assert_called_once_with("wf-1", "sid-1")
|
||||
handle_leader_disconnect.assert_called_once_with("wf-1", "sid-1")
|
||||
broadcast_online_users.assert_called_once_with("wf-1")
|
||||
|
||||
def test_get_or_set_leader_returns_active_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
|
||||
with patch.object(collaboration_service, "is_session_active", return_value=True):
|
||||
# Act
|
||||
result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
|
||||
|
||||
# Assert
|
||||
assert result == "sid-1"
|
||||
repository.set_leader_if_absent.assert_not_called()
|
||||
|
||||
def test_get_or_set_leader_replaces_dead_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
repository.set_leader_if_absent.return_value = True
|
||||
repository.list_sessions.return_value = [
|
||||
{
|
||||
"user_id": "u-2",
|
||||
"username": "B",
|
||||
"avatar": None,
|
||||
"sid": "sid-2",
|
||||
"connected_at": 1,
|
||||
"graph_active": True,
|
||||
}
|
||||
]
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "is_session_active", side_effect=lambda _wf, sid: sid != "sid-1"),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
):
|
||||
# Act
|
||||
result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
|
||||
|
||||
# Assert
|
||||
assert result == "sid-2"
|
||||
repository.delete_session.assert_called_once_with("wf-1", "sid-1")
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
|
||||
|
||||
def test_get_or_set_leader_falls_back_to_existing(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.side_effect = [None, "sid-3"]
|
||||
repository.set_leader_if_absent.return_value = False
|
||||
repository.list_sessions.return_value = [
|
||||
{
|
||||
"user_id": "u-2",
|
||||
"username": "B",
|
||||
"avatar": None,
|
||||
"sid": "sid-2",
|
||||
"connected_at": 1,
|
||||
"graph_active": True,
|
||||
}
|
||||
]
|
||||
|
||||
# Act
|
||||
result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
|
||||
|
||||
# Assert
|
||||
assert result == "sid-3"
|
||||
|
||||
def test_get_or_set_leader_returns_sid_when_leader_still_missing(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.side_effect = [None, None]
|
||||
repository.set_leader_if_absent.return_value = False
|
||||
|
||||
result = collaboration_service.get_or_set_leader("wf-1", "sid-2")
|
||||
|
||||
assert result == "sid-2"
|
||||
|
||||
def test_handle_leader_disconnect_elects_new(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
repository.list_sessions.return_value = [
|
||||
{
|
||||
"user_id": "u-2",
|
||||
"username": "B",
|
||||
"avatar": None,
|
||||
"sid": "sid-2",
|
||||
"connected_at": 1,
|
||||
"graph_active": True,
|
||||
}
|
||||
]
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "is_session_active", return_value=True),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
):
|
||||
# Act
|
||||
collaboration_service.handle_leader_disconnect("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
|
||||
|
||||
def test_handle_leader_disconnect_clears_when_empty(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
repository.list_sessions.return_value = []
|
||||
|
||||
# Act
|
||||
collaboration_service.handle_leader_disconnect("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
|
||||
def test_handle_leader_disconnect_ignores_non_leader_or_missing_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, _socketio = service
|
||||
|
||||
repository.get_current_leader.return_value = None
|
||||
collaboration_service.handle_leader_disconnect("wf-1", "sid-1")
|
||||
|
||||
repository.get_current_leader.return_value = "sid-leader"
|
||||
collaboration_service.handle_leader_disconnect("wf-1", "sid-other")
|
||||
|
||||
repository.set_leader.assert_not_called()
|
||||
repository.delete_leader.assert_not_called()
|
||||
|
||||
def test_broadcast_leader_change_logs_emit_errors(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_session_sids.return_value = ["sid-1", "sid-2"]
|
||||
socketio.emit.side_effect = [RuntimeError("boom"), None]
|
||||
|
||||
with patch("services.workflow_collaboration_service.logging.exception") as exception_mock:
|
||||
collaboration_service.broadcast_leader_change("wf-1", "sid-2")
|
||||
|
||||
assert exception_mock.call_count == 1
|
||||
|
||||
def test_broadcast_online_users_sorts_and_emits(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.list_sessions.return_value = [
|
||||
{"user_id": "u-1", "username": "A", "avatar": None, "sid": "sid-1", "connected_at": 3},
|
||||
{"user_id": "u-2", "username": "B", "avatar": None, "sid": "sid-2", "connected_at": 1},
|
||||
]
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
|
||||
with patch.object(collaboration_service, "is_session_active", return_value=True):
|
||||
# Act
|
||||
collaboration_service.broadcast_online_users("wf-1")
|
||||
|
||||
# Assert
|
||||
socketio.emit.assert_called_once_with(
|
||||
"online_users",
|
||||
{
|
||||
"workflow_id": "wf-1",
|
||||
"users": [
|
||||
{"user_id": "u-2", "username": "B", "avatar": None, "sid": "sid-2", "connected_at": 1},
|
||||
{"user_id": "u-1", "username": "A", "avatar": None, "sid": "sid-1", "connected_at": 3},
|
||||
],
|
||||
"leader": "sid-1",
|
||||
},
|
||||
room="wf-1",
|
||||
)
|
||||
|
||||
def test_broadcast_online_users_reassigns_missing_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
users = [{"user_id": "u-2", "username": "B", "avatar": None, "sid": "sid-2", "connected_at": 1}]
|
||||
repository.get_current_leader.return_value = "sid-old"
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "_prune_inactive_sessions", return_value=users),
|
||||
patch.object(collaboration_service, "_select_graph_leader", return_value="sid-2"),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
):
|
||||
collaboration_service.broadcast_online_users("wf-1")
|
||||
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
|
||||
socketio.emit.assert_called_once_with(
|
||||
"online_users",
|
||||
{"workflow_id": "wf-1", "users": users, "leader": "sid-2"},
|
||||
room="wf-1",
|
||||
)
|
||||
|
||||
def test_refresh_session_state_expires_active_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-1"
|
||||
|
||||
with patch.object(collaboration_service, "is_session_active", return_value=True):
|
||||
# Act
|
||||
collaboration_service.refresh_session_state("wf-1", "sid-1")
|
||||
|
||||
# Assert
|
||||
repository.refresh_session_state.assert_called_once_with("wf-1", "sid-1")
|
||||
repository.expire_leader.assert_called_once_with("wf-1")
|
||||
repository.set_leader.assert_not_called()
|
||||
|
||||
def test_refresh_session_state_sets_leader_when_missing(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = None
|
||||
repository.list_sessions.return_value = [
|
||||
{
|
||||
"user_id": "u-2",
|
||||
"username": "B",
|
||||
"avatar": None,
|
||||
"sid": "sid-2",
|
||||
"connected_at": 1,
|
||||
"graph_active": True,
|
||||
}
|
||||
]
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "is_session_active", return_value=True),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
):
|
||||
# Act
|
||||
collaboration_service.refresh_session_state("wf-1", "sid-2")
|
||||
|
||||
# Assert
|
||||
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
|
||||
|
||||
def test_refresh_session_state_replaces_inactive_existing_leader(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.get_current_leader.return_value = "sid-old"
|
||||
|
||||
with (
|
||||
patch.object(collaboration_service, "is_session_active", return_value=False),
|
||||
patch.object(collaboration_service, "broadcast_leader_change") as broadcast_leader_change,
|
||||
):
|
||||
collaboration_service.refresh_session_state("wf-1", "sid-new")
|
||||
|
||||
repository.delete_leader.assert_called_once_with("wf-1")
|
||||
repository.set_leader.assert_called_once_with("wf-1", "sid-new")
|
||||
broadcast_leader_change.assert_called_once_with("wf-1", "sid-new")
|
||||
|
||||
def test_relay_graph_event_emits_update(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
# Arrange
|
||||
collaboration_service, repository, socketio = service
|
||||
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
|
||||
|
||||
# Act
|
||||
result = collaboration_service.relay_graph_event("sid-1", {"nodes": []})
|
||||
|
||||
# Assert
|
||||
assert result == ({"msg": "graph_update_broadcasted"}, 200)
|
||||
repository.refresh_session_state.assert_called_once_with("wf-1", "sid-1")
|
||||
socketio.emit.assert_called_once_with("graph_update", {"nodes": []}, room="wf-1", skip_sid="sid-1")
|
||||
|
||||
def test_prune_inactive_sessions_handles_empty_and_removes_stale(
|
||||
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
|
||||
) -> None:
|
||||
collaboration_service, repository, _socketio = service
|
||||
repository.list_sessions.return_value = []
|
||||
assert collaboration_service._prune_inactive_sessions("wf-1") == []
|
||||
|
||||
active = {"sid": "sid-1", "user_id": "u-1", "connected_at": 1}
|
||||
stale = {"sid": "sid-2", "user_id": "u-2", "connected_at": 2}
|
||||
repository.list_sessions.return_value = [active, stale]
|
||||
|
||||
with patch.object(
|
||||
collaboration_service,
|
||||
"is_session_active",
|
||||
side_effect=lambda _workflow_id, sid: sid == "sid-1",
|
||||
):
|
||||
users = collaboration_service._prune_inactive_sessions("wf-1")
|
||||
|
||||
assert users == [active]
|
||||
repository.delete_session.assert_called_with("wf-1", "sid-2")
|
||||
|
||||
def test_is_session_active_guard_branches(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
|
||||
collaboration_service, repository, socketio = service
|
||||
socketio.manager.is_connected.return_value = True
|
||||
repository.session_exists.return_value = True
|
||||
repository.sid_mapping_exists.return_value = True
|
||||
|
||||
assert collaboration_service.is_session_active("wf-1", "") is False
|
||||
|
||||
socketio.manager.is_connected.return_value = False
|
||||
assert collaboration_service.is_session_active("wf-1", "sid-1") is False
|
||||
|
||||
socketio.manager.is_connected.side_effect = AttributeError("missing manager")
|
||||
assert collaboration_service.is_session_active("wf-1", "sid-1") is False
|
||||
socketio.manager.is_connected.side_effect = None
|
||||
|
||||
socketio.manager.is_connected.return_value = True
|
||||
repository.session_exists.return_value = False
|
||||
assert collaboration_service.is_session_active("wf-1", "sid-1") is False
|
||||
|
||||
repository.session_exists.return_value = True
|
||||
repository.sid_mapping_exists.return_value = False
|
||||
assert collaboration_service.is_session_active("wf-1", "sid-1") is False
|
||||
@@ -0,0 +1,578 @@
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from werkzeug.exceptions import Forbidden, NotFound
|
||||
|
||||
from services import workflow_comment_service as service_module
|
||||
from services.workflow_comment_service import WorkflowCommentService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_session(monkeypatch: pytest.MonkeyPatch) -> Mock:
|
||||
session = Mock()
|
||||
context_manager = MagicMock()
|
||||
context_manager.__enter__.return_value = session
|
||||
context_manager.__exit__.return_value = False
|
||||
mock_db = MagicMock()
|
||||
mock_db.engine = Mock()
|
||||
empty_scalars = Mock()
|
||||
empty_scalars.all.return_value = []
|
||||
session.scalars.return_value = empty_scalars
|
||||
monkeypatch.setattr(service_module, "Session", Mock(return_value=context_manager))
|
||||
monkeypatch.setattr(service_module, "db", mock_db)
|
||||
monkeypatch.setattr(service_module.send_workflow_comment_mention_email_task, "delay", Mock())
|
||||
return session
|
||||
|
||||
|
||||
def _mock_scalars(result_list: list[object]) -> Mock:
|
||||
scalars = Mock()
|
||||
scalars.all.return_value = result_list
|
||||
return scalars
|
||||
|
||||
|
||||
class TestWorkflowCommentService:
|
||||
def test_validate_content_rejects_empty(self) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
WorkflowCommentService._validate_content(" ")
|
||||
|
||||
def test_validate_content_rejects_too_long(self) -> None:
|
||||
with pytest.raises(ValueError):
|
||||
WorkflowCommentService._validate_content("a" * 1001)
|
||||
|
||||
def test_filter_valid_mentioned_user_ids_filters_by_tenant_and_preserves_order(self, mock_session: Mock) -> None:
|
||||
tenant_member_1 = "123e4567-e89b-12d3-a456-426614174000"
|
||||
tenant_member_2 = "123e4567-e89b-12d3-a456-426614174002"
|
||||
non_tenant_member = "123e4567-e89b-12d3-a456-426614174001"
|
||||
mock_session.scalars.return_value = _mock_scalars([tenant_member_1, tenant_member_2])
|
||||
|
||||
result = WorkflowCommentService._filter_valid_mentioned_user_ids(
|
||||
[
|
||||
tenant_member_1,
|
||||
"",
|
||||
123, # type: ignore[list-item]
|
||||
tenant_member_1,
|
||||
non_tenant_member,
|
||||
tenant_member_2,
|
||||
],
|
||||
session=mock_session,
|
||||
tenant_id="tenant-1",
|
||||
)
|
||||
|
||||
assert result == [
|
||||
tenant_member_1,
|
||||
tenant_member_2,
|
||||
]
|
||||
|
||||
def test_format_comment_excerpt_handles_short_and_long_limits(self) -> None:
|
||||
assert WorkflowCommentService._format_comment_excerpt(" hello ", max_length=10) == "hello"
|
||||
assert WorkflowCommentService._format_comment_excerpt("abcdefghijk", max_length=3) == "abc"
|
||||
assert WorkflowCommentService._format_comment_excerpt(" abcdefghijk ", max_length=8) == "abcde..."
|
||||
|
||||
def test_build_mention_email_payloads_returns_empty_for_no_candidates(self, mock_session: Mock) -> None:
|
||||
assert (
|
||||
WorkflowCommentService._build_mention_email_payloads(
|
||||
session=mock_session,
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
mentioner_id="user-1",
|
||||
mentioned_user_ids=[],
|
||||
content="hello",
|
||||
)
|
||||
== []
|
||||
)
|
||||
assert (
|
||||
WorkflowCommentService._build_mention_email_payloads(
|
||||
session=mock_session,
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
mentioner_id="user-1",
|
||||
mentioned_user_ids=["user-1"],
|
||||
content="hello",
|
||||
)
|
||||
== []
|
||||
)
|
||||
|
||||
def test_dispatch_mention_emails_enqueues_each_payload(self) -> None:
|
||||
delay_mock = Mock()
|
||||
with patch.object(service_module.send_workflow_comment_mention_email_task, "delay", delay_mock):
|
||||
WorkflowCommentService._dispatch_mention_emails(
|
||||
[
|
||||
{"to": "a@example.com"},
|
||||
{"to": "b@example.com"},
|
||||
]
|
||||
)
|
||||
|
||||
assert delay_mock.call_count == 2
|
||||
|
||||
def test_build_mention_email_payloads_skips_accounts_without_email(self, mock_session: Mock) -> None:
|
||||
account_without_email = Mock()
|
||||
account_without_email.email = None
|
||||
account_without_email.name = "No Email"
|
||||
account_without_email.interface_language = "en-US"
|
||||
|
||||
account_with_email = Mock()
|
||||
account_with_email.email = "user@example.com"
|
||||
account_with_email.name = ""
|
||||
account_with_email.interface_language = None
|
||||
|
||||
mock_session.scalar.side_effect = ["My App", "Commenter"]
|
||||
mock_session.scalars.return_value = _mock_scalars([account_without_email, account_with_email])
|
||||
|
||||
payloads = WorkflowCommentService._build_mention_email_payloads(
|
||||
session=mock_session,
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
mentioner_id="user-1",
|
||||
mentioned_user_ids=["user-2"],
|
||||
content="hello",
|
||||
)
|
||||
expected_app_url = f"{service_module.dify_config.CONSOLE_WEB_URL.rstrip('/')}/app/app-1/workflow"
|
||||
|
||||
assert payloads == [
|
||||
{
|
||||
"language": "en-US",
|
||||
"to": "user@example.com",
|
||||
"mentioned_name": "user@example.com",
|
||||
"commenter_name": "Commenter",
|
||||
"app_name": "My App",
|
||||
"comment_content": "hello",
|
||||
"app_url": expected_app_url,
|
||||
}
|
||||
]
|
||||
|
||||
def test_create_comment_creates_mentions(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_at = "ts"
|
||||
|
||||
with (
|
||||
patch.object(service_module, "WorkflowComment", return_value=comment),
|
||||
patch.object(service_module, "WorkflowCommentMention", return_value=Mock()),
|
||||
patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids", return_value=["user-2"]),
|
||||
):
|
||||
result = WorkflowCommentService.create_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
created_by="user-1",
|
||||
content="hello",
|
||||
position_x=1.0,
|
||||
position_y=2.0,
|
||||
mentioned_user_ids=["user-2", "bad-id"],
|
||||
)
|
||||
|
||||
assert result == {"id": "comment-1", "created_at": "ts"}
|
||||
assert mock_session.add.call_args_list[0].args[0] is comment
|
||||
assert mock_session.add.call_count == 2
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_update_comment_raises_not_found(self, mock_session: Mock) -> None:
|
||||
mock_session.scalar.return_value = None
|
||||
|
||||
with pytest.raises(NotFound):
|
||||
WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="user-1",
|
||||
content="hello",
|
||||
)
|
||||
|
||||
def test_update_comment_raises_forbidden(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.created_by = "owner"
|
||||
mock_session.scalar.return_value = comment
|
||||
|
||||
with pytest.raises(Forbidden):
|
||||
WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="intruder",
|
||||
content="hello",
|
||||
)
|
||||
|
||||
def test_update_comment_replaces_mentions(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_by = "owner"
|
||||
mock_session.scalar.return_value = comment
|
||||
|
||||
existing_mentions = [Mock(), Mock()]
|
||||
mock_session.scalars.return_value = _mock_scalars(existing_mentions)
|
||||
|
||||
with patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids", return_value=["user-2"]):
|
||||
result = WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="owner",
|
||||
content="updated",
|
||||
mentioned_user_ids=["user-2", "bad-id"],
|
||||
)
|
||||
|
||||
assert result == {"id": "comment-1", "updated_at": comment.updated_at}
|
||||
assert mock_session.delete.call_count == 2
|
||||
assert mock_session.add.call_count == 1
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_update_comment_preserves_mentions_when_mentioned_user_ids_omitted(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_by = "owner"
|
||||
mock_session.scalar.return_value = comment
|
||||
|
||||
with (
|
||||
patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids") as filter_mentions_mock,
|
||||
patch.object(WorkflowCommentService, "_build_mention_email_payloads") as build_payloads_mock,
|
||||
patch.object(WorkflowCommentService, "_dispatch_mention_emails") as dispatch_mock,
|
||||
):
|
||||
result = WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="owner",
|
||||
content="updated",
|
||||
)
|
||||
|
||||
assert result == {"id": "comment-1", "updated_at": comment.updated_at}
|
||||
mock_session.delete.assert_not_called()
|
||||
mock_session.add.assert_not_called()
|
||||
filter_mentions_mock.assert_not_called()
|
||||
build_payloads_mock.assert_not_called()
|
||||
dispatch_mock.assert_called_once_with([])
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_update_comment_clears_mentions_when_empty_list_provided(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_by = "owner"
|
||||
mock_session.scalar.return_value = comment
|
||||
|
||||
existing_mentions = [Mock(), Mock()]
|
||||
mock_session.scalars.return_value = _mock_scalars(existing_mentions)
|
||||
|
||||
with patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids", return_value=[]):
|
||||
result = WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="owner",
|
||||
content="updated",
|
||||
mentioned_user_ids=[],
|
||||
)
|
||||
|
||||
assert result == {"id": "comment-1", "updated_at": comment.updated_at}
|
||||
assert mock_session.delete.call_count == 2
|
||||
mock_session.add.assert_not_called()
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_update_comment_notifies_only_new_mentions(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_by = "owner"
|
||||
mock_session.scalar.return_value = comment
|
||||
|
||||
existing_mention = Mock()
|
||||
existing_mention.mentioned_user_id = "user-2"
|
||||
mock_session.scalars.return_value = _mock_scalars([existing_mention])
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
WorkflowCommentService,
|
||||
"_filter_valid_mentioned_user_ids",
|
||||
return_value=["user-2", "user-3"],
|
||||
),
|
||||
patch.object(
|
||||
WorkflowCommentService,
|
||||
"_build_mention_email_payloads",
|
||||
return_value=[],
|
||||
) as build_payloads_mock,
|
||||
patch.object(WorkflowCommentService, "_dispatch_mention_emails") as dispatch_mock,
|
||||
):
|
||||
WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="owner",
|
||||
content="updated",
|
||||
mentioned_user_ids=["user-2", "user-3"],
|
||||
)
|
||||
|
||||
assert build_payloads_mock.call_args.kwargs["mentioned_user_ids"] == ["user-3"]
|
||||
dispatch_mock.assert_called_once_with([])
|
||||
|
||||
def test_get_comments_preloads_related_accounts(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.created_by = "user-1"
|
||||
comment.resolved_by = "user-2"
|
||||
reply = Mock()
|
||||
reply.created_by = "user-3"
|
||||
mention = Mock()
|
||||
mention.mentioned_user_id = "user-4"
|
||||
comment.replies = [reply]
|
||||
comment.mentions = [mention]
|
||||
comment.cache_created_by_account = Mock()
|
||||
comment.cache_resolved_by_account = Mock()
|
||||
reply.cache_created_by_account = Mock()
|
||||
mention.cache_mentioned_user_account = Mock()
|
||||
|
||||
account_1 = Mock()
|
||||
account_1.id = "user-1"
|
||||
account_2 = Mock()
|
||||
account_2.id = "user-2"
|
||||
account_3 = Mock()
|
||||
account_3.id = "user-3"
|
||||
account_4 = Mock()
|
||||
account_4.id = "user-4"
|
||||
|
||||
mock_session.scalars.side_effect = [
|
||||
_mock_scalars([comment]),
|
||||
_mock_scalars([account_1, account_2, account_3, account_4]),
|
||||
]
|
||||
|
||||
result = WorkflowCommentService.get_comments("tenant-1", "app-1")
|
||||
|
||||
assert result == [comment]
|
||||
comment.cache_created_by_account.assert_called_once_with(account_1)
|
||||
comment.cache_resolved_by_account.assert_called_once_with(account_2)
|
||||
reply.cache_created_by_account.assert_called_once_with(account_3)
|
||||
mention.cache_mentioned_user_account.assert_called_once_with(account_4)
|
||||
|
||||
def test_preload_accounts_returns_early_for_empty_comments(self, mock_session: Mock) -> None:
|
||||
WorkflowCommentService._preload_accounts(mock_session, [])
|
||||
|
||||
mock_session.scalars.assert_not_called()
|
||||
|
||||
def test_get_comment_raises_not_found_with_provided_session(self) -> None:
|
||||
session = Mock()
|
||||
session.scalar.return_value = None
|
||||
|
||||
with pytest.raises(NotFound):
|
||||
WorkflowCommentService.get_comment("tenant-1", "app-1", "comment-1", session=session)
|
||||
|
||||
def test_get_comment_uses_context_manager_when_session_not_provided(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.created_by = "user-1"
|
||||
comment.resolved_by = None
|
||||
comment.replies = []
|
||||
comment.mentions = []
|
||||
comment.cache_created_by_account = Mock()
|
||||
comment.cache_resolved_by_account = Mock()
|
||||
mock_session.scalar.return_value = comment
|
||||
mock_session.scalars.return_value = _mock_scalars([])
|
||||
|
||||
result = WorkflowCommentService.get_comment("tenant-1", "app-1", "comment-1")
|
||||
|
||||
assert result is comment
|
||||
comment.cache_created_by_account.assert_called_once()
|
||||
comment.cache_resolved_by_account.assert_called_once_with(None)
|
||||
|
||||
def test_delete_comment_raises_forbidden(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.created_by = "owner"
|
||||
|
||||
with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
|
||||
with pytest.raises(Forbidden):
|
||||
WorkflowCommentService.delete_comment("tenant-1", "app-1", "comment-1", "intruder")
|
||||
|
||||
def test_delete_comment_removes_related_entities(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.created_by = "owner"
|
||||
|
||||
mentions = [Mock(), Mock()]
|
||||
replies = [Mock()]
|
||||
mock_session.scalars.side_effect = [_mock_scalars(mentions), _mock_scalars(replies)]
|
||||
|
||||
with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
|
||||
WorkflowCommentService.delete_comment("tenant-1", "app-1", "comment-1", "owner")
|
||||
|
||||
assert mock_session.delete.call_count == 4
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_resolve_comment_sets_fields(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.resolved = False
|
||||
comment.resolved_at = None
|
||||
comment.resolved_by = None
|
||||
|
||||
with (
|
||||
patch.object(WorkflowCommentService, "get_comment", return_value=comment),
|
||||
patch.object(service_module, "naive_utc_now", return_value="now"),
|
||||
):
|
||||
result = WorkflowCommentService.resolve_comment("tenant-1", "app-1", "comment-1", "user-1")
|
||||
|
||||
assert result is comment
|
||||
assert comment.resolved is True
|
||||
assert comment.resolved_at == "now"
|
||||
assert comment.resolved_by == "user-1"
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_resolve_comment_noop_when_already_resolved(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.resolved = True
|
||||
|
||||
with patch.object(WorkflowCommentService, "get_comment", return_value=comment):
|
||||
result = WorkflowCommentService.resolve_comment("tenant-1", "app-1", "comment-1", "user-1")
|
||||
|
||||
assert result is comment
|
||||
mock_session.commit.assert_not_called()
|
||||
|
||||
def test_create_reply_requires_comment(self, mock_session: Mock) -> None:
|
||||
mock_session.get.return_value = None
|
||||
|
||||
with pytest.raises(NotFound):
|
||||
WorkflowCommentService.create_reply("comment-1", "hello", "user-1")
|
||||
|
||||
def test_create_reply_creates_mentions(self, mock_session: Mock) -> None:
|
||||
mock_session.get.return_value = Mock()
|
||||
reply = Mock()
|
||||
reply.id = "reply-1"
|
||||
reply.created_at = "ts"
|
||||
|
||||
with (
|
||||
patch.object(service_module, "WorkflowCommentReply", return_value=reply),
|
||||
patch.object(service_module, "WorkflowCommentMention", return_value=Mock()),
|
||||
patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids", return_value=["user-2"]),
|
||||
):
|
||||
result = WorkflowCommentService.create_reply(
|
||||
comment_id="comment-1",
|
||||
content="hello",
|
||||
created_by="user-1",
|
||||
mentioned_user_ids=["user-2", "bad-id"],
|
||||
)
|
||||
|
||||
assert result == {"id": "reply-1", "created_at": "ts"}
|
||||
assert mock_session.add.call_count == 2
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_update_reply_raises_not_found(self, mock_session: Mock) -> None:
|
||||
mock_session.scalar.return_value = None
|
||||
|
||||
with pytest.raises(NotFound):
|
||||
WorkflowCommentService.update_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="user-1",
|
||||
content="hello",
|
||||
)
|
||||
|
||||
def test_update_reply_raises_forbidden(self, mock_session: Mock) -> None:
|
||||
reply = Mock()
|
||||
reply.created_by = "owner"
|
||||
mock_session.scalar.return_value = reply
|
||||
|
||||
with pytest.raises(Forbidden):
|
||||
WorkflowCommentService.update_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="intruder",
|
||||
content="hello",
|
||||
)
|
||||
|
||||
def test_update_reply_replaces_mentions(self, mock_session: Mock) -> None:
|
||||
reply = Mock()
|
||||
reply.id = "reply-1"
|
||||
reply.comment_id = "comment-1"
|
||||
reply.created_by = "owner"
|
||||
reply.updated_at = "updated"
|
||||
mock_session.scalar.return_value = reply
|
||||
mock_session.scalars.return_value = _mock_scalars([Mock()])
|
||||
comment = Mock()
|
||||
comment.tenant_id = "tenant-1"
|
||||
comment.app_id = "app-1"
|
||||
mock_session.get.return_value = comment
|
||||
|
||||
with patch.object(WorkflowCommentService, "_filter_valid_mentioned_user_ids", return_value=["user-2"]):
|
||||
result = WorkflowCommentService.update_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="owner",
|
||||
content="new",
|
||||
mentioned_user_ids=["user-2", "bad-id"],
|
||||
)
|
||||
|
||||
assert result == {"id": "reply-1", "updated_at": "updated"}
|
||||
assert mock_session.delete.call_count == 1
|
||||
assert mock_session.add.call_count == 1
|
||||
mock_session.commit.assert_called_once()
|
||||
mock_session.refresh.assert_called_once_with(reply)
|
||||
|
||||
def test_update_comment_updates_position_coordinates_when_provided(self, mock_session: Mock) -> None:
|
||||
comment = Mock()
|
||||
comment.id = "comment-1"
|
||||
comment.created_by = "owner"
|
||||
comment.position_x = 1.0
|
||||
comment.position_y = 2.0
|
||||
mock_session.scalar.return_value = comment
|
||||
mock_session.scalars.return_value = _mock_scalars([])
|
||||
|
||||
WorkflowCommentService.update_comment(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
user_id="owner",
|
||||
content="updated",
|
||||
position_x=10.5,
|
||||
position_y=20.5,
|
||||
mentioned_user_ids=[],
|
||||
)
|
||||
|
||||
assert comment.position_x == 10.5
|
||||
assert comment.position_y == 20.5
|
||||
|
||||
def test_delete_reply_raises_forbidden(self, mock_session: Mock) -> None:
|
||||
reply = Mock()
|
||||
reply.created_by = "owner"
|
||||
mock_session.scalar.return_value = reply
|
||||
|
||||
with pytest.raises(Forbidden):
|
||||
WorkflowCommentService.delete_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="intruder",
|
||||
)
|
||||
|
||||
def test_delete_reply_raises_not_found(self, mock_session: Mock) -> None:
|
||||
mock_session.scalar.return_value = None
|
||||
|
||||
with pytest.raises(NotFound):
|
||||
WorkflowCommentService.delete_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="owner",
|
||||
)
|
||||
|
||||
def test_delete_reply_removes_mentions(self, mock_session: Mock) -> None:
|
||||
reply = Mock()
|
||||
reply.created_by = "owner"
|
||||
mock_session.scalar.return_value = reply
|
||||
mock_session.scalars.return_value = _mock_scalars([Mock(), Mock()])
|
||||
|
||||
WorkflowCommentService.delete_reply(
|
||||
tenant_id="tenant-1",
|
||||
app_id="app-1",
|
||||
comment_id="comment-1",
|
||||
reply_id="reply-1",
|
||||
user_id="owner",
|
||||
)
|
||||
|
||||
assert mock_session.delete.call_count == 3
|
||||
mock_session.commit.assert_called_once()
|
||||
|
||||
def test_validate_comment_access_delegates_to_get_comment(self) -> None:
|
||||
comment = Mock()
|
||||
with patch.object(WorkflowCommentService, "get_comment", return_value=comment) as get_comment_mock:
|
||||
result = WorkflowCommentService.validate_comment_access("comment-1", "tenant-1", "app-1")
|
||||
|
||||
assert result is comment
|
||||
get_comment_mock.assert_called_once_with("tenant-1", "app-1", "comment-1")
|
||||
@@ -12,7 +12,7 @@ This test suite covers:
|
||||
import json
|
||||
import uuid
|
||||
from typing import Any, cast
|
||||
from unittest.mock import ANY, MagicMock, patch
|
||||
from unittest.mock import ANY, MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from graphon.entities import WorkflowNodeExecution
|
||||
@@ -713,6 +713,79 @@ class TestWorkflowService:
|
||||
with pytest.raises(ValueError, match="Invalid app mode"):
|
||||
workflow_service.validate_features_structure(app, features)
|
||||
|
||||
# ==================== Draft Workflow Variable Update Tests ====================
|
||||
# These tests verify updating draft workflow environment/conversation variables
|
||||
|
||||
def test_update_draft_workflow_environment_variables_updates_workflow(self, workflow_service, mock_db_session):
|
||||
"""Test update_draft_workflow_environment_variables updates draft fields."""
|
||||
app = TestWorkflowAssociatedDataFactory.create_app_mock()
|
||||
account = TestWorkflowAssociatedDataFactory.create_account_mock()
|
||||
workflow = TestWorkflowAssociatedDataFactory.create_workflow_mock()
|
||||
variables = [Mock()]
|
||||
|
||||
with (
|
||||
patch.object(workflow_service, "get_draft_workflow", return_value=workflow),
|
||||
patch("services.workflow_service.naive_utc_now", return_value="now"),
|
||||
):
|
||||
workflow_service.update_draft_workflow_environment_variables(
|
||||
app_model=app,
|
||||
environment_variables=variables,
|
||||
account=account,
|
||||
)
|
||||
|
||||
assert workflow.environment_variables == variables
|
||||
assert workflow.updated_by == account.id
|
||||
assert workflow.updated_at == "now"
|
||||
mock_db_session.session.commit.assert_called_once()
|
||||
|
||||
def test_update_draft_workflow_environment_variables_raises_when_missing(self, workflow_service):
|
||||
"""Test update_draft_workflow_environment_variables raises when draft missing."""
|
||||
app = TestWorkflowAssociatedDataFactory.create_app_mock()
|
||||
account = TestWorkflowAssociatedDataFactory.create_account_mock()
|
||||
|
||||
with patch.object(workflow_service, "get_draft_workflow", return_value=None):
|
||||
with pytest.raises(ValueError, match="No draft workflow found."):
|
||||
workflow_service.update_draft_workflow_environment_variables(
|
||||
app_model=app,
|
||||
environment_variables=[],
|
||||
account=account,
|
||||
)
|
||||
|
||||
def test_update_draft_workflow_conversation_variables_updates_workflow(self, workflow_service, mock_db_session):
|
||||
"""Test update_draft_workflow_conversation_variables updates draft fields."""
|
||||
app = TestWorkflowAssociatedDataFactory.create_app_mock()
|
||||
account = TestWorkflowAssociatedDataFactory.create_account_mock()
|
||||
workflow = TestWorkflowAssociatedDataFactory.create_workflow_mock()
|
||||
variables = [Mock()]
|
||||
|
||||
with (
|
||||
patch.object(workflow_service, "get_draft_workflow", return_value=workflow),
|
||||
patch("services.workflow_service.naive_utc_now", return_value="now"),
|
||||
):
|
||||
workflow_service.update_draft_workflow_conversation_variables(
|
||||
app_model=app,
|
||||
conversation_variables=variables,
|
||||
account=account,
|
||||
)
|
||||
|
||||
assert workflow.conversation_variables == variables
|
||||
assert workflow.updated_by == account.id
|
||||
assert workflow.updated_at == "now"
|
||||
mock_db_session.session.commit.assert_called_once()
|
||||
|
||||
def test_update_draft_workflow_conversation_variables_raises_when_missing(self, workflow_service):
|
||||
"""Test update_draft_workflow_conversation_variables raises when draft missing."""
|
||||
app = TestWorkflowAssociatedDataFactory.create_app_mock()
|
||||
account = TestWorkflowAssociatedDataFactory.create_account_mock()
|
||||
|
||||
with patch.object(workflow_service, "get_draft_workflow", return_value=None):
|
||||
with pytest.raises(ValueError, match="No draft workflow found."):
|
||||
workflow_service.update_draft_workflow_conversation_variables(
|
||||
app_model=app,
|
||||
conversation_variables=[],
|
||||
account=account,
|
||||
)
|
||||
|
||||
# ==================== Publish Workflow Tests ====================
|
||||
# These tests verify creating published versions from draft workflows
|
||||
|
||||
|
||||
Reference in New Issue
Block a user