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

test: use sqlite3 session in test_workflow_pause_events (#38688)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-29 14:49:15 +09:00 committed by GitHub
parent 8946d7fbf4
commit b6bfae7e3c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 115 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
api/core/app/apps/common/workflow_response_converter.py
View File

@ -76,7 +76,7 @@ from graphon.runtime import GraphRuntimeState
from graphon.variables.segments import ArrayFileSegment, FileSegment, Segment
from graphon.variables.variables import Variable
from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
from libs.datetime_utils import naive_utc_now
from libs.datetime_utils import naive_utc_now, to_utc_timestamp
from models import Account, EndUser
from models.human_input import HumanInputForm
from models.workflow import WorkflowRun
@ -371,7 +371,7 @@ class WorkflowResponseConverter:
pause_reasons,
dispositions_by_form_id=dispositions_by_form_id,
expiration_times_by_form_id={
form_id: int(expiration_time.timestamp())
form_id: to_utc_timestamp(expiration_time)
for form_id, expiration_time in expiration_times_by_form_id.items()
},
)
@ -399,7 +399,7 @@ class WorkflowResponseConverter:
form_token=disposition.form_token if disposition else None,
approval_channels=list(disposition.approval_channels) if disposition else [],
resolved_default_values=reason.resolved_default_values,
expiration_time=int(expiration_time.timestamp()),
expiration_time=to_utc_timestamp(expiration_time),
),
)
)
@ -452,7 +452,7 @@ class WorkflowResponseConverter:
data=HumanInputFormTimeoutResponse.Data(
node_id=event.node_id,
node_title=event.node_title,
expiration_time=int(event.expiration_time.timestamp()),
expiration_time=to_utc_timestamp(event.expiration_time),
),
)

9
api/libs/datetime_utils.py
View File

@ -35,6 +35,15 @@ def ensure_naive_utc(dt: datetime.datetime) -> datetime.datetime:
return dt.astimezone(datetime.UTC).replace(tzinfo=None)
def to_utc_timestamp(dt: datetime.datetime) -> int:
"""Convert a datetime to Unix epoch seconds, assuming naive values are UTC.
Persisted datetimes may be returned without timezone information. Treat
those values as UTC instead of interpreting them in the host timezone.
"""
return int(ensure_naive_utc(dt).replace(tzinfo=datetime.UTC).timestamp())
def parse_time_range(
start: str | None, end: str | None, tzname: str
) -> tuple[datetime.datetime | None, datetime.datetime | None]:

5
api/services/workflow_event_snapshot_service.py
View File

@ -43,6 +43,7 @@ from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus
from graphon.runtime import GraphRuntimeState
from graphon.runtime.graph_runtime_state_protocol import ReadOnlyVariablePool
from graphon.workflow_type_encoder import WorkflowRuntimeTypeConverter
from libs.datetime_utils import to_utc_timestamp
from models.human_input import HumanInputForm
from models.model import AppMode, Message
from models.workflow import WorkflowNodeExecutionTriggeredFrom, WorkflowRun
@ -451,7 +452,7 @@ def _build_human_input_required_events(
)
with session_maker() as session:
for form_id, expiration_time, form_definition in session.execute(stmt):
expiration_times_by_form_id[str(form_id)] = int(expiration_time.timestamp())
expiration_times_by_form_id[str(form_id)] = to_utc_timestamp(expiration_time)
try:
definition_payload = json.loads(form_definition) if form_definition else {}
except (TypeError, json.JSONDecodeError):
@ -594,7 +595,7 @@ def _build_pause_event(
)
for row in session.execute(stmt):
form_id, expiration_time, *_rest = row
expiration_times_by_form_id[str(form_id)] = int(expiration_time.timestamp())
expiration_times_by_form_id[str(form_id)] = to_utc_timestamp(expiration_time)
# Reconnect paths must preserve the same pause-reason contract as live streams;
# otherwise clients see schema drift after resume.
reasons = enrich_human_input_pause_reasons(

143
api/tests/unit_tests/core/app/apps/test_workflow_pause_events.py
View File

@ -3,6 +3,7 @@ from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from sqlalchemy.orm import Session
from core.app.apps.common import workflow_response_converter
from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
@ -24,27 +25,7 @@ from graphon.entities.pause_reason import HitlRequired
from graphon.graph_events import GraphRunPausedEvent
from graphon.runtime import GraphRuntimeState, VariablePool
from models.account import Account
from models.human_input import RecipientType
class _FakeSession:
"""Stub session: `execute` feeds the form-expiration query, `scalars` the recipients."""
def __init__(self, *, execute_rows=(), scalars_rows=()):
self._execute_rows = execute_rows
self._scalars_rows = scalars_rows
def execute(self, _stmt):
return list(self._execute_rows)
def scalars(self, _stmt):
return list(self._scalars_rows)
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
from models.human_input import HumanInputForm, HumanInputFormRecipient, RecipientType
class _RecordingWorkflowAppRunner(WorkflowAppRunner):
@ -63,6 +44,46 @@ class _FakeRuntimeState:
return ["node-pause-1"]
@pytest.fixture
def sqlite_pause_session(sqlite_session: Session, monkeypatch: pytest.MonkeyPatch) -> Session:
"""Bind pause-response queries to the shared SQLite session's database."""
monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=sqlite_session.get_bind()))
return sqlite_session
def _persist_human_input_form(
session: Session,
*,
recipients: list[tuple[RecipientType, str]] | None = None,
) -> datetime:
expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
form = HumanInputForm(
id="form-1",
tenant_id="tenant-id",
app_id="app-id",
workflow_run_id="run-id",
node_id="node-id",
form_definition='{"display_in_ui": true}',
rendered_content="Rendered",
expiration_time=expiration_time,
)
recipient_models = [
HumanInputFormRecipient(
id=f"recipient-{index}",
form_id=form.id,
delivery_id=f"delivery-{index}",
recipient_type=recipient_type,
recipient_payload="{}",
access_token=access_token,
)
for index, (recipient_type, access_token) in enumerate(recipients or ())
]
session.add(form)
session.add_all(recipient_models)
session.commit()
return expiration_time
def _build_runner():
app_entity = SimpleNamespace(
app_config=SimpleNamespace(app_id="app-id"),
@ -154,7 +175,12 @@ def _build_converter(*, invoke_from: InvokeFrom = InvokeFrom.SERVICE_API):
)
def test_queue_workflow_paused_event_to_stream_responses(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize(
"sqlite_session",
[(HumanInputForm, HumanInputFormRecipient)],
indirect=True,
)
def test_queue_workflow_paused_event_to_stream_responses(sqlite_pause_session: Session):
converter = _build_converter()
converter.workflow_start_to_stream_response(
task_id="task",
@ -163,18 +189,14 @@ def test_queue_workflow_paused_event_to_stream_responses(monkeypatch: pytest.Mon
reason=WorkflowStartReason.INITIAL,
)
expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
session = _FakeSession(
execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')],
scalars_rows=[
SimpleNamespace(form_id="form-1", recipient_type=RecipientType.CONSOLE, access_token="console-token"),
SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
expiration_time = _persist_human_input_form(
sqlite_pause_session,
recipients=[
(RecipientType.CONSOLE, "console-token"),
(RecipientType.BACKSTAGE, "backstage-token"),
],
)
monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))
reason = HumanInputRequired(
form_id="form-1",
form_content="Rendered",
@ -216,8 +238,11 @@ def test_queue_workflow_paused_event_to_stream_responses(monkeypatch: pytest.Mon
assert hi_resp.data.expiration_time == int(expiration_time.timestamp())
def _build_paused_human_input_response(monkeypatch, recipients):
"""Drive the live OPENAPI pause path with the given recipients via a fake session."""
def _build_paused_human_input_response(
session: Session,
recipients: list[tuple[RecipientType, str]],
):
"""Drive the live OPENAPI pause path with persisted forms and recipients."""
converter = _build_converter(invoke_from=InvokeFrom.OPENAPI)
converter.workflow_start_to_stream_response(
task_id="task",
@ -226,14 +251,7 @@ def _build_paused_human_input_response(monkeypatch, recipients):
reason=WorkflowStartReason.INITIAL,
)
expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
session = _FakeSession(
execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')],
scalars_rows=list(recipients),
)
monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))
_persist_human_input_form(session, recipients=recipients)
reason = HumanInputRequired(
form_id="form-1",
@ -259,12 +277,17 @@ def _build_paused_human_input_response(monkeypatch, recipients):
return responses
def test_openapi_pause_without_web_app_recipient_emits_approval_channels(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize(
"sqlite_session",
[(HumanInputForm, HumanInputFormRecipient)],
indirect=True,
)
def test_openapi_pause_without_web_app_recipient_emits_approval_channels(sqlite_pause_session: Session):
responses = _build_paused_human_input_response(
monkeypatch,
sqlite_pause_session,
recipients=[
SimpleNamespace(form_id="form-1", recipient_type=RecipientType.EMAIL_MEMBER, access_token="email-token"),
SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
(RecipientType.EMAIL_MEMBER, "email-token"),
(RecipientType.BACKSTAGE, "backstage-token"),
],
)
@ -276,16 +299,17 @@ def test_openapi_pause_without_web_app_recipient_emits_approval_channels(monkeyp
assert pause_resp.data.reasons[0]["approval_channels"] == ["console", "email"]
def test_openapi_pause_with_web_app_recipient_sets_token_and_channels(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize(
"sqlite_session",
[(HumanInputForm, HumanInputFormRecipient)],
indirect=True,
)
def test_openapi_pause_with_web_app_recipient_sets_token_and_channels(sqlite_pause_session: Session):
responses = _build_paused_human_input_response(
monkeypatch,
sqlite_pause_session,
recipients=[
SimpleNamespace(
form_id="form-1",
recipient_type=RecipientType.STANDALONE_WEB_APP,
access_token="web-app-token",
),
SimpleNamespace(form_id="form-1", recipient_type=RecipientType.BACKSTAGE, access_token="backstage-token"),
(RecipientType.STANDALONE_WEB_APP, "web-app-token"),
(RecipientType.BACKSTAGE, "backstage-token"),
],
)
@ -297,7 +321,12 @@ def test_openapi_pause_with_web_app_recipient_sets_token_and_channels(monkeypatc
assert pause_resp.data.reasons[0]["approval_channels"] == ["console"]
def test_queue_workflow_paused_event_resolves_variable_select_options(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize(
"sqlite_session",
[(HumanInputForm, HumanInputFormRecipient)],
indirect=True,
)
def test_queue_workflow_paused_event_resolves_variable_select_options(sqlite_pause_session: Session):
converter = _build_converter()
converter.workflow_start_to_stream_response(
task_id="task",
@ -306,11 +335,7 @@ def test_queue_workflow_paused_event_resolves_variable_select_options(monkeypatc
reason=WorkflowStartReason.INITIAL,
)
expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
session = _FakeSession(execute_rows=[("form-1", expiration_time, '{"display_in_ui": true}')])
monkeypatch.setattr(workflow_response_converter, "Session", lambda **_: session)
monkeypatch.setattr(workflow_response_converter, "db", SimpleNamespace(engine=object()))
_persist_human_input_form(sqlite_pause_session)
reason = HumanInputRequired(
form_id="form-1",

14
api/tests/unit_tests/libs/test_datetime_utils.py
View File

@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest
import pytz
from libs.datetime_utils import naive_utc_now, parse_time_range
from libs.datetime_utils import naive_utc_now, parse_time_range, to_utc_timestamp
def test_naive_utc_now(monkeypatch: pytest.MonkeyPatch):
@ -24,6 +24,18 @@ def test_naive_utc_now(monkeypatch: pytest.MonkeyPatch):
assert naive_time == utc_time
@pytest.mark.parametrize(
"value",
[
datetime.datetime(2024, 1, 1),
datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC),
datetime.datetime(2024, 1, 1, 9, tzinfo=datetime.timezone(datetime.timedelta(hours=9))),
],
)
def test_to_utc_timestamp(value: datetime.datetime):
assert to_utc_timestamp(value) == 1704067200
class TestParseTimeRange:
"""Test cases for parse_time_range function."""

3
api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py
View File

@ -985,7 +985,8 @@ def test_build_snapshot_events_preserves_public_form_token(monkeypatch: pytest.M
)
session_maker = _SessionMaker(
SimpleNamespace(
execute=lambda _stmt: [("form-1", datetime(2024, 1, 1, tzinfo=UTC), '{"display_in_ui": true}')],
# Persisted UTC datetimes can be loaded as naive values.
execute=lambda _stmt: [("form-1", datetime(2024, 1, 1), '{"display_in_ui": true}')],
)
)
pause_entity = _FakePauseEntity(