""" Tests for resolve_service function """ import os import sys import pytest # Add backend to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from src.api.generations.helpers import resolve_service from src.services.provider.registry import ModelType from src.utils.errors import AppException def test_resolve_service_with_valid_composite_id(): """测试使用有效的复合 ID 查找服务""" # This test assumes 'dashscope/qwen-image' is registered # If the model is not registered, this test will fail try: service = resolve_service("dashscope/qwen-image", ModelType.IMAGE) assert service is not None except AppException as e: # If model not found, that's expected in test environment assert e.status_code == 404 def test_resolve_service_invalid_format_no_separator(): """测试无效格式 - 缺少分隔符""" with pytest.raises(ValueError, match="must be in format 'provider/model_key'"): resolve_service("qwen-image", ModelType.IMAGE) def test_resolve_service_invalid_format_empty_string(): """测试无效格式 - 空字符串""" with pytest.raises(ValueError, match="must be in format 'provider/model_key'"): resolve_service("", ModelType.IMAGE) def test_resolve_service_invalid_format_empty_provider(): """测试无效格式 - 空的 provider""" with pytest.raises(ValueError, match="Both provider and model_key must be non-empty"): resolve_service("/qwen-image", ModelType.IMAGE) def test_resolve_service_invalid_format_empty_model_key(): """测试无效格式 - 空的 model_key""" with pytest.raises(ValueError, match="Both provider and model_key must be non-empty"): resolve_service("dashscope/", ModelType.IMAGE) def test_resolve_service_not_found(): """测试模型不存在""" with pytest.raises(AppException) as exc_info: resolve_service("invalid/model", ModelType.IMAGE) assert exc_info.value.status_code == 404 assert "not found" in exc_info.value.message.lower() def test_resolve_service_multiple_separators(): """测试多个分隔符 - 应该只按第一个分隔符拆分""" # This should parse as provider="dashscope", model_key="qwen/image" # It will likely not find the service, but format validation should pass with pytest.raises(AppException) as exc_info: resolve_service("dashscope/qwen/image", ModelType.IMAGE) # Should fail with 404, not ValueError assert exc_info.value.status_code == 404 def test_resolve_service_cache_performance(): """测试 LRU 缓存性能提升""" import time # Clear the cache first resolve_service.cache_clear() model_id = "dashscope/qwen-image" model_type = ModelType.IMAGE # First call - should be slower (cache miss) start_time = time.perf_counter() try: result1 = resolve_service(model_id, model_type) first_call_time = time.perf_counter() - start_time # Second call - should be faster (cache hit) start_time = time.perf_counter() result2 = resolve_service(model_id, model_type) second_call_time = time.perf_counter() - start_time # Verify same result assert result1 is result2, "Cached result should be the same object" # Second call should be significantly faster (at least 2x faster) # Cache hit should be nearly instant (< 0.0001s typically) assert second_call_time < first_call_time, \ f"Cached call ({second_call_time:.6f}s) should be faster than first call ({first_call_time:.6f}s)" print(f"\nCache performance test:") print(f" First call (cache miss): {first_call_time:.6f}s") print(f" Second call (cache hit): {second_call_time:.6f}s") print(f" Speedup: {first_call_time / second_call_time:.2f}x") except AppException as e: # If model not found in test environment, skip performance test if e.status_code == 404: pytest.skip("Model not registered in test environment") raise def test_resolve_service_cache_info(): """测试缓存统计信息""" # Clear the cache first resolve_service.cache_clear() # Check initial cache state cache_info = resolve_service.cache_info() assert cache_info.hits == 0 assert cache_info.misses == 0 assert cache_info.currsize == 0 model_id = "dashscope/qwen-image" model_type = ModelType.IMAGE try: # First call - cache miss resolve_service(model_id, model_type) cache_info = resolve_service.cache_info() assert cache_info.misses == 1 assert cache_info.hits == 0 assert cache_info.currsize == 1 # Second call - cache hit resolve_service(model_id, model_type) cache_info = resolve_service.cache_info() assert cache_info.misses == 1 assert cache_info.hits == 1 assert cache_info.currsize == 1 # Third call - another cache hit resolve_service(model_id, model_type) cache_info = resolve_service.cache_info() assert cache_info.misses == 1 assert cache_info.hits == 2 assert cache_info.currsize == 1 print(f"\nCache statistics: {cache_info}") except AppException as e: # If model not found in test environment, skip cache info test if e.status_code == 404: pytest.skip("Model not registered in test environment") raise