ss-tools/backend/src/services/__tests__/test_llm_provider.py

# region test_llm_provider [TYPE Module]
# @RELATION: VERIFIES -> [src.services.llm_provider:Module]
# @SEMANTICS: tests, llm-provider, encryption, contract
# @PURPOSE: Contract testing for LLMProviderService and EncryptionManager
# endregion test_llm_provider

import os
from unittest.mock import MagicMock

import pytest
from cryptography.fernet import Fernet
from sqlalchemy.orm import Session

from src.models.llm import LLMProvider
from src.plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType
from src.services.llm_provider import EncryptionManager, LLMProviderService

# region _test_encryption_key_fixture [TYPE Global]
# @PURPOSE: Ensure encryption-dependent provider tests run with a valid Fernet key.
# @RELATION: DEPENDS_ON -> [pytest:Module]
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
# endregion _test_encryption_key_fixture


# region test_provider_type_enum_values [TYPE Function]
# @RELATION: VERIFIES -> [LLMProviderType]
# @PURPOSE: Regression guard — prevent accidental value drift when enum variants are added or renamed.
# @INVARIANT: Every LLMProviderType member must have a value matching its lowercase name.
def test_provider_type_enum_values():
    """Verify all LLMProviderType enum values match their expected strings."""
    assert LLMProviderType.OPENAI.value == "openai"
    assert LLMProviderType.OPENROUTER.value == "openrouter"
    assert LLMProviderType.KILO.value == "kilo"
    assert LLMProviderType.LITELLM.value == "litellm"
    # If new providers are added, extend this list — enum value drift breaks string comparisons
    # across executor.py, preview.py, ProviderConfig.svelte, and the DB layer.
    expected_count = 4
    assert len(LLMProviderType) == expected_count, (
        f"Expected {expected_count} provider types, got {len(LLMProviderType)}. "
        "If you added a new one, add its value assertion above and update expected_count."
    )
# endregion test_provider_type_enum_values


# @TEST_CONTRACT: EncryptionManagerModel -> Invariants
# @TEST_INVARIANT: symmetric_encryption
# region test_encryption_cycle [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Verify EncryptionManager round-trip encryption/decryption invariant for non-empty secrets.
def test_encryption_cycle():
    """Verify encrypted data can be decrypted back to original string."""
    manager = EncryptionManager()
    original = "secret_api_key_123"
    encrypted = manager.encrypt(original)
    assert encrypted != original
    assert manager.decrypt(encrypted) == original


# @TEST_EDGE: empty_string_encryption
# endregion test_encryption_cycle


# region test_empty_string_encryption [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Verify EncryptionManager preserves empty-string payloads through encrypt/decrypt cycle.
def test_empty_string_encryption():
    manager = EncryptionManager()
    original = ""
    encrypted = manager.encrypt(original)
    assert manager.decrypt(encrypted) == ""


# @TEST_EDGE: decrypt_invalid_data
# endregion test_empty_string_encryption


# region test_decrypt_invalid_data [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Ensure decrypt rejects invalid ciphertext input by raising an exception.
def test_decrypt_invalid_data():
    manager = EncryptionManager()
    with pytest.raises(Exception):
        manager.decrypt("not-encrypted-string")


# @TEST_FIXTURE: mock_db_session
# endregion test_decrypt_invalid_data


# region mock_db [TYPE Fixture]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: MagicMock(spec=Session) fixture providing a constrained DB session double for LLMProviderService tests.
# @INVARIANT: Chained calls beyond Session spec create unconstrained intermediate mocks; only top-level query/add/commit are spec-enforced.
@pytest.fixture
def mock_db():
    # @RISK: query() returns unconstrained MagicMock — chain beyond query() has no spec protection. Consider create_autospec(Session) for full chain safety.
    return MagicMock(spec=Session)


# endregion mock_db


# region service [TYPE Fixture]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: LLMProviderService fixture wired to mock_db for provider CRUD tests.
@pytest.fixture
def service(mock_db):
    return LLMProviderService(db=mock_db)


# endregion service


# region test_get_all_providers [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Verify provider list retrieval issues query/all calls on the backing DB session.
def test_get_all_providers(service, mock_db):
    service.get_all_providers()
    mock_db.query.assert_called()
    mock_db.query().all.assert_called()


# endregion test_get_all_providers


# region test_create_provider [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Ensure provider creation persists entity and stores API key in encrypted form.
def test_create_provider(service, mock_db):
    config = LLMProviderConfig(
        provider_type=LLMProviderType.OPENAI,
        name="Test OpenAI",
        base_url="https://api.openai.com",
        api_key="sk-test",
        default_model="gpt-4",
        is_active=True,
    )

    provider = service.create_provider(config)

    mock_db.add.assert_called()
    mock_db.commit.assert_called()
    # Verify API key was encrypted
    assert provider.api_key != "sk-test"
    # Decrypt to verify it matches
    assert EncryptionManager().decrypt(provider.api_key) == "sk-test"


# endregion test_create_provider


# region test_get_decrypted_api_key [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Verify service decrypts stored provider API key for an existing provider record.
def test_get_decrypted_api_key(service, mock_db):
    # Setup mock provider
    encrypted_key = EncryptionManager().encrypt("secret-value")
    mock_provider = LLMProvider(id="p1", api_key=encrypted_key)
    mock_db.query().filter().first.return_value = mock_provider

    key = service.get_decrypted_api_key("p1")
    assert key == "secret-value"


# endregion test_get_decrypted_api_key


# region test_get_decrypted_api_key_not_found [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Verify missing provider lookup returns None instead of attempting decryption.
def test_get_decrypted_api_key_not_found(service, mock_db):
    mock_db.query().filter().first.return_value = None
    assert service.get_decrypted_api_key("missing") is None


# endregion test_get_decrypted_api_key_not_found


# region test_update_provider_ignores_masked_placeholder_api_key [TYPE Function]
# @RELATION: BINDS_TO -> [test_llm_provider:Module]
# @PURPOSE: Ensure masked placeholder API keys do not overwrite previously encrypted provider secrets.
def test_update_provider_ignores_masked_placeholder_api_key(service, mock_db):
    existing_encrypted = EncryptionManager().encrypt("secret-value")
    mock_provider = LLMProvider(
        id="p1",
        provider_type="openai",
        name="Existing",
        base_url="https://api.openai.com/v1",
        api_key=existing_encrypted,
        default_model="gpt-4o",
        is_active=True,
    )
    mock_db.query().filter().first.return_value = mock_provider
    config = LLMProviderConfig(
        id="p1",
        provider_type=LLMProviderType.OPENAI,
        name="Existing",
        base_url="https://api.openai.com/v1",
        api_key="********",
        default_model="gpt-4o",
        is_active=False,
    )

    updated = service.update_provider("p1", config)

    assert updated is mock_provider
    assert updated.api_key == existing_encrypted
    assert EncryptionManager().decrypt(updated.api_key) == "secret-value"
    assert updated.is_active is False


# endregion test_update_provider_ignores_masked_placeholder_api_key