Files
dify/api/tests/unit_tests/services/dify_builder/test_resources.py
T
Charles YaoandClaude Opus 4.8 a5f8d71eb7 fix(dify-builder): dataset visibility + normalization guard + test coverage
FINDING 1 (dataset visibility): DatasetService.get_datasets requires user
parameter and silently omits ONLY_ME/PARTIAL_TEAM datasets on RBAC tenants.
Query Dataset model directly by tenant_id for complete grounding inventory.

FINDING 2 (normalization guard): Move fetch+normalize inside _safe to prevent
malformed records from raising after _safe returns. Use lambda builders to wrap
complete resource transformation including record normalization.

FINDING 3 (test coverage): Add test_mixed_failure_partial_success (one source
fails, others succeed) and test_malformed_tool_record_degrades (missing
tool_name key gracefully degrades).

Minor: compute model id once and reuse for both id and label fields; ensure
E302 two blank lines between test functions.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-29 11:24:56 -07:00

59 lines
2.7 KiB
Python

from unittest.mock import patch
def test_list_tenant_resources_normalizes_all_three_kinds():
from services.dify_builder.agent import resources
class _Model:
def __init__(self, m): self.model = m
class _ProvResp:
def __init__(self, p, ms): self.provider, self.models = p, [_Model(m) for m in ms]
class _DS:
def __init__(self, i, n): self.id, self.name = i, n
with patch.object(resources, "_list_models", return_value=[_ProvResp("anthropic", ["claude-opus-4-8"])]), \
patch.object(resources, "_list_datasets", return_value=[_DS("kb-1", "Company KB")]), \
patch.object(resources, "_list_tools", return_value=[{"provider_name": "google", "tool_name": "search", "tool_label": "Google Search", "description": ""}]):
out = resources.list_tenant_resources("t1")
assert out.models[0].id == "anthropic/claude-opus-4-8"
assert out.datasets[0].id == "kb-1" and out.datasets[0].label == "Company KB"
assert out.tools[0].id == "google/search" and out.tools[0].label == "Google Search"
def test_each_source_degrades_to_empty():
from services.dify_builder.agent import resources
with patch.object(resources, "_list_models", side_effect=RuntimeError("boom")), \
patch.object(resources, "_list_datasets", side_effect=RuntimeError("boom")), \
patch.object(resources, "_list_tools", side_effect=RuntimeError("boom")):
out = resources.list_tenant_resources("t1")
assert out == resources.TenantResources(models=[], datasets=[], tools=[])
def test_mixed_failure_partial_success():
from services.dify_builder.agent import resources
class _DS:
def __init__(self, i, n): self.id, self.name = i, n
with patch.object(resources, "_list_models", side_effect=RuntimeError("boom")), \
patch.object(resources, "_list_datasets", return_value=[_DS("kb-1", "Company KB")]), \
patch.object(resources, "_list_tools", return_value=[{"provider_name": "google", "tool_name": "search", "tool_label": "Google Search", "description": ""}]):
out = resources.list_tenant_resources("t1")
assert out.models == []
assert len(out.datasets) == 1 and out.datasets[0].id == "kb-1"
assert len(out.tools) == 1 and out.tools[0].id == "google/search"
def test_malformed_tool_record_degrades():
from services.dify_builder.agent import resources
with patch.object(resources, "_list_models", return_value=[]), \
patch.object(resources, "_list_datasets", return_value=[]), \
patch.object(resources, "_list_tools", return_value=[{"provider_name": "g"}]): # missing tool_name
out = resources.list_tenant_resources("t1")
assert out.tools == []
assert out.models == [] and out.datasets == []