diff --git a/api/core/app/apps/common/workflow_response_converter.py b/api/core/app/apps/common/workflow_response_converter.py index 236b8c58d3a..01ed7c2c842 100644 --- a/api/core/app/apps/common/workflow_response_converter.py +++ b/api/core/app/apps/common/workflow_response_converter.py @@ -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), ), ) diff --git a/api/libs/datetime_utils.py b/api/libs/datetime_utils.py index e0a6ec2cacd..d962d81e78a 100644 --- a/api/libs/datetime_utils.py +++ b/api/libs/datetime_utils.py @@ -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]: diff --git a/api/services/workflow_event_snapshot_service.py b/api/services/workflow_event_snapshot_service.py index 1758c13a803..58f42badc66 100644 --- a/api/services/workflow_event_snapshot_service.py +++ b/api/services/workflow_event_snapshot_service.py @@ -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( diff --git a/api/tests/unit_tests/core/app/apps/test_workflow_pause_events.py b/api/tests/unit_tests/core/app/apps/test_workflow_pause_events.py index c0052bc5bab..41064ea8830 100644 --- a/api/tests/unit_tests/core/app/apps/test_workflow_pause_events.py +++ b/api/tests/unit_tests/core/app/apps/test_workflow_pause_events.py @@ -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", diff --git a/api/tests/unit_tests/libs/test_datetime_utils.py b/api/tests/unit_tests/libs/test_datetime_utils.py index 57314d29d4b..e2dce54d1e0 100644 --- a/api/tests/unit_tests/libs/test_datetime_utils.py +++ b/api/tests/unit_tests/libs/test_datetime_utils.py @@ -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.""" diff --git a/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py b/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py index 58892f0ebb3..4260f0064c6 100644 --- a/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py +++ b/api/tests/unit_tests/services/workflow/test_workflow_event_snapshot_service.py @@ -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(