mirror of
https://github.com/langgenius/dify.git
synced 2026-07-21 18:58:35 +08:00
test: use sqlite3 session in test_agent_app_parameters (#38759)
This commit is contained in:
Asuka Minato
GitHub
parent
b99f8941bb
commit
4bab62d458
@ -1,23 +1,82 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
from typing import Any
|
||||
from uuid import NAMESPACE_URL, uuid5
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from controllers.common.agent_app_parameters import get_published_agent_app_feature_dict_and_user_input_form
|
||||
from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
|
||||
from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
|
||||
from models.agent import Agent, AgentConfigSnapshot, AgentScope, AgentSource, AgentStatus
|
||||
from models.model import AppAnnotationSetting
|
||||
|
||||
|
||||
def _app_model(*, bound_agent_id: str | None, app_model_config=None):
|
||||
def _stable_uuid(value: str) -> str:
|
||||
return str(uuid5(NAMESPACE_URL, value))
|
||||
|
||||
|
||||
def _app_model(*, tenant_id: str, bound_agent_id: str | None, app_model_config: object | None = None):
|
||||
return SimpleNamespace(
|
||||
id="app-1",
|
||||
tenant_id="tenant-1",
|
||||
id=_stable_uuid(f"app:{tenant_id}"),
|
||||
tenant_id=tenant_id,
|
||||
bound_agent_id=bound_agent_id,
|
||||
app_model_config_with_session=lambda *, session: app_model_config,
|
||||
)
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_use_soul_file_upload():
|
||||
def _persist_agent(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
agent_id: str,
|
||||
active_config_snapshot_id: str | None,
|
||||
active_config_is_published: bool,
|
||||
) -> Agent:
|
||||
agent = Agent(
|
||||
id=agent_id,
|
||||
tenant_id=tenant_id,
|
||||
name="Agent",
|
||||
scope=AgentScope.ROSTER,
|
||||
source=AgentSource.AGENT_APP,
|
||||
status=AgentStatus.ACTIVE,
|
||||
active_config_snapshot_id=active_config_snapshot_id,
|
||||
active_config_is_published=active_config_is_published,
|
||||
)
|
||||
session.add(agent)
|
||||
session.commit()
|
||||
return agent
|
||||
|
||||
|
||||
def _persist_snapshot(
|
||||
session: Session,
|
||||
*,
|
||||
snapshot_id: str,
|
||||
tenant_id: str,
|
||||
agent_id: str,
|
||||
config_snapshot: dict[str, Any],
|
||||
) -> AgentConfigSnapshot:
|
||||
snapshot = AgentConfigSnapshot(
|
||||
id=snapshot_id,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
version=1,
|
||||
config_snapshot=config_snapshot,
|
||||
)
|
||||
session.add(snapshot)
|
||||
session.commit()
|
||||
return snapshot
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"sqlite_session",
|
||||
[(Agent, AgentConfigSnapshot, AppAnnotationSetting)],
|
||||
indirect=True,
|
||||
)
|
||||
def test_published_agent_app_parameters_use_soul_file_upload(sqlite_session: Session):
|
||||
tenant_id = _stable_uuid("tenant:one")
|
||||
agent_id = _stable_uuid("agent:one")
|
||||
snapshot_id = _stable_uuid("snapshot:one")
|
||||
app_model_config = SimpleNamespace(
|
||||
to_dict=lambda **_kwargs: {
|
||||
"opening_statement": "Hi from legacy presentation config",
|
||||
@ -27,14 +86,24 @@ def test_published_agent_app_parameters_use_soul_file_upload():
|
||||
},
|
||||
}
|
||||
)
|
||||
app_model = _app_model(bound_agent_id="agent-1", app_model_config=app_model_config)
|
||||
agent = SimpleNamespace(
|
||||
id="agent-1",
|
||||
active_config_snapshot_id="snapshot-1",
|
||||
app_model = _app_model(
|
||||
tenant_id=tenant_id,
|
||||
bound_agent_id=agent_id,
|
||||
app_model_config=app_model_config,
|
||||
)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=snapshot_id,
|
||||
active_config_is_published=True,
|
||||
)
|
||||
snapshot = SimpleNamespace(
|
||||
config_snapshot_dict={
|
||||
_persist_snapshot(
|
||||
sqlite_session,
|
||||
snapshot_id=snapshot_id,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
config_snapshot={
|
||||
"app_features": {
|
||||
"file_upload": {
|
||||
"enabled": True,
|
||||
@ -46,14 +115,12 @@ def test_published_agent_app_parameters_use_soul_file_upload():
|
||||
}
|
||||
},
|
||||
"app_variables": [{"name": "topic", "type": "string", "required": True}],
|
||||
}
|
||||
},
|
||||
)
|
||||
session = MagicMock()
|
||||
session.scalar.side_effect = [agent, snapshot, None]
|
||||
|
||||
features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
|
||||
app_model,
|
||||
session=session,
|
||||
session=sqlite_session,
|
||||
)
|
||||
parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
|
||||
|
||||
@ -69,20 +136,30 @@ def test_published_agent_app_parameters_use_soul_file_upload():
|
||||
assert parameters["user_input_form"] == [{"text-input": {"label": "topic", "variable": "topic", "required": True}}]
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_requires_bound_agent():
|
||||
app_model = _app_model(bound_agent_id=None)
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_requires_bound_agent(sqlite_session: Session):
|
||||
tenant_id = _stable_uuid("tenant:unbound")
|
||||
app_model = _app_model(tenant_id=tenant_id, bound_agent_id=None)
|
||||
|
||||
with pytest.raises(AgentAppGeneratorError, match="no bound Agent"):
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=MagicMock())
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_requires_existing_active_agent():
|
||||
app_model = _app_model(bound_agent_id="agent-1")
|
||||
session = MagicMock()
|
||||
session.scalar.return_value = None
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_requires_existing_active_agent(sqlite_session: Session):
|
||||
requested_tenant_id = _stable_uuid("tenant:requested")
|
||||
agent_id = _stable_uuid("agent:cross-tenant")
|
||||
app_model = _app_model(tenant_id=requested_tenant_id, bound_agent_id=agent_id)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=_stable_uuid("tenant:other"),
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=None,
|
||||
active_config_is_published=False,
|
||||
)
|
||||
|
||||
with pytest.raises(AgentAppGeneratorError, match="no bound Agent"):
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=session)
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
@ -92,68 +169,96 @@ def test_published_agent_app_parameters_requires_existing_active_agent():
|
||||
False,
|
||||
],
|
||||
)
|
||||
def test_published_agent_app_parameters_requires_published_agent(active_config_is_published):
|
||||
app_model = _app_model(bound_agent_id="agent-1")
|
||||
agent = SimpleNamespace(
|
||||
id="agent-1",
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_requires_published_agent(
|
||||
active_config_is_published: bool, sqlite_session: Session
|
||||
):
|
||||
tenant_id = _stable_uuid(f"tenant:published:{active_config_is_published}")
|
||||
agent_id = _stable_uuid(f"agent:published:{active_config_is_published}")
|
||||
app_model = _app_model(tenant_id=tenant_id, bound_agent_id=agent_id)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=None,
|
||||
active_config_is_published=active_config_is_published,
|
||||
)
|
||||
session = MagicMock()
|
||||
session.scalar.return_value = agent
|
||||
|
||||
with pytest.raises(AgentAppNotPublishedError, match="not been published"):
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=session)
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_allows_unpublished_draft_with_active_snapshot():
|
||||
app_model = _app_model(bound_agent_id="agent-1")
|
||||
agent = SimpleNamespace(
|
||||
id="agent-1",
|
||||
active_config_snapshot_id="snapshot-1",
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_allows_unpublished_draft_with_active_snapshot(sqlite_session: Session):
|
||||
tenant_id = _stable_uuid("tenant:unpublished-draft")
|
||||
agent_id = _stable_uuid("agent:unpublished-draft")
|
||||
snapshot_id = _stable_uuid("snapshot:unpublished-draft")
|
||||
app_model = _app_model(tenant_id=tenant_id, bound_agent_id=agent_id)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=snapshot_id,
|
||||
active_config_is_published=False,
|
||||
)
|
||||
snapshot = SimpleNamespace(config_snapshot_dict={})
|
||||
session = MagicMock()
|
||||
session.scalar.side_effect = [agent, snapshot]
|
||||
_persist_snapshot(
|
||||
sqlite_session,
|
||||
snapshot_id=snapshot_id,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
config_snapshot={},
|
||||
)
|
||||
|
||||
features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
|
||||
app_model,
|
||||
session=session,
|
||||
session=sqlite_session,
|
||||
)
|
||||
|
||||
assert features_dict["file_upload"]["enabled"] is True
|
||||
assert user_input_form == []
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_requires_published_snapshot():
|
||||
app_model = _app_model(bound_agent_id="agent-1")
|
||||
agent = SimpleNamespace(
|
||||
id="agent-1",
|
||||
active_config_snapshot_id="snapshot-1",
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_requires_published_snapshot(sqlite_session: Session):
|
||||
tenant_id = _stable_uuid("tenant:missing-snapshot")
|
||||
agent_id = _stable_uuid("agent:missing-snapshot")
|
||||
app_model = _app_model(tenant_id=tenant_id, bound_agent_id=agent_id)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=_stable_uuid("snapshot:missing"),
|
||||
active_config_is_published=True,
|
||||
)
|
||||
session = MagicMock()
|
||||
session.scalar.side_effect = [agent, None]
|
||||
|
||||
with pytest.raises(AgentAppGeneratorError, match="published version not found"):
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=session)
|
||||
get_published_agent_app_feature_dict_and_user_input_form(app_model, session=sqlite_session)
|
||||
|
||||
|
||||
def test_published_agent_app_parameters_allows_missing_legacy_app_model_config():
|
||||
app_model = _app_model(bound_agent_id="agent-1")
|
||||
agent = SimpleNamespace(
|
||||
id="agent-1",
|
||||
active_config_snapshot_id="snapshot-1",
|
||||
@pytest.mark.parametrize("sqlite_session", [(Agent, AgentConfigSnapshot)], indirect=True)
|
||||
def test_published_agent_app_parameters_allows_missing_legacy_app_model_config(sqlite_session: Session):
|
||||
tenant_id = _stable_uuid("tenant:no-legacy-config")
|
||||
agent_id = _stable_uuid("agent:no-legacy-config")
|
||||
snapshot_id = _stable_uuid("snapshot:no-legacy-config")
|
||||
app_model = _app_model(tenant_id=tenant_id, bound_agent_id=agent_id)
|
||||
_persist_agent(
|
||||
sqlite_session,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
active_config_snapshot_id=snapshot_id,
|
||||
active_config_is_published=True,
|
||||
)
|
||||
snapshot = SimpleNamespace(config_snapshot_dict={})
|
||||
session = MagicMock()
|
||||
session.scalar.side_effect = [agent, snapshot]
|
||||
_persist_snapshot(
|
||||
sqlite_session,
|
||||
snapshot_id=snapshot_id,
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent_id,
|
||||
config_snapshot={},
|
||||
)
|
||||
|
||||
features_dict, user_input_form = get_published_agent_app_feature_dict_and_user_input_form(
|
||||
app_model,
|
||||
session=session,
|
||||
session=sqlite_session,
|
||||
)
|
||||
|
||||
assert features_dict["file_upload"] == {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user