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

chore: example use caplog in test (#37470)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Asuka Minato 2026-06-16 10:10:09 +09:00 committed by GitHub
parent 7bed801b0d
commit 2be8a5f633
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 74 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

107
api/tests/test_containers_integration_tests/services/test_end_user_service.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
import logging
from unittest.mock import patch
from uuid import uuid4
@ -104,7 +105,9 @@ class TestEndUserServiceGetOrCreateEndUser:
"""Provide test data factory."""
return TestEndUserServiceFactory()
def test_get_or_create_end_user_with_custom_user_id(self, db_session_with_containers: Session, factory):
def test_get_or_create_end_user_with_custom_user_id(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test getting or creating end user with custom user_id."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -120,7 +123,9 @@ class TestEndUserServiceGetOrCreateEndUser:
assert result.type == InvokeFrom.SERVICE_API
assert result.is_anonymous is False
def test_get_or_create_end_user_without_user_id(self, db_session_with_containers: Session, factory):
def test_get_or_create_end_user_without_user_id(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test getting or creating end user without user_id uses default session."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -133,7 +138,7 @@ class TestEndUserServiceGetOrCreateEndUser:
# Verify _is_anonymous is set correctly (property always returns False)
assert result._is_anonymous is True
def test_get_existing_end_user(self, db_session_with_containers: Session, factory):
def test_get_existing_end_user(self, db_session_with_containers: Session, factory: TestEndUserServiceFactory):
"""Test retrieving an existing end user."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -169,7 +174,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
"""Provide test data factory."""
return TestEndUserServiceFactory()
def test_create_end_user_service_api_type(self, db_session_with_containers: Session, factory):
def test_create_end_user_service_api_type(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test creating new end user with SERVICE_API type."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -191,7 +198,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
assert result.app_id == app_id
assert result.session_id == user_id
def test_create_end_user_web_app_type(self, db_session_with_containers: Session, factory):
def test_create_end_user_web_app_type(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test creating new end user with WEB_APP type."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -210,8 +219,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
# Assert
assert result.type == InvokeFrom.WEB_APP
@patch("services.end_user_service.logger")
def test_upgrade_legacy_end_user_type(self, mock_logger, db_session_with_containers: Session, factory):
def test_upgrade_legacy_end_user_type(
self, caplog: pytest.LogCaptureFixture, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test upgrading legacy end user with different type."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -227,25 +237,32 @@ class TestEndUserServiceGetOrCreateEndUserByType:
session_id=user_id,
invoke_type=InvokeFrom.SERVICE_API,
)
# Act - Request with different type
result = EndUserService.get_or_create_end_user_by_type(
type=InvokeFrom.WEB_APP,
tenant_id=tenant_id,
app_id=app_id,
user_id=user_id,
)
with caplog.at_level(logging.INFO, logger="services.end_user_service"):
# Act - Request with different type
result = EndUserService.get_or_create_end_user_by_type(
type=InvokeFrom.WEB_APP,
tenant_id=tenant_id,
app_id=app_id,
user_id=user_id,
)
# Assert
assert result.id == existing_user.id
assert result.type == InvokeFrom.WEB_APP # Type should be updated
mock_logger.info.assert_called_once()
# Verify log message contains upgrade info
log_call = mock_logger.info.call_args[0][0]
assert "Upgrading legacy EndUser" in log_call
matching_logs = [
record
for record in caplog.records
if record.name == "services.end_user_service"
and record.levelno == logging.INFO
and "Upgrading legacy EndUser" in record.message
]
assert len(matching_logs) == 1
@patch("services.end_user_service.logger")
def test_get_existing_end_user_matching_type(self, mock_logger, db_session_with_containers: Session, factory):
def test_get_existing_end_user_matching_type(
self, mock_logger, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test retrieving existing end user with matching type."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -274,7 +291,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
assert result.type == InvokeFrom.SERVICE_API
mock_logger.info.assert_not_called()
def test_create_anonymous_user_with_default_session(self, db_session_with_containers: Session, factory):
def test_create_anonymous_user_with_default_session(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test creating anonymous user when user_id is None."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -295,7 +314,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
assert result._is_anonymous is True
assert result.external_user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
def test_query_ordering_prioritizes_matching_type(self, db_session_with_containers: Session, factory):
def test_query_ordering_prioritizes_matching_type(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test that query ordering prioritizes records with matching type."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -330,7 +351,9 @@ class TestEndUserServiceGetOrCreateEndUserByType:
assert result.id == matching.id
assert result.id != non_matching.id
def test_external_user_id_matches_session_id(self, db_session_with_containers: Session, factory):
def test_external_user_id_matches_session_id(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
"""Test that external_user_id is set to match session_id."""
# Arrange
app = factory.create_app_and_account(db_session_with_containers)
@ -360,7 +383,7 @@ class TestEndUserServiceGetOrCreateEndUserByType:
],
)
def test_create_end_user_with_different_invoke_types(
self, db_session_with_containers: Session, invoke_type, factory
self, db_session_with_containers: Session, invoke_type: InvokeFrom, factory: TestEndUserServiceFactory
):
"""Test creating end users with different InvokeFrom types."""
# Arrange
@ -389,7 +412,9 @@ class TestEndUserServiceGetEndUserById:
"""Provide test data factory."""
return TestEndUserServiceFactory()
def test_get_end_user_by_id_returns_end_user(self, db_session_with_containers: Session, factory):
def test_get_end_user_by_id_returns_end_user(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
app = factory.create_app_and_account(db_session_with_containers)
existing_user = factory.create_end_user(
db_session_with_containers,
@ -408,7 +433,9 @@ class TestEndUserServiceGetEndUserById:
assert result is not None
assert result.id == existing_user.id
def test_get_end_user_by_id_returns_none(self, db_session_with_containers: Session, factory):
def test_get_end_user_by_id_returns_none(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
app = factory.create_app_and_account(db_session_with_containers)
result = EndUserService.get_end_user_by_id(
@ -427,7 +454,9 @@ class TestEndUserServiceCreateBatch:
def factory(self):
return TestEndUserServiceFactory()
def _create_multiple_apps(self, db_session_with_containers: Session, factory, count: int = 3):
def _create_multiple_apps(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory, count: int = 3
):
"""Create multiple apps under the same tenant."""
first_app = factory.create_app_and_account(db_session_with_containers)
tenant_id = first_app.tenant_id
@ -462,7 +491,9 @@ class TestEndUserServiceCreateBatch:
)
assert result == {}
def test_create_batch_creates_users_for_all_apps(self, db_session_with_containers: Session, factory):
def test_create_batch_creates_users_for_all_apps(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=3)
app_ids = [a.id for a in apps]
user_id = f"user-{uuid4()}"
@ -477,7 +508,9 @@ class TestEndUserServiceCreateBatch:
assert result[app_id].session_id == user_id
assert result[app_id].type == InvokeFrom.SERVICE_API
def test_create_batch_default_session_id(self, db_session_with_containers: Session, factory):
def test_create_batch_default_session_id(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=2)
app_ids = [a.id for a in apps]
@ -490,7 +523,9 @@ class TestEndUserServiceCreateBatch:
assert end_user.session_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
assert end_user._is_anonymous is True
def test_create_batch_deduplicate_app_ids(self, db_session_with_containers: Session, factory):
def test_create_batch_deduplicate_app_ids(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=2)
app_ids = [apps[0].id, apps[1].id, apps[0].id, apps[1].id]
user_id = f"user-{uuid4()}"
@ -501,7 +536,9 @@ class TestEndUserServiceCreateBatch:
assert len(result) == 2
def test_create_batch_returns_existing_users(self, db_session_with_containers: Session, factory):
def test_create_batch_returns_existing_users(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=2)
app_ids = [a.id for a in apps]
user_id = f"user-{uuid4()}"
@ -520,7 +557,9 @@ class TestEndUserServiceCreateBatch:
for app_id in app_ids:
assert first_result[app_id].id == second_result[app_id].id
def test_create_batch_partial_existing_users(self, db_session_with_containers: Session, factory):
def test_create_batch_partial_existing_users(
self, db_session_with_containers: Session, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=3)
user_id = f"user-{uuid4()}"
@ -549,7 +588,9 @@ class TestEndUserServiceCreateBatch:
"invoke_type",
[InvokeFrom.SERVICE_API, InvokeFrom.WEB_APP, InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER],
)
def test_create_batch_all_invoke_types(self, db_session_with_containers: Session, invoke_type, factory):
def test_create_batch_all_invoke_types(
self, db_session_with_containers: Session, invoke_type: InvokeFrom, factory: TestEndUserServiceFactory
):
tenant_id, apps = self._create_multiple_apps(db_session_with_containers, factory, count=1)
user_id = f"user-{uuid4()}"