opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-06-23 04:11:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

test: use caplog for annotation service logging (#37618)

Browse Source
This commit is contained in:
frank 2026-06-18 15:54:47 +08:00 committed by GitHub
parent 762321751c
commit 694bf3754c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
api/tests/unit_tests/services/test_annotation_service.py
View File

@ -2,6 +2,7 @@
Unit tests for services.annotation_service
"""
import logging
from io import BytesIO
from types import SimpleNamespace
from typing import Any, cast
@ -1049,7 +1050,9 @@ class TestAppAnnotationServiceBatchImport:
mock_redis.setnx.assert_called_once_with("app_annotation_batch_import_uuid-3", "waiting")
mock_task.delay.assert_called_once()
def test_batch_import_app_annotations_should_cleanup_active_job_on_unexpected_exception(self) -> None:
def test_batch_import_app_annotations_should_cleanup_active_job_on_unexpected_exception(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Test unexpected runtime errors trigger cleanup and return wrapped error."""
# Arrange
file = _make_file(b"question,answer\nq,a\n")
@ -1067,7 +1070,6 @@ class TestAppAnnotationServiceBatchImport:
patch("services.annotation_service.redis_client") as mock_redis,
patch("services.annotation_service.uuid.uuid4", return_value="uuid-4"),
patch("services.annotation_service.naive_utc_now", return_value=SimpleNamespace(timestamp=lambda: 1)),
patch("services.annotation_service.logger") as mock_logger,
patch(
"configs.dify_config",
new=SimpleNamespace(ANNOTATION_IMPORT_MAX_RECORDS=5, ANNOTATION_IMPORT_MIN_RECORDS=1),
@ -1078,12 +1080,15 @@ class TestAppAnnotationServiceBatchImport:
mock_redis.zrem.side_effect = RuntimeError("cleanup-failed")
# Act
result = AppAnnotationService.batch_import_app_annotations(app.id, file)
with caplog.at_level(logging.DEBUG):
result = AppAnnotationService.batch_import_app_annotations(app.id, file)
# Assert
assert result["error_msg"] == "An error occurred while processing the file: boom"
mock_redis.zrem.assert_called_once_with(f"annotation_import_active:{tenant_id}", "uuid-4")
mock_logger.debug.assert_called_once()
assert len(caplog.records) == 1
assert caplog.records[0].levelname == "DEBUG"
assert "Failed to clean up active job tracking during error handling" in caplog.records[0].message
class TestAppAnnotationServiceHitHistoryAndSettings: