test: use real trace records and SDK span values (#43150)

This commit is contained in:
Asuka Minato authored and GitHub committed 2026-09-29 07:08:47 +00:00
1 parent 8db1712ccd
commit 80b34b5ab6
9 files changed
+218 -140

No files matched your search

@@ -63,7 +63,7 @@ class TestTraceClient:
def test_export(self, mock_exporter_class: MagicMock, trace_client_factory: type[TraceClient]):
mock_exporter = mock_exporter_class.return_value
client = trace_client_factory(service_name="test-service", endpoint="http://test-endpoint")
spans = [MagicMock(spec=ReadableSpan)]
spans = [ReadableSpan(name="test-span")]
client.export(spans)
mock_exporter.export.assert_called_once_with(spans)
@@ -131,8 +131,8 @@ class TestTraceClient:
span_kind=SpanKind.INTERNAL,
)
mock_span = MagicMock(spec=ReadableSpan)
client.span_builder.build_span = MagicMock(return_value=mock_span)
span = ReadableSpan(name="test-span")
client.span_builder.build_span = MagicMock(return_value=span)
with patch.object(client.condition, "notify") as mock_notify:
client.add_span(span_data)
@@ -159,8 +159,8 @@ class TestTraceClient:
status=Status(StatusCode.OK),
span_kind=SpanKind.INTERNAL,
)
mock_span = MagicMock(spec=ReadableSpan)
client.span_builder.build_span = MagicMock(return_value=mock_span)
span = ReadableSpan(name="test-span")
client.span_builder.build_span = MagicMock(return_value=span)
client.add_span(span_data)
assert len(client.queue) == 1
@@ -178,8 +178,8 @@ class TestTraceClient:
mock_exporter.export.side_effect = Exception("Export failed")
client = trace_client_factory(service_name="test-service", endpoint="http://test-endpoint")
mock_span = MagicMock(spec=ReadableSpan)
client.queue.append(mock_span)
span = ReadableSpan(name="test-span")
client.queue.append(span)
with caplog.at_level(logging.WARNING):
client._export_batch()
@@ -207,8 +207,8 @@ class TestTraceClient:
mock_exporter = mock_exporter_class.return_value
client = trace_client_factory(service_name="test-service", endpoint="http://test-endpoint")
mock_span = MagicMock(spec=ReadableSpan)
client.queue.append(mock_span)
span = ReadableSpan(name="test-span")
client.queue.append(span)
client.shutdown()
# Should have called export twice (once in worker/export_batch, once in shutdown)
@@ -55,8 +55,8 @@ from core.ops.entities.trace_entity import (
ToolTraceInfo,
WorkflowTraceInfo,
)
from graphon.entities import WorkflowNodeExecution
from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
from tests.unit_tests.core.ops.trace_fixtures import workflow_node_execution
class RecordingTraceClient:
@@ -464,7 +464,7 @@ def test_get_workflow_node_executions_builds_repo_and_fetches(
def test_build_workflow_node_span_routes_llm_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -477,7 +477,7 @@ def test_build_workflow_node_span_routes_llm_type(trace_instance: AliyunDataTrac
def test_build_workflow_node_span_routes_knowledge_retrieval_type(
trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch
):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -488,7 +488,7 @@ def test_build_workflow_node_span_routes_knowledge_retrieval_type(
def test_build_workflow_node_span_routes_tool_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -499,7 +499,7 @@ def test_build_workflow_node_span_routes_tool_type(trace_instance: AliyunDataTra
def test_build_workflow_node_span_routes_code_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -512,7 +512,7 @@ def test_build_workflow_node_span_routes_code_type(trace_instance: AliyunDataTra
def test_build_workflow_node_span_handles_errors(
trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -541,7 +541,7 @@ def test_build_workflow_task_span(trace_instance: AliyunDataTrace, monkeypatch:
monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: status)
trace_metadata = _make_trace_metadata(links=[_make_link()])
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "title"
node_execution.node_type = node_type
@@ -573,7 +573,7 @@ def test_http_task_span_uses_request_snapshot(trace_instance: AliyunDataTrace, e
"Authorization: Bearer ******\r\nContent-Type: application/json\r\n\r\n"
'{"message":"synthetic 测试"}'
)
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "00000000-0000-0000-0000-000000000006"
node_execution.node_type = BuiltinNodeTypes.HTTP_REQUEST
node_execution.title = "HTTP request"
@@ -625,7 +625,7 @@ def test_http_task_span_uses_request_snapshot(trace_instance: AliyunDataTrace, e
)
@pytest.mark.parametrize("inputs", [None, {}, {"fallback": "original input"}])
def test_http_task_span_falls_back_to_inputs(trace_instance: AliyunDataTrace, process_data, inputs):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "00000000-0000-0000-0000-000000000006"
node_execution.node_type = BuiltinNodeTypes.HTTP_REQUEST
node_execution.title = "HTTP request"
@@ -645,7 +645,7 @@ def test_http_task_span_falls_back_to_inputs(trace_instance: AliyunDataTrace, pr
def test_non_http_task_span_ignores_request_snapshot(trace_instance: AliyunDataTrace):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "00000000-0000-0000-0000-000000000007"
node_execution.node_type = BuiltinNodeTypes.CODE
node_execution.title = "Code"
@@ -671,7 +671,7 @@ def test_build_workflow_tool_span(trace_instance: AliyunDataTrace, monkeypatch:
monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: status)
trace_metadata = _make_trace_metadata(links=[_make_link()])
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "my-tool"
node_execution.inputs = {"a": 1}
@@ -714,7 +714,7 @@ def test_build_workflow_retrieval_span(trace_instance: AliyunDataTrace, monkeypa
)
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "retrieval"
node_execution.inputs = {"query": "q"}
@@ -743,7 +743,7 @@ def test_build_workflow_llm_span(trace_instance: AliyunDataTrace, monkeypatch: p
monkeypatch.setattr(aliyun_trace_module, "format_output_messages", lambda _: "out")
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "llm"
node_execution.inputs = {}
@@ -783,7 +783,7 @@ def test_build_workflow_llm_span_falls_back_to_inputs_on_invoke_failure(
monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: Status(StatusCode.ERROR, "boom"))
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "llm"
node_execution.process_data = {}
@@ -873,7 +873,7 @@ def _make_agent_outputs() -> dict:
def test_build_workflow_node_span_routes_agent_type(trace_instance: AliyunDataTrace, monkeypatch: pytest.MonkeyPatch):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
trace_info = _make_workflow_trace_info()
trace_metadata = _make_trace_metadata()
@@ -890,7 +890,7 @@ def test_build_workflow_agent_span(trace_instance: AliyunDataTrace, monkeypatch:
monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: status)
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "my-agent"
node_execution.inputs = {"query": "ping"}
@@ -926,7 +926,7 @@ def test_build_agent_react_spans(trace_instance: AliyunDataTrace, monkeypatch: p
monkeypatch.setattr(aliyun_trace_module, "generate_span_id", lambda: next(span_ids))
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.outputs = _make_agent_outputs()
node_execution.created_at = _dt()
@@ -972,7 +972,7 @@ def test_build_agent_react_spans(trace_instance: AliyunDataTrace, monkeypatch: p
def test_build_agent_react_spans_returns_empty_without_log(trace_instance: AliyunDataTrace):
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.outputs = {"text": "t"}
node_execution.created_at = _dt()
@@ -988,9 +988,9 @@ def test_workflow_trace_adds_react_spans_for_agent_nodes(
monkeypatch.setattr(aliyun_trace_module, "convert_to_span_id", lambda _, __: 222)
monkeypatch.setattr(aliyun_trace_module, "create_links_from_trace_id", lambda _: [])
agent_node = MagicMock(spec=WorkflowNodeExecution)
agent_node = workflow_node_execution()
agent_node.node_type = BuiltinNodeTypes.AGENT
code_node = MagicMock(spec=WorkflowNodeExecution)
code_node = workflow_node_execution()
code_node.node_type = BuiltinNodeTypes.CODE
monkeypatch.setattr(trace_instance, "add_workflow_span", MagicMock())
@@ -1018,7 +1018,7 @@ def test_build_workflow_llm_span_records_time_to_first_token(
monkeypatch.setattr(aliyun_trace_module, "get_workflow_node_status", lambda _: Status(StatusCode.OK))
trace_metadata = _make_trace_metadata()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node-id"
node_execution.title = "llm"
node_execution.inputs = {}
@@ -37,14 +37,14 @@ from dify_trace_aliyun.utils import (
map_gen_ai_tool_type,
serialize_json_data,
)
from opentelemetry.trace import Link, StatusCode
from opentelemetry.trace import Link, SpanContext, StatusCode
from sqlalchemy.orm import Session
from core.rag.models.document import Document
from graphon.entities import WorkflowNodeExecution
from graphon.enums import WorkflowNodeExecutionStatus
from models import EndUser
from models.enums import EndUserType
from tests.unit_tests.core.ops.trace_fixtures import workflow_node_execution
def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPatch):
@@ -105,7 +105,7 @@ def test_create_status_from_error():
def test_get_workflow_node_status():
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
# SUCCEEDED
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -134,10 +134,10 @@ def test_get_workflow_node_status():
def test_create_links_from_trace_id(monkeypatch: pytest.MonkeyPatch):
# Mock create_link
mock_link = MagicMock(spec=Link)
link = Link(SpanContext(trace_id=1, span_id=2, is_remote=False))
import dify_trace_aliyun.data_exporter.traceclient
monkeypatch.setattr(dify_trace_aliyun.data_exporter.traceclient, "create_link", lambda trace_id_str: mock_link)
monkeypatch.setattr(dify_trace_aliyun.data_exporter.traceclient, "create_link", lambda trace_id_str: link)
# Trace ID None
assert create_links_from_trace_id(None) == []
@@ -145,17 +145,15 @@ def test_create_links_from_trace_id(monkeypatch: pytest.MonkeyPatch):
# Trace ID Present
links = create_links_from_trace_id("trace_id")
assert len(links) == 1
assert links[0] == mock_link
assert links[0] == link
def test_extract_retrieval_documents():
doc1 = MagicMock(spec=Document)
doc1.page_content = "content1"
doc1.metadata = {"dataset_id": "ds1", "doc_id": "di1", "document_id": "dd1", "score": 0.9}
doc2 = MagicMock(spec=Document)
doc2.page_content = "content2"
doc2.metadata = {"dataset_id": "ds2"} # Missing some keys
doc1 = Document(
page_content="content1",
metadata={"dataset_id": "ds1", "doc_id": "di1", "document_id": "dd1", "score": 0.9},
)
doc2 = Document(page_content="content2", metadata={"dataset_id": "ds2"}) # Missing some keys
documents = [doc1, doc2]
extracted = extract_retrieval_documents(documents)
@@ -38,6 +38,9 @@ def _dt() -> datetime:
return datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC)
from tests.unit_tests.core.ops.trace_fixtures import message_trace_info, tool_trace_info, workflow_trace_info
@pytest.fixture
def langfuse_config():
return LangfuseConfig(public_key="pk-123", secret_key="sk-123", host="https://cloud.langfuse.com")
@@ -118,37 +121,37 @@ def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(trace_instance, method, m)
# WorkflowTraceInfo
info = MagicMock(spec=WorkflowTraceInfo)
info = workflow_trace_info()
trace_instance.trace(info)
mocks["workflow_trace"].assert_called_once_with(info)
# MessageTraceInfo
info = MagicMock(spec=MessageTraceInfo)
info = message_trace_info()
trace_instance.trace(info)
mocks["message_trace"].assert_called_once_with(info)
# ModerationTraceInfo
info = MagicMock(spec=ModerationTraceInfo)
info = ModerationTraceInfo(metadata={}, flagged=False, action="allow", preset_response="", query="hello")
trace_instance.trace(info)
mocks["moderation_trace"].assert_called_once_with(info)
# SuggestedQuestionTraceInfo
info = MagicMock(spec=SuggestedQuestionTraceInfo)
info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
trace_instance.trace(info)
mocks["suggested_question_trace"].assert_called_once_with(info)
# DatasetRetrievalTraceInfo
info = MagicMock(spec=DatasetRetrievalTraceInfo)
info = DatasetRetrievalTraceInfo(metadata={})
trace_instance.trace(info)
mocks["dataset_retrieval_trace"].assert_called_once_with(info)
# ToolTraceInfo
info = MagicMock(spec=ToolTraceInfo)
info = tool_trace_info()
trace_instance.trace(info)
mocks["tool_trace"].assert_called_once_with(info)
# GenerateNameTraceInfo
info = MagicMock(spec=GenerateNameTraceInfo)
info = GenerateNameTraceInfo(tenant_id="tenant-1", metadata={})
trace_instance.trace(info)
mocks["generate_name_trace"].assert_called_once_with(info)
@@ -29,6 +29,7 @@ from core.ops.entities.trace_entity import (
from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
from models import EndUser
from models.enums import EndUserType
from tests.unit_tests.core.ops.trace_fixtures import message_trace_info, tool_trace_info, workflow_trace_info
def _dt() -> datetime:
@@ -88,37 +89,37 @@ def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(trace_instance, method, m)
# WorkflowTraceInfo
info = MagicMock(spec=WorkflowTraceInfo)
info = workflow_trace_info()
trace_instance.trace(info)
mocks["workflow_trace"].assert_called_once_with(info)
# MessageTraceInfo
info = MagicMock(spec=MessageTraceInfo)
info = message_trace_info()
trace_instance.trace(info)
mocks["message_trace"].assert_called_once_with(info)
# ModerationTraceInfo
info = MagicMock(spec=ModerationTraceInfo)
info = ModerationTraceInfo(metadata={}, flagged=False, action="allow", preset_response="", query="hello")
trace_instance.trace(info)
mocks["moderation_trace"].assert_called_once_with(info)
# SuggestedQuestionTraceInfo
info = MagicMock(spec=SuggestedQuestionTraceInfo)
info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
trace_instance.trace(info)
mocks["suggested_question_trace"].assert_called_once_with(info)
# DatasetRetrievalTraceInfo
info = MagicMock(spec=DatasetRetrievalTraceInfo)
info = DatasetRetrievalTraceInfo(metadata={})
trace_instance.trace(info)
mocks["dataset_retrieval_trace"].assert_called_once_with(info)
# ToolTraceInfo
info = MagicMock(spec=ToolTraceInfo)
info = tool_trace_info()
trace_instance.trace(info)
mocks["tool_trace"].assert_called_once_with(info)
# GenerateNameTraceInfo
info = MagicMock(spec=GenerateNameTraceInfo)
info = GenerateNameTraceInfo(tenant_id="tenant-1", metadata={})
trace_instance.trace(info)
mocks["generate_name_trace"].assert_called_once_with(info)
@@ -295,7 +296,7 @@ def test_workflow_trace_no_start_time(
def test_workflow_trace_missing_app_id(
trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
) -> None:
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.trace_id = "trace-1"
trace_info.message_id = None
trace_info.workflow_run_id = "run-1"
@@ -376,7 +377,7 @@ def test_message_trace(trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_
def test_message_trace_no_data(trace_instance):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.message_data = None
trace_info.file_list = []
trace_info.message_file_data = None
@@ -387,7 +388,7 @@ def test_message_trace_no_data(trace_instance):
def test_moderation_trace_no_data(trace_instance):
trace_info = MagicMock(spec=ModerationTraceInfo)
trace_info = ModerationTraceInfo(metadata={}, flagged=False, action="allow", preset_response="", query="hello")
trace_info.message_data = None
trace_instance.add_run = MagicMock()
trace_instance.moderation_trace(trace_info)
@@ -395,7 +396,7 @@ def test_moderation_trace_no_data(trace_instance):
def test_suggested_question_trace_no_data(trace_instance):
trace_info = MagicMock(spec=SuggestedQuestionTraceInfo)
trace_info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
trace_info.message_data = None
trace_instance.add_run = MagicMock()
trace_instance.suggested_question_trace(trace_info)
@@ -403,7 +404,7 @@ def test_suggested_question_trace_no_data(trace_instance):
def test_dataset_retrieval_trace_no_data(trace_instance):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_data = None
trace_instance.add_run = MagicMock()
trace_instance.dataset_retrieval_trace(trace_info)
@@ -31,6 +31,9 @@ def _dt() -> datetime:
return datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC)
from tests.unit_tests.core.ops.trace_fixtures import message_trace_info, tool_trace_info, workflow_trace_info
@pytest.fixture
def opik_config():
return OpikConfig(
@@ -101,37 +104,37 @@ def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(trace_instance, method, m)
# WorkflowTraceInfo
info = MagicMock(spec=WorkflowTraceInfo)
info = workflow_trace_info()
trace_instance.trace(info)
mocks["workflow_trace"].assert_called_once_with(info)
# MessageTraceInfo
info = MagicMock(spec=MessageTraceInfo)
info = message_trace_info()
trace_instance.trace(info)
mocks["message_trace"].assert_called_once_with(info)
# ModerationTraceInfo
info = MagicMock(spec=ModerationTraceInfo)
info = ModerationTraceInfo(metadata={}, flagged=False, action="allow", preset_response="", query="hello")
trace_instance.trace(info)
mocks["moderation_trace"].assert_called_once_with(info)
# SuggestedQuestionTraceInfo
info = MagicMock(spec=SuggestedQuestionTraceInfo)
info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
trace_instance.trace(info)
mocks["suggested_question_trace"].assert_called_once_with(info)
# DatasetRetrievalTraceInfo
info = MagicMock(spec=DatasetRetrievalTraceInfo)
info = DatasetRetrievalTraceInfo(metadata={})
trace_instance.trace(info)
mocks["dataset_retrieval_trace"].assert_called_once_with(info)
# ToolTraceInfo
info = MagicMock(spec=ToolTraceInfo)
info = tool_trace_info()
trace_instance.trace(info)
mocks["tool_trace"].assert_called_once_with(info)
# GenerateNameTraceInfo
info = MagicMock(spec=GenerateNameTraceInfo)
info = GenerateNameTraceInfo(tenant_id="tenant-1", metadata={})
trace_instance.trace(info)
mocks["generate_name_trace"].assert_called_once_with(info)
@@ -1,5 +1,5 @@
from datetime import datetime
from unittest.mock import MagicMock, patch
from unittest.mock import patch
from dify_trace_tencent.entities.semconv import (
GEN_AI_IS_ENTRY,
@@ -20,13 +20,15 @@ from opentelemetry.trace import StatusCode
from core.ops.entities.trace_entity import (
DatasetRetrievalTraceInfo,
MessageTraceInfo,
ToolTraceInfo,
WorkflowTraceInfo,
)
from core.rag.models.document import Document
from graphon.entities import WorkflowNodeExecution
from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
from tests.unit_tests.core.ops.trace_fixtures import (
message_trace_info,
tool_trace_info,
workflow_node_execution,
workflow_trace_info,
)
class TestTencentSpanBuilder:
@@ -39,7 +41,7 @@ class TestTencentSpanBuilder:
mock_convert.assert_called_once_with(dt)
def test_build_workflow_spans(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.workflow_run_id = "run_id"
trace_info.error = None
trace_info.start_time = datetime.now()
@@ -61,7 +63,7 @@ class TestTencentSpanBuilder:
assert spans[1].parent_span_id == 2
def test_build_workflow_spans_no_message(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.workflow_run_id = "run_id"
trace_info.error = "some error"
trace_info.start_time = datetime.now()
@@ -82,10 +84,10 @@ class TestTencentSpanBuilder:
assert spans[0].attributes[GEN_AI_IS_ENTRY] == "true"
def test_build_workflow_llm_span(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {"conversation_id": "conv_id"}
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.created_at = datetime.now()
node_execution.finished_at = datetime.now()
@@ -109,10 +111,10 @@ class TestTencentSpanBuilder:
assert span.attributes[GEN_AI_USAGE_INPUT_TOKENS] == "10"
def test_build_workflow_llm_span_usage_in_outputs(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {}
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.created_at = datetime.now()
node_execution.finished_at = datetime.now()
@@ -132,7 +134,7 @@ class TestTencentSpanBuilder:
assert GEN_AI_IS_STREAMING_REQUEST not in span.attributes
def test_build_message_span_standalone(self):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.message_id = "msg_id"
trace_info.error = None
trace_info.start_time = datetime.now()
@@ -152,7 +154,7 @@ class TestTencentSpanBuilder:
assert span.attributes[INPUT_VALUE] == str(trace_info.inputs)
def test_build_message_span_standalone_with_error(self):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.message_id = "msg_id"
trace_info.error = "some error"
trace_info.start_time = datetime.now()
@@ -172,7 +174,7 @@ class TestTencentSpanBuilder:
assert span.attributes[INPUT_VALUE] == ""
def test_build_tool_span(self):
trace_info = MagicMock(spec=ToolTraceInfo)
trace_info = tool_trace_info()
trace_info.message_id = "msg_id"
trace_info.tool_name = "search"
trace_info.error = "tool error"
@@ -192,7 +194,7 @@ class TestTencentSpanBuilder:
assert span.attributes[TOOL_NAME] == "search"
def test_build_retrieval_span(self):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_id = "msg_id"
trace_info.inputs = "query"
trace_info.error = None
@@ -214,7 +216,7 @@ class TestTencentSpanBuilder:
assert "content" in span.attributes[RETRIEVAL_DOCUMENT]
def test_build_retrieval_span_with_error(self):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_id = "msg_id"
trace_info.inputs = ""
trace_info.error = "retrieval failed"
@@ -231,7 +233,7 @@ class TestTencentSpanBuilder:
assert span.status.description == "retrieval failed"
def test_get_workflow_node_status(self):
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.status = WorkflowNodeExecutionStatus.SUCCEEDED
assert TencentSpanBuilder._get_workflow_node_status(node).status_code == StatusCode.OK
@@ -252,10 +254,10 @@ class TestTencentSpanBuilder:
assert TencentSpanBuilder._get_workflow_node_status(node).status_code == StatusCode.UNSET
def test_build_workflow_retrieval_span(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {"conversation_id": "conv_id"}
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.title = "my retrieval"
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -274,10 +276,10 @@ class TestTencentSpanBuilder:
assert "c1" in span.attributes[RETRIEVAL_DOCUMENT]
def test_build_workflow_retrieval_span_empty(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {}
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.title = "my retrieval"
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -295,9 +297,9 @@ class TestTencentSpanBuilder:
assert span.attributes[RETRIEVAL_DOCUMENT] == ""
def test_build_workflow_tool_span(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.title = "my tool"
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -317,9 +319,9 @@ class TestTencentSpanBuilder:
assert "some" in span.attributes[TOOL_DESCRIPTION]
def test_build_workflow_tool_span_no_metadata(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.title = "my tool"
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -338,10 +340,10 @@ class TestTencentSpanBuilder:
assert span.attributes[TOOL_PARAMETERS] == "{}"
def test_build_workflow_task_span(self):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {"conversation_id": "conv_id"}
node_execution = MagicMock(spec=WorkflowNodeExecution)
node_execution = workflow_node_execution()
node_execution.id = "node_id"
node_execution.title = "my task"
node_execution.status = WorkflowNodeExecutionStatus.SUCCEEDED
@@ -20,11 +20,16 @@ from core.ops.entities.trace_entity import (
ToolTraceInfo,
WorkflowTraceInfo,
)
from graphon.entities import WorkflowNodeExecution
from graphon.enums import BuiltinNodeTypes
from models import Account, App, Tenant, TenantAccountJoin
from models.account import TenantAccountRole
from models.model import AppMode, IconType
from tests.unit_tests.core.ops.trace_fixtures import (
message_trace_info,
tool_trace_info,
workflow_node_execution,
workflow_trace_info,
)
logger = logging.getLogger(__name__)
@@ -180,7 +185,7 @@ class TestTencentDataTrace:
tencent_data_trace.trace_client.get_project_url.assert_called_once()
def test_workflow_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.workflow_run_id = "run-id"
trace_info.trace_id = "parent-trace-id"
@@ -204,7 +209,7 @@ class TestTencentDataTrace:
mock_dur.assert_called_once_with(trace_info)
def test_workflow_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.workflow_run_id = "run-id"
with (
@@ -217,7 +222,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process workflow trace" in caplog.text
def test_message_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.message_id = "msg-id"
trace_info.trace_id = "parent-trace-id"
@@ -241,7 +246,7 @@ class TestTencentDataTrace:
mock_dur.assert_called_once_with(trace_info)
def test_message_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
with (
patch(
@@ -253,7 +258,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process message trace" in caplog.text
def test_tool_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
trace_info = MagicMock(spec=ToolTraceInfo)
trace_info = tool_trace_info()
trace_info.message_id = "msg-id"
mock_trace_utils.convert_to_span_id.return_value = 456
@@ -267,14 +272,14 @@ class TestTencentDataTrace:
tencent_data_trace.trace_client.add_span.assert_called_once()
def test_tool_trace_no_msg_id(self, tencent_data_trace):
trace_info = MagicMock(spec=ToolTraceInfo)
trace_info = tool_trace_info()
trace_info.message_id = None
tencent_data_trace.tool_trace(trace_info)
tencent_data_trace.trace_client.add_span.assert_not_called()
def test_tool_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=ToolTraceInfo)
trace_info = tool_trace_info()
trace_info.message_id = "msg-id"
with (
@@ -287,7 +292,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process tool trace" in caplog.text
def test_dataset_retrieval_trace(self, tencent_data_trace, mock_trace_utils, mock_span_builder):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_id = "msg-id"
mock_trace_utils.convert_to_span_id.return_value = 456
@@ -301,14 +306,14 @@ class TestTencentDataTrace:
tencent_data_trace.trace_client.add_span.assert_called_once()
def test_dataset_retrieval_trace_no_msg_id(self, tencent_data_trace):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_id = None
tencent_data_trace.dataset_retrieval_trace(trace_info)
tencent_data_trace.trace_client.add_span.assert_not_called()
def test_dataset_retrieval_trace_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=DatasetRetrievalTraceInfo)
trace_info = DatasetRetrievalTraceInfo(metadata={})
trace_info.message_id = "msg-id"
with (
@@ -321,7 +326,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process dataset retrieval trace" in caplog.text
def test_suggested_question_trace(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=SuggestedQuestionTraceInfo)
trace_info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
with caplog.at_level(logging.INFO):
tencent_data_trace.suggested_question_trace(trace_info)
assert "[Tencent APM] Processing suggested question trace" in caplog.text
@@ -329,7 +334,7 @@ class TestTencentDataTrace:
def test_suggested_question_trace_exception(
self, tencent_data_trace, monkeypatch, caplog: pytest.LogCaptureFixture
):
trace_info = MagicMock(spec=SuggestedQuestionTraceInfo)
trace_info = SuggestedQuestionTraceInfo(metadata={}, total_tokens=0, suggested_question=[], level="info")
target_logger = logging.getLogger("dify_trace_tencent.tencent_trace")
monkeypatch.setattr(target_logger, "info", MagicMock(side_effect=Exception("error")))
with caplog.at_level(logging.ERROR):
@@ -337,14 +342,14 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process suggested question trace" in caplog.text
def test_process_workflow_nodes(self, tencent_data_trace, mock_trace_utils):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.workflow_run_id = "run-id"
mock_trace_utils.convert_to_span_id.return_value = 111
node1 = MagicMock(spec=WorkflowNodeExecution)
node1 = workflow_node_execution()
node1.id = "n1"
node1.node_type = BuiltinNodeTypes.LLM
node2 = MagicMock(spec=WorkflowNodeExecution)
node2 = workflow_node_execution()
node2.id = "n2"
node2.node_type = BuiltinNodeTypes.TOOL
@@ -361,24 +366,29 @@ class TestTencentDataTrace:
def test_process_workflow_nodes_node_exception(
self, tencent_data_trace, mock_trace_utils, caplog: pytest.LogCaptureFixture
):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
mock_trace_utils.convert_to_span_id.return_value = 111
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.id = "n1"
next_node = workflow_node_execution()
next_node.id = "n2"
with (
patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node]),
patch.object(tencent_data_trace, "_build_workflow_node_span", side_effect=Exception("node error")),
patch.object(tencent_data_trace, "_get_workflow_node_executions", return_value=[node, next_node]),
patch.object(
tencent_data_trace, "_build_workflow_node_span", side_effect=[Exception("node error"), "next-span"]
),
caplog.at_level(logging.ERROR),
):
tencent_data_trace._process_workflow_nodes(trace_info, 123)
assert "[Tencent APM] Failed to process workflow nodes" in caplog.text
assert "[Tencent APM] Failed to process node execution: n1" in caplog.text
tencent_data_trace.trace_client.add_span.assert_called_once_with("next-span")
def test_process_workflow_nodes_exception(
self, tencent_data_trace, mock_trace_utils, caplog: pytest.LogCaptureFixture
):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
mock_trace_utils.convert_to_span_id.side_effect = Exception("outer error")
with caplog.at_level(logging.ERROR):
@@ -386,7 +396,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to process workflow nodes" in caplog.text
def test_build_workflow_node_span(self, tencent_data_trace, mock_span_builder):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
nodes = [
(BuiltinNodeTypes.LLM, mock_span_builder.build_workflow_llm_span),
@@ -396,7 +406,7 @@ class TestTencentDataTrace:
]
for node_type, builder_method in nodes:
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.node_type = node_type
builder_method.return_value = "span"
@@ -408,7 +418,7 @@ class TestTencentDataTrace:
def test_build_workflow_node_span_exception(
self, tencent_data_trace, mock_span_builder, caplog: pytest.LogCaptureFixture
):
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.node_type = BuiltinNodeTypes.LLM
node.id = "n1"
mock_span_builder.build_workflow_llm_span.side_effect = Exception("error")
@@ -456,7 +466,7 @@ class TestTencentDataTrace:
sqlite3_session.add_all([app, tenant_join])
sqlite3_session.commit()
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {"app_id": app.id}
trace_info.workflow_run_id = "run-1"
database = SimpleNamespace(engine=sqlite3_session.get_bind())
@@ -480,7 +490,7 @@ class TestTencentDataTrace:
monkeypatch: pytest.MonkeyPatch,
sqlite3_session: Session,
) -> None:
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {}
monkeypatch.setattr(
"dify_trace_tencent.tencent_trace.db",
@@ -500,7 +510,7 @@ class TestTencentDataTrace:
monkeypatch: pytest.MonkeyPatch,
sqlite3_session: Session,
) -> None:
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.metadata = {"app_id": "app-1"}
monkeypatch.setattr(
"dify_trace_tencent.tencent_trace.db",
@@ -513,7 +523,7 @@ class TestTencentDataTrace:
assert len([r for r in caplog.records if r.levelno == logging.ERROR]) >= 1
def test_get_user_id_workflow(self, tencent_data_trace):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.tenant_id = "tenant-1"
trace_info.metadata = {"user_id": "user-1"}
@@ -528,21 +538,21 @@ class TestTencentDataTrace:
assert user_id == "unknown"
def test_get_user_id_only_user_id(self, tencent_data_trace):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.metadata = {"user_id": "user-1"}
user_id = tencent_data_trace._get_user_id(trace_info)
assert user_id == "user-1"
def test_get_user_id_anonymous(self, tencent_data_trace):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.metadata = {}
user_id = tencent_data_trace._get_user_id(trace_info)
assert user_id == "anonymous"
def test_get_user_id_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.tenant_id = "t"
trace_info.metadata = {"user_id": "u"}
@@ -555,7 +565,7 @@ class TestTencentDataTrace:
assert "[Tencent APM] Failed to get user ID" in caplog.text
def test_record_llm_metrics_usage_in_process_data(self, tencent_data_trace):
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.process_data = {
"usage": {
"latency": 2.5,
@@ -578,7 +588,7 @@ class TestTencentDataTrace:
assert tencent_data_trace.trace_client.record_token_usage.call_count == 2
def test_record_llm_metrics_usage_in_outputs(self, tencent_data_trace):
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.process_data = {}
node.outputs = {"usage": {"latency": 1.0, "prompt_tokens": 5}}
@@ -587,7 +597,7 @@ class TestTencentDataTrace:
tencent_data_trace.trace_client.record_token_usage.assert_called_once()
def test_record_llm_metrics_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
node = MagicMock(spec=WorkflowNodeExecution)
node = workflow_node_execution()
node.process_data = None
node.outputs = None
@@ -596,7 +606,7 @@ class TestTencentDataTrace:
# Should not crash
def test_record_message_llm_metrics(self, tencent_data_trace):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.metadata = {"ls_provider": "openai", "ls_model_name": "gpt-4"}
trace_info.message_data = {"provider_response_latency": 1.1}
trace_info.is_streaming_request = True
@@ -613,7 +623,7 @@ class TestTencentDataTrace:
assert tencent_data_trace.trace_client.record_token_usage.call_count == 2
def test_record_message_llm_metrics_object_data(self, tencent_data_trace):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.metadata = {}
msg_data = MagicMock()
msg_data.provider_response_latency = 1.1
@@ -626,15 +636,16 @@ class TestTencentDataTrace:
tencent_data_trace.trace_client.record_llm_duration.assert_called_once()
def test_record_message_llm_metrics_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info.metadata = None
trace_info = message_trace_info()
# Inject malformed metadata to exercise the metrics error handler.
trace_info = trace_info.model_copy(update={"metadata": None})
with caplog.at_level(logging.DEBUG):
tencent_data_trace._record_message_llm_metrics(trace_info)
# Should not crash
def test_record_workflow_trace_duration(self, tencent_data_trace):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
from datetime import datetime, timedelta
now = datetime.now()
@@ -662,7 +673,7 @@ class TestTencentDataTrace:
assert attributes["has_conversation"] == "true"
def test_record_workflow_trace_duration_fallback(self, tencent_data_trace):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.start_time = None
trace_info.workflow_run_elapsed_time = 4.5
trace_info.workflow_run_status = "failed"
@@ -678,14 +689,14 @@ class TestTencentDataTrace:
assert attributes["has_conversation"] == "false"
def test_record_workflow_trace_duration_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=WorkflowTraceInfo)
trace_info = workflow_trace_info()
trace_info.start_time = MagicMock() # This might cause total_seconds() to fail if not mocked right
with caplog.at_level(logging.DEBUG):
tencent_data_trace._record_workflow_trace_duration(trace_info)
def test_record_message_trace_duration(self, tencent_data_trace):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
from datetime import datetime, timedelta
now = datetime.now()
@@ -700,7 +711,7 @@ class TestTencentDataTrace:
)
def test_record_message_trace_duration_exception(self, tencent_data_trace, caplog: pytest.LogCaptureFixture):
trace_info = MagicMock(spec=MessageTraceInfo)
trace_info = message_trace_info()
trace_info.start_time = None
with caplog.at_level(logging.DEBUG):
@@ -0,0 +1,60 @@
"""Validated trace records with baseline values for provider unit tests."""
from datetime import datetime
from core.ops.entities.trace_entity import MessageTraceInfo, ToolTraceInfo, WorkflowTraceInfo
from graphon.entities import WorkflowNodeExecution
from graphon.enums import BuiltinNodeTypes
def workflow_trace_info() -> WorkflowTraceInfo:
return WorkflowTraceInfo(
workflow_id="workflow-1",
tenant_id="tenant-1",
workflow_run_id="run-1",
workflow_run_elapsed_time=1.0,
workflow_run_status="succeeded",
workflow_run_inputs={},
workflow_run_outputs={},
workflow_run_version="1",
total_tokens=0,
file_list=[],
query="",
metadata={},
)
def message_trace_info() -> MessageTraceInfo:
return MessageTraceInfo(
message_id="message-1",
conversation_model="gpt-4",
conversation_mode="chat",
message_tokens=0,
answer_tokens=0,
total_tokens=0,
metadata={},
)
def tool_trace_info() -> ToolTraceInfo:
return ToolTraceInfo(
tool_name="search",
tool_inputs={},
tool_outputs="",
tool_config={},
tool_parameters={},
time_cost=0,
metadata={},
)
def workflow_node_execution() -> WorkflowNodeExecution:
return WorkflowNodeExecution(
id="execution-1",
workflow_id="workflow-1",
index=1,
node_id="node-1",
node_type=BuiltinNodeTypes.LLM,
title="LLM",
created_at=datetime(2026, 1, 1),
)