Return the form expired error code in get form definition in WEbAPp Api (vibe-kanban 758765b0)

api/tests/unit_tests/controllers/web/test_human_input_form.py

@@ -15,6 +15,7 @@ from werkzeug.exceptions import Forbidden
 import controllers.web.human_input_form as human_input_module
 import controllers.web.site as site_module
 from models.human_input import RecipientType
+from services.human_input_service import FormExpiredError
 
 HumanInputFormApi = human_input_module.HumanInputFormApi
 TenantStatus = human_input_module.TenantStatus
@@ -59,7 +60,7 @@ class _FakeDB:
 def test_get_form_includes_site(monkeypatch: pytest.MonkeyPatch, app: Flask):
     """GET returns form definition merged with site payload."""
 
-    expiration_time = datetime(2024, 1, 1, tzinfo=UTC)
+    expiration_time = datetime(2099, 1, 1, tzinfo=UTC)
 
     class _FakeDefinition:
         def model_dump(self):
@@ -176,7 +177,7 @@ def test_get_form_includes_site(monkeypatch: pytest.MonkeyPatch, app: Flask):
 def test_get_form_allows_backstage_token(monkeypatch: pytest.MonkeyPatch, app: Flask):
     """GET returns form payload for backstage token."""
 
-    expiration_time = datetime(2024, 1, 2, tzinfo=UTC)
+    expiration_time = datetime(2099, 1, 2, tzinfo=UTC)
 
     class _FakeDefinition:
         def model_dump(self):
@@ -289,7 +290,7 @@ def test_get_form_allows_backstage_token(monkeypatch: pytest.MonkeyPatch, app: F
 def test_get_form_raises_forbidden_when_site_missing(monkeypatch: pytest.MonkeyPatch, app: Flask):
     """GET raises Forbidden if site cannot be resolved."""
 
-    expiration_time = datetime(2024, 1, 3, tzinfo=UTC)
+    expiration_time = datetime(2099, 1, 3, tzinfo=UTC)
 
     class _FakeDefinition:
         def model_dump(self):
@@ -356,3 +357,21 @@ def test_submit_form_accepts_backstage_token(monkeypatch: pytest.MonkeyPatch, ap
         form_data={"content": "ok"},
         submission_end_user_id=None,
     )
+
+
+def test_get_form_raises_expired(monkeypatch: pytest.MonkeyPatch, app: Flask):
+    class _FakeForm:
+        pass
+
+    form = _FakeForm()
+    service_mock = MagicMock()
+    service_mock.get_form_by_token.return_value = form
+    service_mock.ensure_form_active.side_effect = FormExpiredError("form-id")
+    monkeypatch.setattr(human_input_module, "HumanInputService", lambda engine: service_mock)
+    monkeypatch.setattr(human_input_module, "db", _FakeDB(_FakeSession({})))
+
+    with app.test_request_context("/api/form/human_input/token-1", method="GET"):
+        with pytest.raises(FormExpiredError):
+            HumanInputFormApi().get("token-1")
+
+    service_mock.ensure_form_active.assert_called_once_with(form)

api/tests/unit_tests/services/test_human_input_service.py

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
+import services.human_input_service as human_input_service_module
 from core.repositories.human_input_reposotiry import (
     HumanInputFormRecord,
     HumanInputFormSubmissionRepository,
@@ -20,7 +21,7 @@ from core.workflow.nodes.human_input.enums import (
     TimeoutUnit,
 )
 from models.human_input import RecipientType
-from services.human_input_service import HumanInputService, InvalidFormDataError
+from services.human_input_service import Form, FormExpiredError, HumanInputService, InvalidFormDataError
 
 
 @pytest.fixture
@@ -53,6 +54,7 @@ def sample_form_record():
             timeout_unit=TimeoutUnit.HOUR,
         ),
         rendered_content="<p>hello</p>",
+        created_at=datetime.utcnow(),
         expiration_time=datetime.utcnow() + timedelta(hours=1),
         status=HumanInputFormStatus.WAITING,
         selected_action_id=None,
@@ -95,6 +97,20 @@ def test_enqueue_resume_dispatches_task_for_workflow(mocker, mock_session_factor
     assert call_kwargs["kwargs"]["payload"]["workflow_run_id"] == "workflow-run-id"
 
 
+def test_ensure_form_active_respects_global_timeout(monkeypatch, sample_form_record, mock_session_factory):
+    session_factory, _ = mock_session_factory
+    service = HumanInputService(session_factory)
+    expired_record = dataclasses.replace(
+        sample_form_record,
+        created_at=datetime.utcnow() - timedelta(hours=2),
+        expiration_time=datetime.utcnow() + timedelta(hours=2),
+    )
+    monkeypatch.setattr(human_input_service_module.dify_config, "HITL_GLOBAL_TIMEOUT_SECONDS", 3600)
+
+    with pytest.raises(FormExpiredError):
+        service.ensure_form_active(Form(expired_record))
+
+
 def test_enqueue_resume_dispatches_task_for_advanced_chat(mocker, mock_session_factory):
     session_factory, session = mock_session_factory
     service = HumanInputService(session_factory)