opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-28 23:59:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use SQLite in Aliyun trace utilities (#38983)

Browse Source
This commit is contained in:
Asuka Minato 2026-07-28 14:40:27 +09:00 committed by GitHub
parent ea58129ebe
commit 80ff108fc0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
api/providers/trace/trace-aliyun/tests/unit_tests/aliyun_trace/test_aliyun_trace_utils.py
View File

@ -1,3 +1,5 @@
"""Unit tests for Aliyun trace utility transformations and database lookups."""
import json
from collections.abc import Mapping
from typing import Any, cast
@ -25,11 +27,13 @@ from dify_trace_aliyun.utils import (
serialize_json_data,
)
from opentelemetry.trace import Link, 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
def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPatch):
@ -40,35 +44,40 @@ def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPat
assert get_user_id_from_message_data(message_data) == "account_id"
def test_get_user_id_from_message_data_with_end_user(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize("sqlite3_session", [(EndUser,)], indirect=True)
def test_get_user_id_from_message_data_with_end_user(monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session) -> None:
message_data = MagicMock()
message_data.from_account_id = "account_id"
message_data.from_end_user_id = "end_user_id"
end_user_data = MagicMock(spec=EndUser)
end_user_data.session_id = "session_id"
mock_session = MagicMock()
mock_session.get.return_value = end_user_data
end_user_data = EndUser(
id="end_user_id",
tenant_id="tenant_id",
app_id="app_id",
type=EndUserType.BROWSER,
session_id="session_id",
)
sqlite3_session.add(end_user_data)
sqlite3_session.commit()
from dify_trace_aliyun.utils import db
monkeypatch.setattr(db, "session", mock_session)
monkeypatch.setattr(db, "session", sqlite3_session)
assert get_user_id_from_message_data(message_data) == "session_id"
def test_get_user_id_from_message_data_end_user_not_found(monkeypatch: pytest.MonkeyPatch):
@pytest.mark.parametrize("sqlite3_session", [(EndUser,)], indirect=True)
def test_get_user_id_from_message_data_end_user_not_found(
monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
) -> None:
message_data = MagicMock()
message_data.from_account_id = "account_id"
message_data.from_end_user_id = "end_user_id"
mock_session = MagicMock()
mock_session.get.return_value = None
from dify_trace_aliyun.utils import db
monkeypatch.setattr(db, "session", mock_session)
monkeypatch.setattr(db, "session", sqlite3_session)
assert get_user_id_from_message_data(message_data) == "account_id"