Kimirock/dify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: use real collaboration visibility state (#43229)

Browse Source
This commit is contained in:
Asuka Minato
2026-10-03 10:12:10 +00:00
committed by GitHub
parent 01f632d059
commit 8222397ea9
2 changed files with 159 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
api/tests/unit_tests/services/test_workflow_collaboration_service.py
View File

@@ -682,12 +682,11 @@ class TestWorkflowCollaborationService:
)
def test_relay_collaboration_event_graph_view_state_updates_without_broadcast(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> 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-other"
repository.update_session_graph_active.return_value = True
collaboration_service, repository, socketio = real_service
seed_session(repository, "sid-1")
repository.set_leader("wf-1", "sid-other")
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
@@ -698,15 +697,17 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
session_info = repository.get_session_info("wf-1", "sid-1")
assert session_info is not None
assert session_info["graph_active"] is False
socketio.emit.assert_not_called()
repository.set_leader.assert_not_called()
assert repository.get_current_leader("wf-1") == "sid-other"
def test_graph_view_state_serializes_visibility_write_with_leader_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
events: list[str] = []
@contextmanager
@@ -715,8 +716,12 @@ class TestWorkflowCollaborationService:
yield
events.append("lock_exit")
repository.graph_view_state_lock.return_value = graph_view_lock()
repository.update_session_graph_active.side_effect = lambda *_args: events.append("visibility_write") or True
update_visibility = repository.update_session_graph_active
def record_visibility(workflow_id: str, sid: str, active: bool, sequence: int) -> bool:
events.append("visibility_write")
return update_visibility(workflow_id, sid, active, sequence)
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 4},
@@ -724,6 +729,8 @@ class TestWorkflowCollaborationService:
}
with (
patch.object(repository, "graph_view_state_lock", return_value=graph_view_lock()),
patch.object(repository, "update_session_graph_active", side_effect=record_visibility),
patch.object(collaboration_service, "refresh_session_state"),
patch.object(
collaboration_service,
@@ -737,11 +744,11 @@ class TestWorkflowCollaborationService:
assert events == ["lock_enter", "visibility_write", "leader_demotion", "lock_exit"]
def test_relay_collaboration_event_graph_view_state_ignores_stale_update_without_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> None:
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
repository.update_session_graph_active.return_value = False
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
assert repository.update_session_graph_active("wf-1", "sid-1", True, 3)
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 2},
@@ -755,34 +762,38 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_ignored"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 2)
session_info = repository.get_session_info("wf-1", "sid-1")
assert session_info is not None
assert session_info["graph_active"] is True
demote_leader.assert_not_called()
def test_relay_collaboration_event_graph_view_state_demotes_hidden_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> 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-1"
repository.update_session_graph_active.return_value = True
repository.list_sessions.return_value = [
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
{
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-2",
"connected_at": 2,
"graph_active": True,
},
]
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
repository.set_leader("wf-1", "sid-1")
seed_sessions(
repository,
[
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
{
"user_id": "u-2",
"username": "B",
"avatar": None,
"sid": "sid-2",
"connected_at": 2,
"graph_active": True,
},
],
)
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
@@ -797,27 +808,31 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
session_info = repository.get_session_info("wf-1", "sid-1")
assert session_info is not None
assert session_info["graph_active"] is False
assert repository.get_current_leader("wf-1") == "sid-2"
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
def test_relay_collaboration_event_graph_view_state_keeps_leader_when_all_hidden(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> 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-1"
repository.update_session_graph_active.return_value = True
repository.list_sessions.return_value = [
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
]
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
repository.set_leader("wf-1", "sid-1")
seed_sessions(
repository,
[
{
"user_id": "u-1",
"username": "A",
"avatar": None,
"sid": "sid-1",
"connected_at": 1,
"graph_active": False,
},
],
)
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
@@ -832,17 +847,15 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.set_leader.assert_not_called()
repository.delete_leader.assert_not_called()
assert repository.get_current_leader("wf-1") == "sid-1"
broadcast_leader_change.assert_not_called()
def test_relay_collaboration_event_graph_view_state_non_leader_no_demotion(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> 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"
repository.update_session_graph_active.return_value = True
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
repository.set_leader("wf-1", "sid-leader")
payload = {
"type": "graph_view_state",
"data": {"graphActive": False, "sequence": 3},
@@ -856,35 +869,37 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-1", payload)
assert result == ({"msg": "graph_view_state_updated"}, 200)
repository.update_session_graph_active.assert_called_once_with("wf-1", "sid-1", False, 3)
repository.set_leader.assert_not_called()
session_info = repository.get_session_info("wf-1", "sid-1")
assert session_info is not None
assert session_info["graph_active"] is False
assert repository.get_current_leader("wf-1") == "sid-leader"
broadcast_leader_change.assert_not_called()
repository.list_sessions.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,
},
]
def test_relay_collaboration_event_sync_request_reelects_active_leader(self, real_service: ServiceFixture) -> None:
collaboration_service, repository, socketio = real_service
seed_session(repository, "sid-2")
repository.set_leader("wf-1", "sid-old")
seed_sessions(
repository,
[
{
"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,
},
],
)
socketio.call.return_value = {"success": True, "hash": "hash-2", "updatedAt": 456}
payload = {
"type": "sync_request",
@@ -904,14 +919,13 @@ class TestWorkflowCollaborationService:
assert result[1] == 200
assert result[0]["success"] is True
repository.delete_leader.assert_called_once_with("wf-1")
repository.set_leader.assert_called_once_with("wf-1", "sid-2")
assert repository.get_current_leader("wf-1") == "sid-2"
broadcast_leader_change.assert_called_once_with("wf-1", "sid-2")
socketio.call.assert_called_once_with(
"collaboration_update",
{
"type": "sync_request",
"userId": "u-1",
"userId": "u-2",
"data": {"reason": "join", "requestId": "req-1"},
"timestamp": 123,
},
@@ -920,12 +934,12 @@ class TestWorkflowCollaborationService:
)
def test_relay_collaboration_event_sync_request_returns_when_no_active_leader(
self, service: tuple[WorkflowCollaborationService, Mock, Mock]
self, real_service: ServiceFixture
) -> 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 = []
collaboration_service, repository, socketio = real_service
seed_session(repository, "sid-2")
repository.set_leader("wf-1", "sid-old")
seed_sessions(repository, [])
payload = {
"type": "sync_request",
"data": {"reason": "join", "requestId": "req-1"},
@@ -939,13 +953,13 @@ class TestWorkflowCollaborationService:
result = collaboration_service.relay_collaboration_event("sid-2", payload)
assert result == ({"msg": "no_active_leader", "requestId": "req-1"}, 503)
repository.delete_leader.assert_called_once_with("wf-1")
assert repository.get_current_leader("wf-1") is None
socketio.call.assert_not_called()
def test_relay_graph_event_unauthorized(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
def test_relay_graph_event_unauthorized(self, real_service: ServiceFixture) -> None:
# Arrange
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = None
collaboration_service, repository, _socketio = real_service
assert repository.get_sid_mapping("sid-1") is None
# Act
result = collaboration_service.relay_graph_event("sid-1", {"nodes": []})
@@ -953,21 +967,21 @@ class TestWorkflowCollaborationService:
# Assert
assert result == ({"msg": "unauthorized"}, 401)
def test_disconnect_session_no_mapping(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
def test_disconnect_session_no_mapping(self, real_service: ServiceFixture) -> None:
# Arrange
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = None
collaboration_service, repository, _socketio = real_service
assert repository.get_sid_mapping("sid-1") is None
# Act
collaboration_service.disconnect_session("sid-1")
# Assert
repository.delete_session.assert_not_called()
assert repository.list_sessions("wf-1") == []
def test_disconnect_session_cleans_up(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
def test_disconnect_session_cleans_up(self, real_service: ServiceFixture) -> None:
# Arrange
collaboration_service, repository, _socketio = service
repository.get_sid_mapping.return_value = {"workflow_id": "wf-1", "user_id": "u-1"}
collaboration_service, repository, _socketio = real_service
seed_session(repository, "sid-1")
with (
patch.object(collaboration_service, "handle_leader_disconnect") as handle_leader_disconnect,
@@ -977,7 +991,8 @@ class TestWorkflowCollaborationService:
collaboration_service.disconnect_session("sid-1")
# Assert
repository.delete_session.assert_called_once_with("wf-1", "sid-1")
assert not repository.session_exists("wf-1", "sid-1")
assert repository.get_sid_mapping("sid-1") is None
handle_leader_disconnect.assert_called_once_with("wf-1", "sid-1")
broadcast_online_users.assert_called_once_with("wf-1")

34
api/tests/unit_tests/services/workflow_collaboration_fixtures.py
View File

@@ -1,8 +1,16 @@
"""Stateful Redis command transport for collaboration service tests."""
import json
from collections.abc import Sequence
from hashlib import sha1
from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository, WorkflowSessionInfo
from redis.lock import Lock
from repositories.workflow_collaboration_repository import (
_UPDATE_SESSION_GRAPH_ACTIVE_LUA,
WorkflowCollaborationRepository,
WorkflowSessionInfo,
)
class RedisState:
@@ -18,6 +26,30 @@ class RedisState:
return value if isinstance(value, bytes) else str(value).encode()
def execute(self, command: str, *args: object, **_kwargs: object) -> object:
if command == "EVAL":
script, numkeys, workflow_key, sid, active, sequence = args
assert script == _UPDATE_SESSION_GRAPH_ACTIVE_LUA
assert numkeys == 1
fields = self.hashes.get(self._bytes(workflow_key), {})
raw = fields.get(self._bytes(sid))
if raw is None:
return 0
data = json.loads(raw)
incoming = int(str(sequence))
if incoming <= data.get("graph_active_sequence", -1):
return 0
data.update(graph_active=active == "1", graph_active_sequence=incoming)
fields[self._bytes(sid)] = json.dumps(data).encode()
return 1
if command == "EVALSHA":
script_hash, numkeys, lock_key, token = args
assert script_hash == sha1(Lock.LUA_RELEASE_SCRIPT.encode()).hexdigest()
assert numkeys == 1
key = self._bytes(lock_key)
if self.values.get(key) != self._bytes(token):
return 0
del self.values[key]
return 1
key = self._bytes(args[0])
match command:
case "GET":