|
from unittest import mock |
|
from uuid import uuid4 |
|
|
|
import contexts |
|
from constants import HIDDEN_VALUE |
|
from core.variables import FloatVariable, IntegerVariable, SecretVariable, StringVariable |
|
from models.workflow import Workflow |
|
|
|
|
|
def test_environment_variables(): |
|
contexts.tenant_id.set("tenant_id") |
|
|
|
|
|
workflow = Workflow( |
|
tenant_id="tenant_id", |
|
app_id="app_id", |
|
type="workflow", |
|
version="draft", |
|
graph="{}", |
|
features="{}", |
|
created_by="account_id", |
|
environment_variables=[], |
|
conversation_variables=[], |
|
) |
|
|
|
|
|
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] |
|
workflow.environment_variables = variables |
|
|
|
|
|
assert workflow.environment_variables == variables |
|
|
|
|
|
def test_update_environment_variables(): |
|
contexts.tenant_id.set("tenant_id") |
|
|
|
|
|
workflow = Workflow( |
|
tenant_id="tenant_id", |
|
app_id="app_id", |
|
type="workflow", |
|
version="draft", |
|
graph="{}", |
|
features="{}", |
|
created_by="account_id", |
|
environment_variables=[], |
|
conversation_variables=[], |
|
) |
|
|
|
|
|
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] |
|
|
|
|
|
workflow.environment_variables = variables |
|
assert workflow.environment_variables == [variable1, variable2, variable3, variable4] |
|
|
|
|
|
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") |
|
|
|
|
|
workflow = Workflow( |
|
tenant_id="tenant_id", |
|
app_id="app_id", |
|
type="workflow", |
|
version="draft", |
|
graph="{}", |
|
features="{}", |
|
created_by="account_id", |
|
environment_variables=[], |
|
conversation_variables=[], |
|
) |
|
|
|
|
|
|
|
with ( |
|
mock.patch("core.helper.encrypter.encrypt_token", return_value="encrypted_token"), |
|
mock.patch("core.helper.encrypter.decrypt_token", return_value="secret"), |
|
): |
|
|
|
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" |
|
|