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

test: use sqlite3 session in test_snippet_workflow (#38677)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-22 18:20:21 +09:00 committed by GitHub
parent 8ee3a7eabc
commit 9b2aa8b216
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 64 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

129
api/tests/unit_tests/controllers/console/snippets/test_snippet_workflow.py
View File

@ -8,6 +8,8 @@ from unittest.mock import Mock
import pytest
from flask import Flask
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker
from werkzeug.exceptions import HTTPException, NotFound
from controllers.console.snippets import snippet_workflow as snippet_workflow_module
@ -36,7 +38,9 @@ def _snippet(**overrides) -> CustomizedSnippet:
@pytest.fixture(autouse=True)
def _patch_snippet_service_factory(monkeypatch: pytest.MonkeyPatch) -> None:
def _patch_snippet_service_factory(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
snippet_session_maker = sessionmaker(bind=sqlite_engine, expire_on_commit=False)
def factory():
try:
return snippet_workflow_module.SnippetService(snippet_workflow_module._snippet_session_maker())
@ -44,7 +48,7 @@ def _patch_snippet_service_factory(monkeypatch: pytest.MonkeyPatch) -> None:
return snippet_workflow_module.SnippetService()
monkeypatch.setattr(snippet_workflow_module, "_snippet_service", factory)
monkeypatch.setattr(snippet_workflow_module, "_snippet_session_maker", Mock(return_value=Mock()))
monkeypatch.setattr(snippet_workflow_module, "_snippet_session_maker", lambda: snippet_session_maker)
def test_get_snippet_requires_snippet_id(app):
@ -150,28 +154,28 @@ def test_published_workflow_get_returns_none_when_not_published(app) -> None:
assert handler(api, snippet=SimpleNamespace(id="snippet-1", is_published=False)) is None
def test_published_workflow_post_returns_400_when_publish_fails(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(CustomizedSnippet,)], indirect=True)
def test_published_workflow_post_returns_400_when_publish_fails(
app: Flask,
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
sqlite_session: Session,
) -> None:
user = _account("account-1")
snippet = _snippet()
merged_snippet = _snippet()
session = SimpleNamespace(merge=Mock(return_value=merged_snippet), commit=Mock())
sqlite_session.add(snippet)
sqlite_session.commit()
class SessionContext:
def __init__(self, engine):
self.engine = engine
def fail_publish(*, session: Session, snippet: CustomizedSnippet, account: Account):
snippet.name = "Uncommitted name"
session.add(snippet)
raise ValueError("No valid workflow found.")
def __enter__(self):
return session
def __exit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(snippet_workflow_module, "Session", SessionContext)
monkeypatch.setattr(snippet_workflow_module, "db", SimpleNamespace(engine=object()))
monkeypatch.setattr(snippet_workflow_module, "db", SimpleNamespace(engine=sqlite_engine))
monkeypatch.setattr(
snippet_workflow_module,
"SnippetService",
lambda: SimpleNamespace(publish_workflow=Mock(side_effect=ValueError("No valid workflow found."))),
lambda: SimpleNamespace(publish_workflow=Mock(side_effect=fail_publish)),
)
api = snippet_workflow_module.SnippetPublishedWorkflowApi()
@ -182,7 +186,8 @@ def test_published_workflow_post_returns_400_when_publish_fails(app: Flask, monk
assert status_code == 400
assert response == {"message": "No valid workflow found."}
session.commit.assert_not_called()
sqlite_session.refresh(snippet)
assert snippet.name == "Snippet"
def test_default_block_configs_delegates_to_service(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@ -203,7 +208,11 @@ def test_default_block_configs_delegates_to_service(app: Flask, monkeypatch: pyt
get_default_block_configs.assert_called_once()
def test_list_published_snippet_workflows_includes_input_fields(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
def test_list_published_snippet_workflows_includes_input_fields(
app: Flask,
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
) -> None:
workflow = SimpleNamespace(
id="workflow-1",
graph_dict={"nodes": [], "edges": []},
@ -224,18 +233,7 @@ def test_list_published_snippet_workflows_includes_input_fields(app: Flask, monk
input_fields = [{"variable": "query", "type": "text"}]
snippet = _snippet(input_fields=json.dumps(input_fields))
class SessionContext:
def __init__(self, engine):
self.engine = engine
def __enter__(self):
return Mock()
def __exit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(snippet_workflow_module, "Session", SessionContext)
monkeypatch.setattr(snippet_workflow_module, "db", SimpleNamespace(engine=object()))
monkeypatch.setattr(snippet_workflow_module, "db", SimpleNamespace(engine=sqlite_engine))
monkeypatch.setattr(
snippet_workflow_module,
"SnippetService",
@ -364,8 +362,11 @@ def test_restore_published_snippet_workflow_to_draft_returns_400_for_invalid_gra
assert exc.value.description == "invalid snippet workflow graph"
@pytest.mark.parametrize("sqlite_session", [(CustomizedSnippet,)], indirect=True)
def test_update_published_snippet_workflow_returns_updated_workflow(
app: Flask, monkeypatch: pytest.MonkeyPatch
app: Flask,
monkeypatch: pytest.MonkeyPatch,
sqlite_session: Session,
) -> None:
workflow = SimpleNamespace(
id="workflow-1",
@ -387,21 +388,15 @@ def test_update_published_snippet_workflow_returns_updated_workflow(
user = _account("account-1")
input_fields = [{"variable": "query", "type": "text"}]
snippet = _snippet(input_fields=json.dumps(input_fields))
session = SimpleNamespace()
update_workflow = Mock(return_value=workflow)
sqlite_session.add(snippet)
sqlite_session.commit()
class TransactionContext:
def __enter__(self):
return session
def update_persisted_snippet(*, session: Session, snippet: CustomizedSnippet, **_kwargs):
merged_snippet = session.merge(snippet)
merged_snippet.description = "Updated in transaction"
return workflow
def __exit__(self, exc_type, exc, tb):
return False
class SessionMaker:
def begin(self):
return TransactionContext()
monkeypatch.setattr(snippet_workflow_module, "_snippet_session_maker", Mock(return_value=SessionMaker()))
update_workflow = Mock(side_effect=update_persisted_snippet)
monkeypatch.setattr(
snippet_workflow_module,
"SnippetService",
@ -418,16 +413,18 @@ def test_update_published_snippet_workflow_returns_updated_workflow(
):
response = handler(api, user, snippet, workflow_id="workflow-1")
update_workflow.assert_called_once_with(
session=session,
snippet=snippet,
workflow_id="workflow-1",
account=user,
data={"marked_name": "v1", "marked_comment": "first version"},
)
update_workflow.assert_called_once()
update_call = update_workflow.call_args.kwargs
assert isinstance(update_call["session"], Session)
assert update_call["snippet"] is snippet
assert update_call["workflow_id"] == "workflow-1"
assert update_call["account"] is user
assert update_call["data"] == {"marked_name": "v1", "marked_comment": "first version"}
assert response["marked_name"] == "v1"
assert response["marked_comment"] == "first version"
assert response["input_fields"] == input_fields
sqlite_session.refresh(snippet)
assert snippet.description == "Updated in transaction"
def test_update_published_snippet_workflow_returns_400_when_no_fields(app: Flask) -> None:
@ -441,26 +438,25 @@ def test_update_published_snippet_workflow_returns_400_when_no_fields(app: Flask
assert response == {"message": "No valid fields to update"}
def test_update_published_snippet_workflow_raises_not_found(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
@pytest.mark.parametrize("sqlite_session", [(CustomizedSnippet,)], indirect=True)
def test_update_published_snippet_workflow_raises_not_found(
app: Flask,
monkeypatch: pytest.MonkeyPatch,
sqlite_session: Session,
) -> None:
user = _account("account-1")
snippet = _snippet()
sqlite_session.add(snippet)
sqlite_session.commit()
class TransactionContext:
def __enter__(self):
return SimpleNamespace()
def update_missing_workflow(*, session: Session, snippet: CustomizedSnippet, **_kwargs):
merged_snippet = session.merge(snippet)
merged_snippet.name = "Rolled back name"
def __exit__(self, exc_type, exc, tb):
return False
class SessionMaker:
def begin(self):
return TransactionContext()
monkeypatch.setattr(snippet_workflow_module, "_snippet_session_maker", Mock(return_value=SessionMaker()))
monkeypatch.setattr(
snippet_workflow_module,
"SnippetService",
lambda: SimpleNamespace(update_workflow=Mock(return_value=None)),
lambda: SimpleNamespace(update_workflow=Mock(side_effect=update_missing_workflow)),
)
api = snippet_workflow_module.SnippetWorkflowByIdApi()
@ -474,6 +470,9 @@ def test_update_published_snippet_workflow_raises_not_found(app: Flask, monkeypa
with pytest.raises(NotFound, match="Workflow not found"):
handler(api, user, snippet, workflow_id="missing-workflow")
sqlite_session.refresh(snippet)
assert snippet.name == "Snippet"
def test_workflow_run_detail_raises_not_found_when_run_missing(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
snippet = _snippet()