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

test: use sqlite3 session in test_human_input_forms (#38777)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-22 11:27:30 +09:00 committed by GitHub
parent 97489d5a44
commit f35655a5bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 51 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
api/tests/unit_tests/core/workflow/test_human_input_forms.py
View File

@ -1,6 +1,7 @@
from types import SimpleNamespace
from uuid import uuid4
import pytest
from sqlalchemy.orm import Session
from core.workflow.human_input_forms import (
load_form_dispositions_by_form_id,
@ -11,19 +12,24 @@ from core.workflow.human_input_policy import (
HumanInputSurface,
disposition_for_surface,
)
from models.human_input import RecipientType
from models.human_input import HumanInputFormRecipient, RecipientType
TABLES = (HumanInputFormRecipient,)
class _FakeSession:
def __init__(self, recipients: list[SimpleNamespace]) -> None:
self._recipients = recipients
def scalars(self, _stmt):
return self._recipients
def _recipient(form_id: str, recipient_type: RecipientType, access_token: str) -> HumanInputFormRecipient:
return HumanInputFormRecipient(
form_id=form_id,
delivery_id=str(uuid4()),
recipient_type=recipient_type,
recipient_payload="{}",
access_token=access_token,
)
def _recipient(form_id: str, recipient_type: RecipientType, access_token: str | None) -> SimpleNamespace:
return SimpleNamespace(form_id=form_id, recipient_type=recipient_type, access_token=access_token)
def _persist_recipients(session: Session, recipients: list[HumanInputFormRecipient]) -> None:
session.add_all(recipients)
session.commit()
@pytest.mark.parametrize(
@ -35,60 +41,75 @@ def _recipient(form_id: str, recipient_type: RecipientType, access_token: str |
(HumanInputSurface.SERVICE_API, "web-token"),
],
)
def test_load_form_tokens_picks_token_for_surface(surface, expected_token) -> None:
session = _FakeSession(
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_load_form_tokens_picks_token_for_surface(surface, expected_token, sqlite_session: Session) -> None:
_persist_recipients(
sqlite_session,
[
_recipient("form-1", RecipientType.STANDALONE_WEB_APP, "web-token"),
_recipient("form-1", RecipientType.CONSOLE, "console-token"),
_recipient("form-1", RecipientType.BACKSTAGE, "backstage-token"),
]
_recipient("form-2", RecipientType.BACKSTAGE, "decoy-token"),
],
)
assert load_form_tokens_by_form_id(["form-1"], session=session, surface=surface) == {"form-1": expected_token}
assert load_form_tokens_by_form_id(["form-1"], session=sqlite_session, surface=surface) == {
"form-1": expected_token
}
def test_load_form_tokens_drops_forms_without_actionable_token() -> None:
session = _FakeSession(
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_load_form_tokens_drops_forms_without_actionable_token(sqlite_session: Session) -> None:
_persist_recipients(
sqlite_session,
[
_recipient("form-1", RecipientType.EMAIL_MEMBER, "email-token"),
_recipient("form-1", RecipientType.CONSOLE, None),
]
_recipient("form-1", RecipientType.CONSOLE, ""),
],
)
assert load_form_tokens_by_form_id(["form-1"], session=session) == {}
assert load_form_tokens_by_form_id(["form-1"], session=sqlite_session) == {}
def test_load_form_tokens_service_api_surface_uses_web_token() -> None:
session = _FakeSession(
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_load_form_tokens_service_api_surface_uses_web_token(sqlite_session: Session) -> None:
_persist_recipients(
sqlite_session,
[
_recipient("form-1", RecipientType.STANDALONE_WEB_APP, "web-token"),
_recipient("form-1", RecipientType.CONSOLE, "console-token"),
_recipient("form-1", RecipientType.BACKSTAGE, "backstage-token"),
]
],
)
assert load_form_tokens_by_form_id(["form-1"], session=session, surface=HumanInputSurface.SERVICE_API) == {
assert load_form_tokens_by_form_id(["form-1"], session=sqlite_session, surface=HumanInputSurface.SERVICE_API) == {
"form-1": "web-token"
}
def test_load_dispositions_openapi_webapp_form_is_resumable() -> None:
session = _FakeSession(
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_load_dispositions_openapi_webapp_form_is_resumable(sqlite_session: Session) -> None:
_persist_recipients(
sqlite_session,
[
_recipient("form-1", RecipientType.STANDALONE_WEB_APP, "web-token"),
_recipient("form-1", RecipientType.BACKSTAGE, "backstage-token"),
]
],
)
assert load_form_dispositions_by_form_id(["form-1"], session=session, surface=HumanInputSurface.OPENAPI) == {
assert load_form_dispositions_by_form_id(["form-1"], session=sqlite_session, surface=HumanInputSurface.OPENAPI) == {
"form-1": FormDisposition(form_token="web-token", approval_channels=["console"])
}
def test_load_dispositions_openapi_backstage_only_form_yields_channels_not_token() -> None:
session = _FakeSession([_recipient("form-1", RecipientType.BACKSTAGE, "backstage-token")])
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_load_dispositions_openapi_backstage_only_form_yields_channels_not_token(sqlite_session: Session) -> None:
_persist_recipients(
sqlite_session,
[_recipient("form-1", RecipientType.BACKSTAGE, "backstage-token")],
)
assert load_form_dispositions_by_form_id(["form-1"], session=session, surface=HumanInputSurface.OPENAPI) == {
assert load_form_dispositions_by_form_id(["form-1"], session=sqlite_session, surface=HumanInputSurface.OPENAPI) == {
"form-1": FormDisposition(form_token=None, approval_channels=["console"])
}