busya
c33792c8ee
- profile_utils: 40 tests, 100% coverage (sanitize, normalize, mask, validate payload) - security_badge_service: 17 tests, 100% coverage (role/permission extraction, security summary) - services_mapping: 4 tests, 100% coverage (client resolution, get_suggestions) - superset_lookup_service: 10 tests, 100% coverage (resolve environment, lookup success/degraded) - notification_providers: 16 tests, 95% coverage (SMTP, Telegram, Slack providers) - llm_provider: 25 tests, 91% coverage (mask_api_key, CRUD with encrypted API keys) - rbac_permission_catalog: 12 tests, 79% coverage (route scanning, sync to DB)
241 lines
9.6 KiB
Python
241 lines
9.6 KiB
Python
# #region Test.LLMProvider [C:3] [TYPE Module] [SEMANTICS test,llm,provider,mask,encryption]
|
|
# @BRIEF Tests for services/llm_provider.py — mask_api_key, is_masked_or_placeholder, LLMProviderService.
|
|
# @RELATION BINDS_TO -> [llm_provider]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import os
|
|
os.environ.setdefault("ENCRYPTION_KEY", "MldUHg5kwSAcPnnYxmhWDS6ASb6e_bWQRV5gtwHrjQ0=")
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
class TestMaskApiKey:
|
|
"""mask_api_key — safe display of API keys."""
|
|
|
|
def test_none_returns_empty(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
assert mask_api_key(None) == ""
|
|
|
|
def test_empty_returns_empty(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
assert mask_api_key("") == ""
|
|
|
|
def test_short_key_returns_four_asterisks(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
assert mask_api_key("abc") == "****"
|
|
assert mask_api_key("abcd") == "****"
|
|
|
|
def test_medium_key_shows_two_each_side(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
result = mask_api_key("abcdefgh") # 8 chars
|
|
assert result == "ab...gh"
|
|
|
|
def test_long_key_shows_four_each_side(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
result = mask_api_key("sk-1234567890abcdef") # 18 chars
|
|
assert result == "sk-1...cdef"
|
|
|
|
def test_key_length_five_shows_two_each_side(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
result = mask_api_key("abcde") # 5 chars
|
|
assert result == "ab...de"
|
|
|
|
def test_key_length_nine_shows_four_each_side(self):
|
|
from src.services.llm_provider import mask_api_key
|
|
result = mask_api_key("abcdefghi") # 9 chars
|
|
assert result == "abcd...fghi"
|
|
|
|
|
|
class TestIsMaskedOrPlaceholder:
|
|
"""is_masked_or_placeholder — detect masked/placeholder keys."""
|
|
|
|
def test_none_returns_true(self):
|
|
from src.services.llm_provider import is_masked_or_placeholder
|
|
assert is_masked_or_placeholder(None) is True
|
|
|
|
def test_empty_returns_true(self):
|
|
from src.services.llm_provider import is_masked_or_placeholder
|
|
assert is_masked_or_placeholder("") is True
|
|
|
|
def test_placeholder_returns_true(self):
|
|
from src.services.llm_provider import is_masked_or_placeholder
|
|
assert is_masked_or_placeholder("********") is True
|
|
|
|
def test_partial_mask_returns_true(self):
|
|
from src.services.llm_provider import is_masked_or_placeholder
|
|
assert is_masked_or_placeholder("sk-...abc") is True
|
|
|
|
def test_real_key_returns_false(self):
|
|
from src.services.llm_provider import is_masked_or_placeholder
|
|
assert is_masked_or_placeholder("sk-real-key-12345") is False
|
|
|
|
|
|
class TestLLMProviderService:
|
|
"""LLMProviderService — CRUD with encrypted API keys."""
|
|
|
|
@pytest.fixture
|
|
def db(self):
|
|
return MagicMock()
|
|
|
|
@pytest.fixture
|
|
def service(self, db):
|
|
from src.services.llm_provider import LLMProviderService
|
|
return LLMProviderService(db)
|
|
|
|
def test_get_all_providers(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
expected = [MagicMock(spec=LLMProvider)]
|
|
db.query.return_value.all.return_value = expected
|
|
result = service.get_all_providers()
|
|
assert result == expected
|
|
db.query.assert_called_once_with(LLMProvider)
|
|
|
|
def test_get_provider_found(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
expected = MagicMock(spec=LLMProvider)
|
|
db.query.return_value.filter.return_value.first.return_value = expected
|
|
result = service.get_provider("prov-1")
|
|
assert result is expected
|
|
db.query.assert_called_once_with(LLMProvider)
|
|
|
|
def test_get_provider_not_found(self, service, db):
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
result = service.get_provider("nonexistent")
|
|
assert result is None
|
|
|
|
def test_create_provider(self, service, db):
|
|
from src.services.llm_provider import LLMProviderService
|
|
from src.models.llm import LLMProvider
|
|
# Build a proper config mock with enum-like provider_type
|
|
config = MagicMock()
|
|
config.name = "GPT-4"
|
|
config.provider_type = MagicMock()
|
|
config.provider_type.value = "openai"
|
|
config.api_key = "sk-real-key-12345"
|
|
config.base_url = "https://api.openai.com"
|
|
config.is_active = True
|
|
config.is_multimodal = True
|
|
config.default_model = "gpt-4"
|
|
config.max_images = None
|
|
config.context_window = 8192
|
|
config.max_output_tokens = 4096
|
|
|
|
with patch.object(service.encryption, "encrypt", return_value="encrypted-key"):
|
|
with patch("src.services.llm_provider.LLMProvider") as mock_provider_cls:
|
|
mock_provider = MagicMock(spec=LLMProvider)
|
|
mock_provider_cls.return_value = mock_provider
|
|
db.add.return_value = None
|
|
db.commit.return_value = None
|
|
db.refresh.return_value = None
|
|
|
|
result = service.create_provider(config)
|
|
assert result is mock_provider
|
|
db.add.assert_called_once()
|
|
|
|
def test_update_provider(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
existing = MagicMock(spec=LLMProvider)
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
|
|
config = MagicMock()
|
|
config.name = "GPT-4 Turbo"
|
|
config.provider_type = MagicMock()
|
|
config.provider_type.value = "openai"
|
|
config.api_key = "sk-new-key"
|
|
config.base_url = "https://api.openai.com"
|
|
config.is_active = True
|
|
config.is_multimodal = True
|
|
config.default_model = "gpt-4-turbo"
|
|
config.max_images = None
|
|
config.context_window = 128000
|
|
config.max_output_tokens = 4096
|
|
|
|
with patch.object(service.encryption, "encrypt", return_value="new-encrypted"):
|
|
result = service.update_provider("prov-1", config)
|
|
db.commit.assert_called_once()
|
|
|
|
def test_delete_provider(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
existing = MagicMock(spec=LLMProvider)
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
|
|
result = service.delete_provider("prov-1")
|
|
db.delete.assert_called_once_with(existing)
|
|
db.commit.assert_called_once()
|
|
assert result is True
|
|
|
|
def test_delete_provider_not_found(self, service, db):
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
result = service.delete_provider("nonexistent")
|
|
assert result is False
|
|
|
|
def test_set_max_images(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
existing = MagicMock(spec=LLMProvider)
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
result = service.set_max_images("prov-1", 5)
|
|
assert existing.max_images == 5
|
|
db.commit.assert_called_once()
|
|
assert result is existing
|
|
|
|
def test_set_max_images_not_found(self, service, db):
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
result = service.set_max_images("nonexistent", 5)
|
|
assert result is None
|
|
|
|
def test_get_decrypted_api_key(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
existing = MagicMock(spec=LLMProvider)
|
|
existing.api_key_encrypted = "encrypted-value"
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
with patch.object(service.encryption, "decrypt", return_value="decrypted-key"):
|
|
result = service.get_decrypted_api_key("prov-1")
|
|
assert result == "decrypted-key"
|
|
|
|
def test_get_decrypted_api_key_decryption_failure(self, service, db):
|
|
from src.models.llm import LLMProvider
|
|
existing = MagicMock(spec=LLMProvider)
|
|
existing.api_key_encrypted = "corrupted"
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
with patch.object(service.encryption, "decrypt", side_effect=Exception("decrypt failed")):
|
|
result = service.get_decrypted_api_key("prov-1")
|
|
assert result is None
|
|
|
|
def test_get_decrypted_api_key_not_found(self, service, db):
|
|
db.query.return_value.filter.return_value.first.return_value = None
|
|
result = service.get_decrypted_api_key("nonexistent")
|
|
assert result is None
|
|
|
|
def test_update_provider_skips_encryption_for_masked_key(self):
|
|
"""Cover line 151: masked keys skip encryption."""
|
|
from src.services.llm_provider import LLMProviderService
|
|
from src.models.llm import LLMProvider
|
|
db = MagicMock()
|
|
existing = MagicMock(spec=LLMProvider)
|
|
db.query.return_value.filter.return_value.first.return_value = existing
|
|
service = LLMProviderService(db)
|
|
|
|
config = MagicMock()
|
|
config.api_key = "********" # masked
|
|
config.name = "Test"
|
|
config.provider_type = MagicMock()
|
|
config.provider_type.value = "test"
|
|
config.base_url = "http://test"
|
|
config.is_active = True
|
|
config.is_multimodal = False
|
|
config.default_model = None
|
|
config.max_images = None
|
|
config.context_window = None
|
|
config.max_output_tokens = None
|
|
|
|
with patch.object(service.encryption, "encrypt") as mock_encrypt:
|
|
result = service.update_provider("prov-1", config)
|
|
mock_encrypt.assert_not_called()
|
|
db.commit.assert_called_once()
|
|
# #endregion Test.LLMProvider
|