mirror of
https://gitee.com/dify_ai/dify.git
synced 2024-12-02 19:27:48 +08:00
124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
from unittest import mock
|
|
from uuid import uuid4
|
|
|
|
import contexts
|
|
from constants import HIDDEN_VALUE
|
|
from core.app.segments import FloatVariable, IntegerVariable, SecretVariable, StringVariable
|
|
from models.workflow import Workflow
|
|
|
|
|
|
def test_environment_variables():
|
|
contexts.tenant_id.set("tenant_id")
|
|
|
|
# Create a Workflow instance
|
|
workflow = Workflow(
|
|
tenant_id="tenant_id",
|
|
app_id="app_id",
|
|
type="workflow",
|
|
version="draft",
|
|
graph="{}",
|
|
features="{}",
|
|
created_by="account_id",
|
|
environment_variables=[],
|
|
conversation_variables=[],
|
|
)
|
|
|
|
# Create some EnvironmentVariable instances
|
|
variable1 = StringVariable.model_validate({"name": "var1", "value": "value1", "id": str(uuid4())})
|
|
variable2 = IntegerVariable.model_validate({"name": "var2", "value": 123, "id": str(uuid4())})
|
|
variable3 = SecretVariable.model_validate({"name": "var3", "value": "secret", "id": str(uuid4())})
|
|
variable4 = FloatVariable.model_validate({"name": "var4", "value": 3.14, "id": str(uuid4())})
|
|
|
|
with (
|
|
mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
|
|
mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
|
|
):
|
|
# Set the environment_variables property of the Workflow instance
|
|
variables = [variable1, variable2, variable3, variable4]
|
|
workflow.environment_variables = variables
|
|
|
|
# Get the environment_variables property and assert its value
|
|
assert workflow.environment_variables == variables
|
|
|
|
|
|
def test_update_environment_variables():
|
|
contexts.tenant_id.set("tenant_id")
|
|
|
|
# Create a Workflow instance
|
|
workflow = Workflow(
|
|
tenant_id="tenant_id",
|
|
app_id="app_id",
|
|
type="workflow",
|
|
version="draft",
|
|
graph="{}",
|
|
features="{}",
|
|
created_by="account_id",
|
|
environment_variables=[],
|
|
conversation_variables=[],
|
|
)
|
|
|
|
# Create some EnvironmentVariable instances
|
|
variable1 = StringVariable.model_validate({"name": "var1", "value": "value1", "id": str(uuid4())})
|
|
variable2 = IntegerVariable.model_validate({"name": "var2", "value": 123, "id": str(uuid4())})
|
|
variable3 = SecretVariable.model_validate({"name": "var3", "value": "secret", "id": str(uuid4())})
|
|
variable4 = FloatVariable.model_validate({"name": "var4", "value": 3.14, "id": str(uuid4())})
|
|
|
|
with (
|
|
mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
|
|
mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
|
|
):
|
|
variables = [variable1, variable2, variable3, variable4]
|
|
|
|
# Set the environment_variables property of the Workflow instance
|
|
workflow.environment_variables = variables
|
|
assert workflow.environment_variables == [variable1, variable2, variable3, variable4]
|
|
|
|
# Update the name of variable3 and keep the value as it is
|
|
variables[2] = variable3.model_copy(
|
|
update={
|
|
"name": "new name",
|
|
"value": HIDDEN_VALUE,
|
|
}
|
|
)
|
|
|
|
workflow.environment_variables = variables
|
|
assert workflow.environment_variables[2].name == "new name"
|
|
assert workflow.environment_variables[2].value == variable3.value
|
|
|
|
|
|
def test_to_dict():
|
|
contexts.tenant_id.set("tenant_id")
|
|
|
|
# Create a Workflow instance
|
|
workflow = Workflow(
|
|
tenant_id="tenant_id",
|
|
app_id="app_id",
|
|
type="workflow",
|
|
version="draft",
|
|
graph="{}",
|
|
features="{}",
|
|
created_by="account_id",
|
|
environment_variables=[],
|
|
conversation_variables=[],
|
|
)
|
|
|
|
# Create some EnvironmentVariable instances
|
|
|
|
with (
|
|
mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"),
|
|
mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"),
|
|
):
|
|
# Set the environment_variables property of the Workflow instance
|
|
workflow.environment_variables = [
|
|
SecretVariable.model_validate({"name": "secret", "value": "secret", "id": str(uuid4())}),
|
|
StringVariable.model_validate({"name": "text", "value": "text", "id": str(uuid4())}),
|
|
]
|
|
|
|
workflow_dict = workflow.to_dict()
|
|
assert workflow_dict["environment_variables"][0]["value"] == ""
|
|
assert workflow_dict["environment_variables"][1]["value"] == "text"
|
|
|
|
workflow_dict = workflow.to_dict(include_secret=True)
|
|
assert workflow_dict["environment_variables"][0]["value"] == "secret"
|
|
assert workflow_dict["environment_variables"][1]["value"] == "text"
|