test_file.py 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657
  1. import json
  2. from core.file import File, FileTransferMethod, FileType, FileUploadConfig
  3. from models.workflow import Workflow
  4. def test_file_to_dict():
  5. file = File(
  6. id="file1",
  7. tenant_id="tenant1",
  8. type=FileType.IMAGE,
  9. transfer_method=FileTransferMethod.REMOTE_URL,
  10. remote_url="https://example.com/image1.jpg",
  11. storage_key="storage_key",
  12. )
  13. file_dict = file.to_dict()
  14. assert "_storage_key" not in file_dict
  15. assert "url" in file_dict
  16. def test_workflow_features_with_image():
  17. # Create a feature dict that mimics the old structure with image config
  18. features = {
  19. "file_upload": {
  20. "image": {"enabled": True, "number_limits": 5, "transfer_methods": ["remote_url", "local_file"]}
  21. }
  22. }
  23. # Create a workflow instance with the features
  24. workflow = Workflow(
  25. tenant_id="tenant-1",
  26. app_id="app-1",
  27. type="chat",
  28. version="1.0",
  29. graph="{}",
  30. features=json.dumps(features),
  31. created_by="user-1",
  32. environment_variables=[],
  33. conversation_variables=[],
  34. )
  35. # Get the converted features through the property
  36. converted_features = json.loads(workflow.features)
  37. # Create FileUploadConfig from the converted features
  38. file_upload_config = FileUploadConfig.model_validate(converted_features["file_upload"])
  39. # Validate the config
  40. assert file_upload_config.number_limits == 5
  41. assert list(file_upload_config.allowed_file_types) == [FileType.IMAGE]
  42. assert list(file_upload_config.allowed_file_upload_methods) == [
  43. FileTransferMethod.REMOTE_URL,
  44. FileTransferMethod.LOCAL_FILE,
  45. ]
  46. assert list(file_upload_config.allowed_file_extensions) == []