40eb4e43cd
- Auto-formatted code with ruff format - Applied ruff linting fixes with --fix Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
347 lines
15 KiB
Python
347 lines
15 KiB
Python
"""
|
|
Additional tests for app/utils/settings_service.py to increase coverage from 76.8% to 90%+.
|
|
|
|
Focuses on:
|
|
- Encryption/decryption error paths
|
|
- Database error handling (SQLAlchemyError scenarios)
|
|
- Validation edge cases
|
|
- Mixed encryption states
|
|
"""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import ApplicationSettings
|
|
from app.utils.settings_service import (
|
|
delete_setting_from_db,
|
|
get_all_settings_from_db,
|
|
get_setting_from_db,
|
|
save_setting_to_db,
|
|
validate_setting_value,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSettingsServiceEncryption:
|
|
"""Test encryption/decryption error paths in settings service."""
|
|
|
|
@patch("app.utils.encryption.decrypt_value")
|
|
@patch("app.utils.settings_service.get_setting_metadata")
|
|
def test_get_setting_decryption_fails(self, mock_metadata, mock_decrypt, db_session: Session):
|
|
"""Test get_setting_from_db when decryption fails."""
|
|
# Setup: Create a setting that appears to be encrypted
|
|
setting = ApplicationSettings(key="openai_api_key", value="enc:corrupted_data")
|
|
db_session.add(setting)
|
|
db_session.commit()
|
|
|
|
# Mock metadata to indicate it's sensitive
|
|
mock_metadata.return_value = {"sensitive": True}
|
|
|
|
# Mock decrypt_value to return failure marker
|
|
mock_decrypt.return_value = "[DECRYPTION FAILED]"
|
|
|
|
# Test
|
|
result = get_setting_from_db(db_session, "openai_api_key")
|
|
|
|
# Should return the decryption failed marker
|
|
assert result == "[DECRYPTION FAILED]"
|
|
mock_decrypt.assert_called_once()
|
|
|
|
@patch("app.utils.encryption.decrypt_value")
|
|
@patch("app.utils.settings_service.get_setting_metadata")
|
|
def test_get_setting_decryption_raises_exception(self, mock_metadata, mock_decrypt, db_session: Session):
|
|
"""Test get_setting_from_db when decryption raises an exception."""
|
|
# Setup
|
|
setting = ApplicationSettings(key="session_secret", value="enc:bad_data")
|
|
db_session.add(setting)
|
|
db_session.commit()
|
|
|
|
# Mock metadata
|
|
mock_metadata.return_value = {"sensitive": True}
|
|
|
|
# Mock decrypt to raise exception
|
|
mock_decrypt.side_effect = Exception("Invalid encryption format")
|
|
|
|
# Test - exception should propagate since it's not caught by SQLAlchemyError
|
|
with pytest.raises(Exception, match="Invalid encryption format"):
|
|
get_setting_from_db(db_session, "session_secret")
|
|
|
|
@patch("app.utils.encryption.encrypt_value")
|
|
@patch("app.utils.encryption.is_encryption_available")
|
|
@patch("app.utils.settings_service.get_setting_metadata")
|
|
def test_save_setting_encryption_unavailable(
|
|
self, mock_metadata, mock_encrypt_available, mock_encrypt, db_session: Session
|
|
):
|
|
"""Test save_setting_to_db when encryption is unavailable for sensitive data."""
|
|
# Mock: Setting is sensitive but encryption is unavailable
|
|
mock_metadata.return_value = {"sensitive": True}
|
|
mock_encrypt_available.return_value = False
|
|
|
|
# Test: Save a sensitive setting
|
|
result = save_setting_to_db(db_session, "admin_password", "sensitive_password")
|
|
|
|
# Should succeed and store in plaintext with warning logged
|
|
assert result is True
|
|
mock_encrypt.assert_not_called()
|
|
|
|
# Verify it was stored in plaintext
|
|
stored = db_session.query(ApplicationSettings).filter_by(key="admin_password").first()
|
|
assert stored.value == "sensitive_password"
|
|
|
|
@patch("app.utils.encryption.encrypt_value")
|
|
@patch("app.utils.encryption.is_encryption_available")
|
|
@patch("app.utils.settings_service.get_setting_metadata")
|
|
def test_save_setting_encryption_available(
|
|
self, mock_metadata, mock_encrypt_available, mock_encrypt, db_session: Session
|
|
):
|
|
"""Test save_setting_to_db successfully encrypts sensitive data."""
|
|
# Mock: Setting is sensitive and encryption is available
|
|
mock_metadata.return_value = {"sensitive": True}
|
|
mock_encrypt_available.return_value = True
|
|
mock_encrypt.return_value = "enc:encrypted_value"
|
|
|
|
# Test
|
|
result = save_setting_to_db(db_session, "openai_api_key", "sk-test123")
|
|
|
|
# Should succeed and store encrypted
|
|
assert result is True
|
|
mock_encrypt.assert_called_once_with("sk-test123")
|
|
|
|
# Verify stored value is encrypted
|
|
stored = db_session.query(ApplicationSettings).filter_by(key="openai_api_key").first()
|
|
assert stored.value == "enc:encrypted_value"
|
|
|
|
@patch("app.utils.encryption.decrypt_value")
|
|
@patch("app.utils.settings_service.get_setting_metadata")
|
|
def test_get_all_settings_partial_decryption_failure(self, mock_metadata, mock_decrypt, db_session: Session):
|
|
"""Test get_all_settings_from_db when some decryptions fail."""
|
|
# Setup: Create mixed sensitive and non-sensitive settings
|
|
db_session.add(ApplicationSettings(key="debug", value="true"))
|
|
db_session.add(ApplicationSettings(key="openai_api_key", value="enc:key1"))
|
|
db_session.add(ApplicationSettings(key="workdir", value="/tmp/work"))
|
|
db_session.add(ApplicationSettings(key="azure_ai_key", value="enc:key2"))
|
|
db_session.commit()
|
|
|
|
# Mock metadata to identify sensitive settings
|
|
def metadata_side_effect(key):
|
|
if key in ["openai_api_key", "azure_ai_key"]:
|
|
return {"sensitive": True}
|
|
return {"sensitive": False}
|
|
|
|
mock_metadata.side_effect = metadata_side_effect
|
|
|
|
# Mock decrypt to fail for one sensitive setting
|
|
def decrypt_side_effect(value):
|
|
if value == "enc:key1":
|
|
return "decrypted_key1"
|
|
elif value == "enc:key2":
|
|
return "[DECRYPTION FAILED]"
|
|
return value
|
|
|
|
mock_decrypt.side_effect = decrypt_side_effect
|
|
|
|
# Test
|
|
result = get_all_settings_from_db(db_session)
|
|
|
|
# Should return all settings with partial decryption failure
|
|
assert len(result) == 4
|
|
assert result["debug"] == "true"
|
|
assert result["workdir"] == "/tmp/work"
|
|
assert result["openai_api_key"] == "decrypted_key1"
|
|
assert result["azure_ai_key"] == "[DECRYPTION FAILED]"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSettingsServiceDatabaseErrors:
|
|
"""Test database error handling in settings service."""
|
|
|
|
def test_get_setting_database_error(self, db_session: Session):
|
|
"""Test get_setting_from_db handles database errors gracefully."""
|
|
# Mock the query to raise SQLAlchemyError
|
|
with patch.object(db_session, "query") as mock_query:
|
|
mock_query.side_effect = SQLAlchemyError("Database connection lost")
|
|
|
|
# Should return None on error
|
|
result = get_setting_from_db(db_session, "test_key")
|
|
assert result is None
|
|
|
|
def test_save_setting_database_error_with_rollback(self, db_session: Session):
|
|
"""Test save_setting_to_db handles commit errors and rolls back."""
|
|
# Create a spy for rollback
|
|
original_rollback = db_session.rollback
|
|
rollback_called = []
|
|
|
|
def spy_rollback():
|
|
rollback_called.append(True)
|
|
original_rollback()
|
|
|
|
db_session.rollback = spy_rollback
|
|
|
|
# Mock commit to raise error
|
|
with patch.object(db_session, "commit") as mock_commit:
|
|
mock_commit.side_effect = SQLAlchemyError("Database locked")
|
|
|
|
# Should return False on error
|
|
result = save_setting_to_db(db_session, "test_key", "test_value")
|
|
assert result is False
|
|
|
|
# Verify rollback was called
|
|
assert len(rollback_called) > 0
|
|
|
|
def test_delete_setting_database_error_with_rollback(self, db_session: Session):
|
|
"""Test delete_setting_from_db handles errors and rolls back."""
|
|
# Create a setting first
|
|
save_setting_to_db(db_session, "test_key", "test_value")
|
|
|
|
# Mock delete to raise error
|
|
with patch.object(db_session, "commit") as mock_commit:
|
|
mock_commit.side_effect = SQLAlchemyError("Foreign key constraint violation")
|
|
|
|
# Should return False on error
|
|
result = delete_setting_from_db(db_session, "test_key")
|
|
assert result is False
|
|
|
|
def test_get_all_settings_database_error_returns_empty_dict(self, db_session: Session):
|
|
"""Test get_all_settings_from_db returns empty dict on error."""
|
|
# Mock query to raise error
|
|
with patch.object(db_session, "query") as mock_query:
|
|
mock_query.side_effect = SQLAlchemyError("Connection timeout")
|
|
|
|
# Should return empty dict on error
|
|
result = get_all_settings_from_db(db_session)
|
|
assert result == {}
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSettingsValidationEdgeCases:
|
|
"""Test validation edge cases in validate_setting_value."""
|
|
|
|
def test_validate_with_none_value_for_required(self):
|
|
"""Test validation when None is passed for required field."""
|
|
# None value should fail for required field
|
|
is_valid, error = validate_setting_value("database_url", None)
|
|
assert is_valid is False
|
|
# The function converts None to empty string check, which should fail for required
|
|
|
|
def test_validate_with_empty_string_for_required(self):
|
|
"""Test validation when empty string is passed for required field."""
|
|
# Empty string should fail for required field
|
|
is_valid, error = validate_setting_value("database_url", "")
|
|
assert is_valid is False
|
|
assert "required" in error.lower()
|
|
|
|
def test_validate_boolean_case_insensitive(self):
|
|
"""Test boolean validation is case insensitive."""
|
|
# Test various case combinations
|
|
test_cases = ["TRUE", "False", "YES", "no", "1", "0"]
|
|
for value in test_cases:
|
|
is_valid, error = validate_setting_value("debug", value)
|
|
assert is_valid is True, f"Failed for value: {value}"
|
|
|
|
def test_validate_boolean_with_whitespace(self):
|
|
"""Test boolean validation fails with whitespace."""
|
|
# Whitespace should cause validation to fail
|
|
is_valid, error = validate_setting_value("debug", " true ")
|
|
assert is_valid is False
|
|
assert "boolean" in error.lower()
|
|
|
|
def test_validate_boolean_with_invalid_value(self):
|
|
"""Test boolean validation with completely invalid values."""
|
|
invalid_values = ["maybe", "yep", "nope", "2", "on", "off"]
|
|
for value in invalid_values:
|
|
is_valid, error = validate_setting_value("debug", value)
|
|
assert is_valid is False, f"Should fail for value: {value}"
|
|
assert "boolean" in error.lower()
|
|
|
|
def test_validate_integer_with_float_string(self):
|
|
"""Test integer validation rejects float strings."""
|
|
# Float string should fail integer validation
|
|
is_valid, error = validate_setting_value("email_port", "1.5")
|
|
assert is_valid is False
|
|
assert "integer" in error.lower()
|
|
|
|
def test_validate_integer_with_scientific_notation(self):
|
|
"""Test integer validation with scientific notation."""
|
|
# Scientific notation should technically pass int() but semantically odd
|
|
is_valid, error = validate_setting_value("email_port", "1e3")
|
|
# This will actually pass int("1e3") since Python 3 handles it
|
|
# But let's test the actual behavior
|
|
try:
|
|
int("1e3")
|
|
# If int() accepts it, validation should pass
|
|
assert is_valid is True
|
|
except ValueError:
|
|
# If int() rejects it, validation should fail
|
|
assert is_valid is False
|
|
|
|
def test_validate_integer_negative(self):
|
|
"""Test integer validation accepts negative integers."""
|
|
is_valid, error = validate_setting_value("ftp_port", "-1")
|
|
assert is_valid is True # Validation only checks it's an integer
|
|
|
|
def test_validate_integer_very_large(self):
|
|
"""Test integer validation with very large numbers."""
|
|
is_valid, error = validate_setting_value("email_port", "999999999999")
|
|
assert is_valid is True # Python handles arbitrary precision integers
|
|
|
|
def test_validate_integer_with_letters(self):
|
|
"""Test integer validation fails with letters."""
|
|
is_valid, error = validate_setting_value("email_port", "abc")
|
|
assert is_valid is False
|
|
assert "integer" in error.lower()
|
|
|
|
def test_validate_session_secret_boundary_31_chars(self):
|
|
"""Test session_secret validation at 31 characters (should fail)."""
|
|
secret_31 = "a" * 31
|
|
is_valid, error = validate_setting_value("session_secret", secret_31)
|
|
assert is_valid is False
|
|
assert "32 characters" in error
|
|
|
|
def test_validate_session_secret_boundary_32_chars(self):
|
|
"""Test session_secret validation at 32 characters (should pass)."""
|
|
secret_32 = "a" * 32
|
|
is_valid, error = validate_setting_value("session_secret", secret_32)
|
|
assert is_valid is True
|
|
assert error is None
|
|
|
|
def test_validate_session_secret_boundary_33_chars(self):
|
|
"""Test session_secret validation at 33 characters (should pass)."""
|
|
secret_33 = "a" * 33
|
|
is_valid, error = validate_setting_value("session_secret", secret_33)
|
|
assert is_valid is True
|
|
assert error is None
|
|
|
|
def test_validate_session_secret_empty_string(self):
|
|
"""Test session_secret validation with empty string."""
|
|
is_valid, error = validate_setting_value("session_secret", "")
|
|
# Empty string check: `if key == "session_secret" and value and len(value) < 32`
|
|
# The `and value` clause means empty string passes through (not checked for length)
|
|
# However, session_secret is marked required in metadata, so empty string should fail
|
|
# Let's check if session_secret is actually required
|
|
from app.utils.settings_service import get_setting_metadata
|
|
|
|
metadata = get_setting_metadata("session_secret")
|
|
if metadata.get("required", False):
|
|
# If required, empty string should fail
|
|
assert is_valid is False
|
|
else:
|
|
# If not required, empty string passes
|
|
assert is_valid is True
|
|
|
|
def test_validate_non_required_empty_string(self):
|
|
"""Test validation of non-required field with empty string."""
|
|
# Non-required field with empty string should pass
|
|
is_valid, error = validate_setting_value("dropbox_folder", "")
|
|
assert is_valid is True
|
|
assert error is None
|
|
|
|
def test_validate_unknown_setting(self):
|
|
"""Test validation of unknown setting key."""
|
|
# Unknown settings should use default metadata and pass basic validation
|
|
is_valid, error = validate_setting_value("unknown_setting_xyz", "some_value")
|
|
assert is_valid is True
|
|
assert error is None
|