test(services): add unit tests for 7 service-layer modules
- 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)
This commit is contained in:
240
backend/tests/services/test_llm_provider.py
Normal file
240
backend/tests/services/test_llm_provider.py
Normal file
@@ -0,0 +1,240 @@
|
||||
# #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
|
||||
205
backend/tests/services/test_notification_providers.py
Normal file
205
backend/tests/services/test_notification_providers.py
Normal file
@@ -0,0 +1,205 @@
|
||||
# #region Test.NotificationProviders [C:3] [TYPE Module] [SEMANTICS test,notification,provider,smtp,telegram,slack]
|
||||
# @BRIEF Tests for notifications/providers.py — SMTP, Telegram, Slack providers with mocked network.
|
||||
# @RELATION BINDS_TO -> [providers]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
import pytest
|
||||
|
||||
|
||||
class TestSMTPProvider:
|
||||
"""SMTPProvider — delivers notifications via SMTP."""
|
||||
|
||||
@pytest.fixture
|
||||
def config(self):
|
||||
return {
|
||||
"host": "smtp.example.com",
|
||||
"port": 587,
|
||||
"username": "user",
|
||||
"password": "pass",
|
||||
"from_email": "noreply@example.com",
|
||||
"use_tls": True,
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def provider(self, config):
|
||||
from src.services.notifications.providers import SMTPProvider
|
||||
return SMTPProvider(config)
|
||||
|
||||
def test_init(self, provider):
|
||||
assert provider.host == "smtp.example.com"
|
||||
assert provider.port == 587
|
||||
assert provider.from_email == "noreply@example.com"
|
||||
assert provider.use_tls is True
|
||||
|
||||
def test_init_defaults(self):
|
||||
from src.services.notifications.providers import SMTPProvider
|
||||
p = SMTPProvider({"host": "smtp.x.com"})
|
||||
assert p.port == 587
|
||||
assert p.use_tls is True
|
||||
assert p.username is None
|
||||
assert p.password is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_success(self, provider):
|
||||
mock_smtp = AsyncMock()
|
||||
mock_smtp.__aenter__.return_value = mock_smtp
|
||||
|
||||
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
||||
result = await provider.send("admin@x.com", "Test", "Body")
|
||||
assert result is True
|
||||
mock_smtp.sendmail.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failure_returns_false(self, provider):
|
||||
mock_smtp = AsyncMock()
|
||||
mock_smtp.__aenter__.side_effect = ConnectionError("SMTP timeout")
|
||||
|
||||
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
||||
result = await provider.send("admin@x.com", "Test", "Body")
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_no_auth(self, config):
|
||||
from src.services.notifications.providers import SMTPProvider
|
||||
p = SMTPProvider({"host": "smtp.x.com", "username": None, "password": None})
|
||||
mock_smtp = AsyncMock()
|
||||
mock_smtp.__aenter__.return_value = mock_smtp
|
||||
|
||||
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
||||
result = await p.send("admin@x.com", "Test", "Body")
|
||||
assert result is True
|
||||
mock_smtp.login.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_no_tls(self, config):
|
||||
from src.services.notifications.providers import SMTPProvider
|
||||
p = SMTPProvider({"host": "smtp.x.com", "use_tls": False})
|
||||
mock_smtp = AsyncMock()
|
||||
mock_smtp.__aenter__.return_value = mock_smtp
|
||||
|
||||
with patch("src.services.notifications.providers.aiosmtplib.SMTP", return_value=mock_smtp):
|
||||
result = await p.send("admin@x.com", "Test", "Body")
|
||||
assert result is True
|
||||
mock_smtp.starttls.assert_not_called()
|
||||
|
||||
|
||||
class TestTelegramProvider:
|
||||
"""TelegramProvider — delivers via Telegram Bot API."""
|
||||
|
||||
@pytest.fixture
|
||||
def config(self):
|
||||
return {"bot_token": "123:ABC"}
|
||||
|
||||
@pytest.fixture
|
||||
def provider(self, config):
|
||||
from src.services.notifications.providers import TelegramProvider
|
||||
return TelegramProvider(config)
|
||||
|
||||
def test_init(self, provider):
|
||||
assert provider.bot_token == "123:ABC"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_success(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.return_value = None
|
||||
mock_client.post.return_value = mock_response
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("@admin", "Alert", "DB issue")
|
||||
assert result is True
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_no_token_returns_false(self):
|
||||
from src.services.notifications.providers import TelegramProvider
|
||||
p = TelegramProvider({"bot_token": None})
|
||||
result = await p.send("@admin", "Alert", "Body")
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failure_returns_false(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = ConnectionError("API timeout")
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("@admin", "Alert", "Body")
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_http_error_returns_false(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = ConnectionError("HTTP 500")
|
||||
mock_client.post.return_value = mock_response
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("@admin", "Alert", "Body")
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestSlackProvider:
|
||||
"""SlackProvider — delivers via Slack Webhooks."""
|
||||
|
||||
@pytest.fixture
|
||||
def config(self):
|
||||
return {"webhook_url": "https://hooks.slack.com/services/xxx"}
|
||||
|
||||
@pytest.fixture
|
||||
def provider(self, config):
|
||||
from src.services.notifications.providers import SlackProvider
|
||||
return SlackProvider(config)
|
||||
|
||||
def test_init(self, provider):
|
||||
assert provider.webhook_url == "https://hooks.slack.com/services/xxx"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_success(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.return_value = None
|
||||
mock_client.post.return_value = mock_response
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("#alerts", "Alert", "DB issue")
|
||||
assert result is True
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_no_webhook_returns_false(self):
|
||||
from src.services.notifications.providers import SlackProvider
|
||||
p = SlackProvider({"webhook_url": None})
|
||||
result = await p.send("#alerts", "Alert", "Body")
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_failure_returns_false(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post.side_effect = ConnectionError("Webhook timeout")
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("#alerts", "Alert", "Body")
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_http_error_returns_false(self, provider):
|
||||
mock_client = AsyncMock()
|
||||
mock_response = MagicMock()
|
||||
mock_response.raise_for_status.side_effect = ConnectionError("HTTP 500")
|
||||
mock_client.post.return_value = mock_response
|
||||
|
||||
with patch("src.services.notifications.providers._get_http_client",
|
||||
return_value=mock_client):
|
||||
result = await provider.send("#alerts", "Alert", "Body")
|
||||
assert result is False
|
||||
# #endregion Test.NotificationProviders
|
||||
373
backend/tests/services/test_profile_utils.py
Normal file
373
backend/tests/services/test_profile_utils.py
Normal file
@@ -0,0 +1,373 @@
|
||||
# #region Test.ProfileUtils [C:3] [TYPE Module] [SEMANTICS test,profile,utils,pure-functions]
|
||||
# @BRIEF Tests for profile_utils.py — sanitize, normalize, mask, validate payload, build default preference, owner tokens.
|
||||
# @RELATION BINDS_TO -> [ProfileUtils]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class TestSanitizeText:
|
||||
"""sanitize_text — normalize optional text into trimmed form or None."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_text
|
||||
assert sanitize_text(None) is None
|
||||
|
||||
def test_empty_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_text
|
||||
assert sanitize_text("") is None
|
||||
|
||||
def test_whitespace_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_text
|
||||
assert sanitize_text(" ") is None
|
||||
|
||||
def test_trimmed_text(self):
|
||||
from src.services.profile_utils import sanitize_text
|
||||
assert sanitize_text(" hello ") == "hello"
|
||||
|
||||
def test_already_trimmed(self):
|
||||
from src.services.profile_utils import sanitize_text
|
||||
assert sanitize_text("world") == "world"
|
||||
|
||||
|
||||
class TestSanitizeSecret:
|
||||
"""sanitize_secret — normalize secret value into trimmed form or None."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_secret
|
||||
assert sanitize_secret(None) is None
|
||||
|
||||
def test_empty_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_secret
|
||||
assert sanitize_secret("") is None
|
||||
|
||||
def test_whitespace_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_secret
|
||||
assert sanitize_secret(" ") is None
|
||||
|
||||
def test_trimmed_secret(self):
|
||||
from src.services.profile_utils import sanitize_secret
|
||||
assert sanitize_secret(" my-token-abc ") == "my-token-abc"
|
||||
|
||||
|
||||
class TestSanitizeUsername:
|
||||
"""sanitize_username — delegates to sanitize_text."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
from src.services.profile_utils import sanitize_username
|
||||
assert sanitize_username(None) is None
|
||||
|
||||
def test_valid_username(self):
|
||||
from src.services.profile_utils import sanitize_username
|
||||
assert sanitize_username(" admin ") == "admin"
|
||||
|
||||
|
||||
class TestNormalizeUsername:
|
||||
"""normalize_username — trim + lower."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
from src.services.profile_utils import normalize_username
|
||||
assert normalize_username(None) is None
|
||||
|
||||
def test_empty_returns_none(self):
|
||||
from src.services.profile_utils import normalize_username
|
||||
assert normalize_username("") is None
|
||||
|
||||
def test_trim_and_lower(self):
|
||||
from src.services.profile_utils import normalize_username
|
||||
assert normalize_username(" AdminUser ") == "adminuser"
|
||||
|
||||
def test_already_normalized(self):
|
||||
from src.services.profile_utils import normalize_username
|
||||
assert normalize_username("guest") == "guest"
|
||||
|
||||
|
||||
class TestNormalizeStartPage:
|
||||
"""normalize_start_page — map aliases to canonical values."""
|
||||
|
||||
def test_reports_logs_maps_to_reports(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("reports-logs") == "reports"
|
||||
|
||||
def test_dashboards_passes_through(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("dashboards") == "dashboards"
|
||||
|
||||
def test_datasets_passes_through(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("datasets") == "datasets"
|
||||
|
||||
def test_reports_passes_through(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("reports") == "reports"
|
||||
|
||||
def test_unknown_defaults_to_dashboards(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("settings") == "dashboards"
|
||||
|
||||
def test_none_defaults_to_dashboards(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page(None) == "dashboards"
|
||||
|
||||
def test_case_insensitive(self):
|
||||
from src.services.profile_utils import normalize_start_page
|
||||
assert normalize_start_page("DATASETS") == "datasets"
|
||||
|
||||
|
||||
class TestNormalizeDensity:
|
||||
"""normalize_density — map aliases to canonical values."""
|
||||
|
||||
def test_free_maps_to_comfortable(self):
|
||||
from src.services.profile_utils import normalize_density
|
||||
assert normalize_density("free") == "comfortable"
|
||||
|
||||
def test_compact_passes_through(self):
|
||||
from src.services.profile_utils import normalize_density
|
||||
assert normalize_density("compact") == "compact"
|
||||
|
||||
def test_comfortable_passes_through(self):
|
||||
from src.services.profile_utils import normalize_density
|
||||
assert normalize_density("comfortable") == "comfortable"
|
||||
|
||||
def test_unknown_defaults_to_comfortable(self):
|
||||
from src.services.profile_utils import normalize_density
|
||||
assert normalize_density("spacious") == "comfortable"
|
||||
|
||||
def test_none_defaults_to_comfortable(self):
|
||||
from src.services.profile_utils import normalize_density
|
||||
assert normalize_density(None) == "comfortable"
|
||||
|
||||
|
||||
class TestMaskSecretValue:
|
||||
"""mask_secret_value — safe display value for secrets."""
|
||||
|
||||
def test_none_returns_none(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value(None) is None
|
||||
|
||||
def test_empty_returns_none(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value("") is None
|
||||
|
||||
def test_short_secret_returns_asterisks(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value("ab") == "***"
|
||||
|
||||
def test_exactly_four_chars_returns_asterisks(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value("abcd") == "***"
|
||||
|
||||
def test_longer_secret_masked(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value("my-token-value") == "my***ue"
|
||||
|
||||
def test_whitespace_stripped_before_masking(self):
|
||||
from src.services.profile_utils import mask_secret_value
|
||||
assert mask_secret_value(" ab ") == "***"
|
||||
|
||||
|
||||
class TestValidateUpdatePayload:
|
||||
"""validate_update_payload — field-level validation for preference mutation."""
|
||||
|
||||
def test_valid_payload_returns_empty(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email="admin@example.com",
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert errors == []
|
||||
|
||||
def test_username_with_spaces(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin user",
|
||||
show_only_my_dashboards=False,
|
||||
git_email=None,
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("space" in e.lower() for e in errors)
|
||||
|
||||
def test_show_only_requires_username_when_true(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username=None,
|
||||
show_only_my_dashboards=True,
|
||||
git_email=None,
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("username" in e.lower() and "required" in e.lower() for e in errors)
|
||||
|
||||
def test_invalid_git_email(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email="not-an-email",
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("email" in e.lower() for e in errors)
|
||||
|
||||
def test_git_email_with_spaces(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email="admin @x.com",
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("email" in e.lower() for e in errors)
|
||||
|
||||
def test_git_email_starts_with_at(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email="@example.com",
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("email" in e.lower() for e in errors)
|
||||
|
||||
def test_invalid_start_page(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email=None,
|
||||
start_page="settings",
|
||||
dashboards_table_density="comfortable",
|
||||
)
|
||||
assert any("start page" in e.lower() for e in errors)
|
||||
|
||||
def test_invalid_density(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email=None,
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="spacious",
|
||||
)
|
||||
assert any("density" in e.lower() for e in errors)
|
||||
|
||||
def test_invalid_notification_email(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email=None,
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
email_address="bad-email",
|
||||
)
|
||||
assert any("notification email" in e.lower() for e in errors)
|
||||
|
||||
def test_valid_notification_email(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin",
|
||||
show_only_my_dashboards=False,
|
||||
git_email=None,
|
||||
start_page="dashboards",
|
||||
dashboards_table_density="comfortable",
|
||||
email_address="user@example.com",
|
||||
)
|
||||
assert not any("notification email" in e.lower() for e in errors)
|
||||
|
||||
def test_multiple_errors(self):
|
||||
from src.services.profile_utils import validate_update_payload
|
||||
errors = validate_update_payload(
|
||||
superset_username="admin user",
|
||||
show_only_my_dashboards=True,
|
||||
git_email="bad",
|
||||
start_page="unknown",
|
||||
dashboards_table_density="spacious",
|
||||
)
|
||||
assert len(errors) >= 4
|
||||
|
||||
|
||||
class TestBuildDefaultPreference:
|
||||
"""build_default_preference — default ProfilePreference for unconfigured users."""
|
||||
|
||||
def test_returns_profile_preference(self):
|
||||
from src.services.profile_utils import build_default_preference
|
||||
result = build_default_preference("user-1")
|
||||
assert result.user_id == "user-1"
|
||||
assert result.superset_username is None
|
||||
assert result.show_only_my_dashboards is False
|
||||
assert result.show_only_slug_dashboards is True
|
||||
assert result.start_page == "dashboards"
|
||||
assert result.dashboards_table_density == "comfortable"
|
||||
assert result.has_git_personal_access_token is False
|
||||
assert result.notify_on_fail is True
|
||||
|
||||
|
||||
class TestNormalizeOwnerTokens:
|
||||
"""normalize_owner_tokens — deduplicated lower-cased tokens from owners."""
|
||||
|
||||
def test_none_returns_empty(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
assert normalize_owner_tokens(None) == []
|
||||
|
||||
def test_empty_list(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
assert normalize_owner_tokens([]) == []
|
||||
|
||||
def test_dict_owner_with_username(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = [{"username": "AdminUser"}]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert "adminuser" in tokens
|
||||
|
||||
def test_dict_owner_with_full_name(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = [{"first_name": "John", "last_name": "Doe"}]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert "john" in tokens
|
||||
assert "doe" in tokens
|
||||
assert "john_doe" or "john doe" in tokens
|
||||
|
||||
def test_string_owner(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = [" Admin "]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert "admin" in tokens
|
||||
|
||||
def test_dedup(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = ["admin", " Admin "]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert tokens == ["admin"]
|
||||
|
||||
def test_skip_empty(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = ["", " ", None]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert tokens == []
|
||||
|
||||
def test_dict_owner_with_email(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = [{"email": "User@Example.com"}]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert "user@example.com" in tokens
|
||||
|
||||
def test_dict_owner_multiple_candidates(self):
|
||||
from src.services.profile_utils import normalize_owner_tokens
|
||||
owners = [{"username": "jdoe", "first_name": "John", "last_name": "Doe", "email": "john@example.com"}]
|
||||
tokens = normalize_owner_tokens(owners)
|
||||
assert "jdoe" in tokens
|
||||
assert "john" in tokens
|
||||
assert "doe" in tokens
|
||||
assert "john@example.com" in tokens
|
||||
# #endregion Test.ProfileUtils
|
||||
173
backend/tests/services/test_rbac_permission_catalog.py
Normal file
173
backend/tests/services/test_rbac_permission_catalog.py
Normal file
@@ -0,0 +1,173 @@
|
||||
# #region Test.RbacPermissionCatalog [C:3] [TYPE Module] [SEMANTICS test,rbac,permission,discovery,route]
|
||||
# @BRIEF Tests for services/rbac_permission_catalog.py — permission discovery, route scanning, sync.
|
||||
# @RELATION BINDS_TO -> [rbac_permission_catalog]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock, patch, PropertyMock
|
||||
import pytest
|
||||
|
||||
|
||||
class TestIterRouteFiles:
|
||||
"""_iter_route_files — discover Python files in routes directory."""
|
||||
|
||||
def test_directory_not_found(self):
|
||||
from src.services.rbac_permission_catalog import _iter_route_files
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path("/nonexistent/path")):
|
||||
result = _iter_route_files()
|
||||
assert result == []
|
||||
|
||||
def test_finds_py_files(self):
|
||||
from src.services.rbac_permission_catalog import _iter_route_files
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
routes_dir = Path(tmpdir) / "routes"
|
||||
routes_dir.mkdir()
|
||||
(routes_dir / "test_route.py").write_text("")
|
||||
(routes_dir / "__init__.py").write_text("")
|
||||
(routes_dir / "subdir").mkdir()
|
||||
(routes_dir / "subdir" / "nested.py").write_text("")
|
||||
# These should be excluded
|
||||
(routes_dir / "__tests__").mkdir()
|
||||
(routes_dir / "__tests__" / "test_excluded.py").write_text("")
|
||||
(routes_dir / "__pycache__").mkdir()
|
||||
(routes_dir / "__pycache__" / "cached.py").write_text("")
|
||||
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", routes_dir):
|
||||
files = _iter_route_files()
|
||||
filenames = [f.name for f in files]
|
||||
assert "test_route.py" in filenames
|
||||
assert "__init__.py" in filenames
|
||||
assert "nested.py" in filenames
|
||||
assert "test_excluded.py" not in filenames
|
||||
assert "cached.py" not in filenames
|
||||
|
||||
|
||||
class TestDiscoverRoutePermissions:
|
||||
"""_discover_route_permissions — extract has_permission() calls from route files."""
|
||||
|
||||
def test_extracts_permissions(self):
|
||||
from src.services.rbac_permission_catalog import _discover_route_permissions
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
route_file = Path(tmpdir) / "test_route.py"
|
||||
route_file.write_text("""
|
||||
has_permission("dashboards", "READ")
|
||||
has_permission("datasets", "WRITE")
|
||||
# Not a permission call:
|
||||
x = has_permission("users", "ADMIN")
|
||||
""")
|
||||
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
|
||||
result = _discover_route_permissions()
|
||||
assert ("dashboards", "READ") in result
|
||||
assert ("datasets", "WRITE") in result
|
||||
assert ("users", "ADMIN") in result
|
||||
|
||||
def test_empty_file(self):
|
||||
from src.services.rbac_permission_catalog import _discover_route_permissions
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
route_file = Path(tmpdir) / "empty.py"
|
||||
route_file.write_text("# No permissions here")
|
||||
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
|
||||
result = _discover_route_permissions()
|
||||
assert result == set()
|
||||
|
||||
def test_multiple_files(self):
|
||||
from src.services.rbac_permission_catalog import _discover_route_permissions
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
f1 = Path(tmpdir) / "routes_a.py"
|
||||
f1.write_text('has_permission("dashboard", "READ")')
|
||||
f2 = Path(tmpdir) / "routes_b.py"
|
||||
f2.write_text('has_permission("dataset", "WRITE")')
|
||||
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
|
||||
result = _discover_route_permissions()
|
||||
assert ("dashboard", "READ") in result
|
||||
assert ("dataset", "WRITE") in result
|
||||
|
||||
def test_single_quote_variants(self):
|
||||
from src.services.rbac_permission_catalog import _discover_route_permissions
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
f = Path(tmpdir) / "test.py"
|
||||
f.write_text("""has_permission('dashboard', 'READ')""")
|
||||
|
||||
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
|
||||
result = _discover_route_permissions()
|
||||
assert ("dashboard", "READ") in result
|
||||
|
||||
|
||||
class TestDiscoverDeclaredPermissions:
|
||||
"""discover_declared_permissions — cached public API."""
|
||||
|
||||
def test_returns_permissions(self):
|
||||
from src.services.rbac_permission_catalog import discover_declared_permissions
|
||||
with patch("src.services.rbac_permission_catalog._discover_route_permissions",
|
||||
return_value={("dashboard", "READ"), ("dataset", "WRITE")}):
|
||||
result = discover_declared_permissions()
|
||||
assert ("dashboard", "READ") in result
|
||||
assert ("dataset", "WRITE") in result
|
||||
|
||||
def test_caching(self):
|
||||
from src.services.rbac_permission_catalog import discover_declared_permissions
|
||||
mock_fn = MagicMock(return_value={("dashboard", "READ")})
|
||||
with patch("src.services.rbac_permission_catalog._discover_route_permissions", mock_fn):
|
||||
result1 = discover_declared_permissions()
|
||||
result2 = discover_declared_permissions()
|
||||
assert result1 == result2
|
||||
# With no plugin_loader, it calls _discover_route_permissions
|
||||
# The lru_cache should only call once if args match
|
||||
# discover_declared_permissions takes optional plugin_loader
|
||||
|
||||
def test_with_plugin_loader(self):
|
||||
from src.services.rbac_permission_catalog import discover_declared_permissions
|
||||
mock_loader = MagicMock()
|
||||
mock_loader.list_plugins.return_value = []
|
||||
mock_fn = MagicMock(return_value={("dashboard", "READ")})
|
||||
with patch("src.services.rbac_permission_catalog._discover_route_permissions", mock_fn):
|
||||
result = discover_declared_permissions(plugin_loader=mock_loader)
|
||||
# Should call plugin loader to discover plugin permissions
|
||||
assert ("dashboard", "READ") in result
|
||||
|
||||
|
||||
class TestSyncPermissionCatalog:
|
||||
"""sync_permission_catalog — sync discovered permissions to DB."""
|
||||
|
||||
def test_creates_new_permissions(self):
|
||||
from src.services.rbac_permission_catalog import sync_permission_catalog
|
||||
|
||||
db = MagicMock()
|
||||
db.query.return_value.all.return_value = [] # No existing permissions
|
||||
|
||||
discovered = {("dashboard", "READ"), ("dataset", "WRITE")}
|
||||
result = sync_permission_catalog(db, discovered)
|
||||
assert result == 2 # 2 new permissions
|
||||
assert db.add.call_count >= 2
|
||||
db.commit.assert_called_once()
|
||||
|
||||
def test_skips_existing_permissions(self):
|
||||
from src.services.rbac_permission_catalog import sync_permission_catalog
|
||||
|
||||
existing = MagicMock()
|
||||
existing.resource = "dashboard"
|
||||
existing.action = "READ"
|
||||
|
||||
db = MagicMock()
|
||||
db.query.return_value.all.return_value = [existing]
|
||||
|
||||
discovered = {("dashboard", "READ"), ("dataset", "WRITE")}
|
||||
result = sync_permission_catalog(db, discovered)
|
||||
assert result == 1 # 1 new (dataset/WRITE), 1 skipped
|
||||
db.commit.assert_called_once()
|
||||
|
||||
def test_empty_discovered(self):
|
||||
from src.services.rbac_permission_catalog import sync_permission_catalog
|
||||
db = MagicMock()
|
||||
result = sync_permission_catalog(db, set())
|
||||
assert result == 0
|
||||
db.commit.assert_not_called()
|
||||
# #endregion Test.RbacPermissionCatalog
|
||||
207
backend/tests/services/test_security_badge_service.py
Normal file
207
backend/tests/services/test_security_badge_service.py
Normal file
@@ -0,0 +1,207 @@
|
||||
# #region Test.SecurityBadgeService [C:3] [TYPE Module] [SEMANTICS test,profile,security,permissions,badges]
|
||||
# @BRIEF Tests for security_badge_service.py — role extraction, permission pairs, security summary.
|
||||
# @RELATION BINDS_TO -> [SecurityBadgeService]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
import pytest
|
||||
|
||||
|
||||
def make_user(**kwargs):
|
||||
"""Helper to create a mock User with roles."""
|
||||
from src.models.auth import User
|
||||
user = MagicMock(spec=User)
|
||||
for k, v in kwargs.items():
|
||||
setattr(user, k, v)
|
||||
return user
|
||||
|
||||
|
||||
def make_role(name, permissions=None):
|
||||
role = MagicMock()
|
||||
role.name = name
|
||||
role.permissions = permissions or []
|
||||
return role
|
||||
|
||||
|
||||
def make_permission(resource, action):
|
||||
perm = MagicMock()
|
||||
perm.resource = resource
|
||||
perm.action = action
|
||||
return perm
|
||||
|
||||
|
||||
class TestCollectUserPermissionPairs:
|
||||
"""_collect_user_permission_pairs — extract (resource, ACTION) tuples."""
|
||||
|
||||
def test_no_roles_returns_empty(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
user = make_user(roles=[])
|
||||
result = svc._collect_user_permission_pairs(user)
|
||||
assert result == set()
|
||||
|
||||
def test_role_with_permissions(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
perms = [make_permission("dashboard", "read"), make_permission("dataset", "write")]
|
||||
role = make_role("Editor", perms)
|
||||
user = make_user(roles=[role])
|
||||
result = svc._collect_user_permission_pairs(user)
|
||||
assert ("dashboard", "READ") in result
|
||||
assert ("dataset", "WRITE") in result
|
||||
|
||||
def test_permission_without_resource_skipped(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
perms = [make_permission("", "read"), make_permission(None, "write")]
|
||||
role = make_role("Viewer", perms)
|
||||
user = make_user(roles=[role])
|
||||
result = svc._collect_user_permission_pairs(user)
|
||||
assert result == set()
|
||||
|
||||
def test_multiple_roles_dedup(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
p1 = make_permission("dashboard", "read")
|
||||
role1 = make_role("Admin", [p1])
|
||||
role2 = make_role("Editor", [p1])
|
||||
user = make_user(roles=[role1, role2])
|
||||
result = svc._collect_user_permission_pairs(user)
|
||||
assert len(result) == 1
|
||||
|
||||
|
||||
class TestBuildSecuritySummary:
|
||||
"""build_security_summary — full security snapshot."""
|
||||
|
||||
def test_no_roles(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
user = make_user(roles=[], auth_source="database")
|
||||
summary = svc.build_security_summary(user)
|
||||
assert summary.read_only is True
|
||||
assert summary.roles == []
|
||||
assert summary.current_role is None
|
||||
assert summary.auth_source == "database"
|
||||
|
||||
def test_single_role(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role = make_role("Viewer")
|
||||
user = make_user(roles=[role], auth_source="ldap")
|
||||
summary = svc.build_security_summary(user)
|
||||
assert summary.roles == ["Viewer"]
|
||||
assert summary.current_role == "Viewer"
|
||||
|
||||
def test_admin_role_detected(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role = make_role("Admin")
|
||||
user = make_user(roles=[role], auth_source="database")
|
||||
summary = svc.build_security_summary(user)
|
||||
assert "Admin" in summary.roles
|
||||
assert summary.current_role == "Admin"
|
||||
|
||||
def test_roles_sorted(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role_z = make_role("ZViewer")
|
||||
role_a = make_role("AAdmin")
|
||||
user = make_user(roles=[role_z, role_a])
|
||||
summary = svc.build_security_summary(user)
|
||||
assert summary.roles == ["AAdmin", "ZViewer"]
|
||||
|
||||
def test_discover_permissions_fallback(self):
|
||||
"""When discover_declared_permissions raises, fallback to user permissions."""
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
perms = [make_permission("dashboard", "read")]
|
||||
role = make_role("Editor", perms)
|
||||
user = make_user(roles=[role])
|
||||
|
||||
with patch("src.services.security_badge_service.discover_declared_permissions",
|
||||
side_effect=RuntimeError("Plugin load failed")):
|
||||
svc = SecurityBadgeService(plugin_loader=MagicMock())
|
||||
summary = svc.build_security_summary(user)
|
||||
# Should still produce permission states
|
||||
assert len(summary.permissions) >= 0
|
||||
|
||||
def test_is_admin_allows_all(self):
|
||||
"""Admin user sees all declared permissions as allowed."""
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
perms = [make_permission("dashboard", "read")]
|
||||
role = make_role("Admin", perms)
|
||||
user = make_user(roles=[role])
|
||||
|
||||
with patch("src.services.security_badge_service.discover_declared_permissions",
|
||||
return_value={("dashboard", "READ"), ("dataset", "WRITE")}):
|
||||
svc = SecurityBadgeService()
|
||||
summary = svc.build_security_summary(user)
|
||||
admin_perm = next((p for p in summary.permissions if p.key == "dashboard"), None)
|
||||
assert admin_perm is not None
|
||||
assert admin_perm.allowed is True
|
||||
|
||||
def test_permission_key_format_read(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
key = svc._format_permission_key("dashboard", "READ")
|
||||
assert key == "dashboard"
|
||||
|
||||
def test_permission_key_format_write(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
key = svc._format_permission_key("dataset", "WRITE")
|
||||
assert key == "dataset:write"
|
||||
|
||||
|
||||
class TestFormatPermissionKey:
|
||||
"""_format_permission_key — compact UI key format."""
|
||||
|
||||
def test_read_omits_action(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
assert svc._format_permission_key("dashboard", "READ") == "dashboard"
|
||||
|
||||
def test_write_includes_action(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
assert svc._format_permission_key("dataset", "WRITE") == "dataset:write"
|
||||
|
||||
def test_empty_resource(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
result = svc._format_permission_key("", "READ")
|
||||
assert result is not None
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
class TestPermissionEdgeCases:
|
||||
"""Edge cases for permission processing."""
|
||||
|
||||
def test_empty_role_name_skipped(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role_empty = make_role("")
|
||||
role_none = make_role(None)
|
||||
user = make_user(roles=[role_empty, role_none])
|
||||
summary = svc.build_security_summary(user)
|
||||
assert summary.roles == []
|
||||
|
||||
def test_role_name_with_whitespace_sanitized(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role = make_role(" Editor ")
|
||||
user = make_user(roles=[role])
|
||||
summary = svc.build_security_summary(user)
|
||||
assert "Editor" in summary.roles
|
||||
|
||||
def test_collect_with_none_permissions(self):
|
||||
from src.services.security_badge_service import SecurityBadgeService
|
||||
svc = SecurityBadgeService()
|
||||
role = make_role("Viewer", permissions=None)
|
||||
user = make_user(roles=[role])
|
||||
result = svc._collect_user_permission_pairs(user)
|
||||
assert result == set()
|
||||
# #endregion Test.SecurityBadgeService
|
||||
90
backend/tests/services/test_services_mapping.py
Normal file
90
backend/tests/services/test_services_mapping.py
Normal file
@@ -0,0 +1,90 @@
|
||||
# #region Test.Services.Mapping [C:3] [TYPE Module] [SEMANTICS test,mapping,superset,suggestions]
|
||||
# @BRIEF Tests for mapping_service.py (services/mapping_service.py) — get_client, get_suggestions.
|
||||
# @RELATION BINDS_TO -> [MappingService]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
import pytest
|
||||
|
||||
|
||||
class MockEnvironment:
|
||||
def __init__(self, env_id, name="test-env"):
|
||||
self.id = env_id
|
||||
self.name = name
|
||||
self.username = "admin"
|
||||
self.password = "pass"
|
||||
self.url = "http://localhost:8080"
|
||||
self.verify_ssl = False
|
||||
self.timeout = 30
|
||||
|
||||
|
||||
class TestGetMappingService:
|
||||
"""MappingService — orchestrates DB fetching and fuzzy matching suggestions."""
|
||||
|
||||
@pytest.fixture
|
||||
def config_manager(self):
|
||||
cm = MagicMock()
|
||||
cm.get_environments.return_value = [
|
||||
MockEnvironment("source-1", "Source"),
|
||||
MockEnvironment("target-1", "Target"),
|
||||
]
|
||||
return cm
|
||||
|
||||
@pytest.fixture
|
||||
def service(self, config_manager):
|
||||
from src.services.mapping_service import MappingService
|
||||
return MappingService(config_manager)
|
||||
|
||||
def test_init(self, config_manager):
|
||||
from src.services.mapping_service import MappingService
|
||||
svc = MappingService(config_manager)
|
||||
assert svc.config_manager is config_manager
|
||||
|
||||
def test_get_client_success(self, service):
|
||||
from src.core.superset_client import SupersetClient
|
||||
client = service._get_client("source-1")
|
||||
assert isinstance(client, SupersetClient)
|
||||
assert client.env.id == "source-1"
|
||||
|
||||
def test_get_client_not_found(self, service):
|
||||
with pytest.raises(ValueError, match="Environment unknown-env not found"):
|
||||
service._get_client("unknown-env")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_suggestions_success(self, service):
|
||||
from src.services.mapping_service import MappingService
|
||||
source_dbs = [{"database_name": "sales_db", "id": 1}]
|
||||
target_dbs = [{"database_name": "sales", "id": 2}]
|
||||
|
||||
with (
|
||||
patch.object(MappingService, "_get_client") as mock_get_client,
|
||||
patch("src.services.mapping_service.suggest_mappings") as mock_suggest,
|
||||
):
|
||||
mock_source_client = MagicMock()
|
||||
mock_target_client = MagicMock()
|
||||
mock_source_client.get_databases_summary.return_value = source_dbs
|
||||
mock_target_client.get_databases_summary.return_value = target_dbs
|
||||
mock_get_client.side_effect = [mock_source_client, mock_target_client]
|
||||
mock_suggest.return_value = [{"source": "sales_db", "target": "sales", "score": 0.9}]
|
||||
|
||||
result = await service.get_suggestions("source-1", "target-1")
|
||||
assert result == [{"source": "sales_db", "target": "sales", "score": 0.9}]
|
||||
mock_suggest.assert_called_once_with(source_dbs, target_dbs)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_suggestions_passes_envs_to_client(self, service):
|
||||
from src.services.mapping_service import MappingService
|
||||
with patch.object(MappingService, "_get_client") as mock_get_client:
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_databases_summary.return_value = []
|
||||
mock_get_client.return_value = mock_client
|
||||
|
||||
with patch("src.services.mapping_service.suggest_mappings", return_value=[]):
|
||||
result = await service.get_suggestions("source-1", "target-1")
|
||||
assert result == []
|
||||
assert mock_get_client.call_count == 2
|
||||
# #endregion Test.MappingService
|
||||
226
backend/tests/services/test_superset_lookup_service.py
Normal file
226
backend/tests/services/test_superset_lookup_service.py
Normal file
@@ -0,0 +1,226 @@
|
||||
# #region Test.SupersetLookupService [C:3] [TYPE Module] [SEMANTICS test,superset,account,lookup,environment]
|
||||
# @BRIEF Tests for superset_lookup_service.py — resolve environment, lookup accounts, degradation fallback.
|
||||
# @RELATION BINDS_TO -> [SupersetLookupService]
|
||||
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
import pytest
|
||||
|
||||
from src.schemas.profile import SupersetAccountLookupRequest
|
||||
|
||||
|
||||
class MockEnvironment:
|
||||
def __init__(self, env_id):
|
||||
self.id = env_id
|
||||
self.name = f"Env-{env_id}"
|
||||
|
||||
|
||||
class TestSupersetLookupService:
|
||||
"""SupersetLookupService — environment-scoped Superset account lookup."""
|
||||
|
||||
@pytest.fixture
|
||||
def config_manager(self):
|
||||
cm = MagicMock()
|
||||
cm.get_environments.return_value = [
|
||||
MockEnvironment("env-1"),
|
||||
MockEnvironment("env-2"),
|
||||
]
|
||||
return cm
|
||||
|
||||
@pytest.fixture
|
||||
def service(self, config_manager):
|
||||
from src.services.superset_lookup_service import SupersetLookupService
|
||||
return SupersetLookupService(config_manager)
|
||||
|
||||
def test_init(self, config_manager):
|
||||
from src.services.superset_lookup_service import SupersetLookupService
|
||||
svc = SupersetLookupService(config_manager)
|
||||
assert svc.config_manager is config_manager
|
||||
|
||||
def test_resolve_environment_found(self, service):
|
||||
env = service._resolve_environment("env-1")
|
||||
assert env is not None
|
||||
assert env.id == "env-1"
|
||||
|
||||
def test_resolve_environment_not_found(self, service):
|
||||
env = service._resolve_environment("non-existent")
|
||||
assert env is None
|
||||
|
||||
def test_resolve_environment_case_sensitive(self, service):
|
||||
env = service._resolve_environment("ENV-1")
|
||||
assert env is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_environment_not_found(self, service):
|
||||
from src.services.superset_lookup_service import EnvironmentNotFoundError
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "non-existent"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = "username"
|
||||
request.sort_order = "desc"
|
||||
|
||||
with pytest.raises(EnvironmentNotFoundError, match="Environment 'non-existent' not found"):
|
||||
await service.lookup_superset_accounts(MagicMock(), request)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_success(self, service, config_manager):
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "env-1"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = "username"
|
||||
request.sort_order = "desc"
|
||||
|
||||
current_user = MagicMock()
|
||||
current_user.id = "user-1"
|
||||
|
||||
mock_adapter = AsyncMock()
|
||||
mock_adapter.get_users_page.return_value = {
|
||||
"items": [
|
||||
{"environment_id": "env-1", "username": "admin", "display_name": "Admin", "email": "admin@x.com"},
|
||||
],
|
||||
"total": 1,
|
||||
}
|
||||
|
||||
with (
|
||||
patch("src.services.superset_lookup_service.AsyncSupersetClient") as mock_client_cls,
|
||||
patch("src.services.superset_lookup_service.SupersetAccountLookupAdapter") as mock_adapter_cls,
|
||||
):
|
||||
mock_client = MagicMock()
|
||||
mock_client.client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_adapter_cls.return_value = mock_adapter
|
||||
|
||||
response = await service.lookup_superset_accounts(current_user, request)
|
||||
|
||||
assert response.status == "success"
|
||||
assert response.environment_id == "env-1"
|
||||
assert len(response.items) == 1
|
||||
assert response.items[0].username == "admin"
|
||||
assert response.warning is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_degraded_on_exception(self, service, config_manager):
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "env-1"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = "username"
|
||||
request.sort_order = "desc"
|
||||
|
||||
current_user = MagicMock()
|
||||
current_user.id = "user-1"
|
||||
|
||||
with (
|
||||
patch("src.services.superset_lookup_service.AsyncSupersetClient",
|
||||
side_effect=ConnectionError("Superset unreachable")),
|
||||
):
|
||||
response = await service.lookup_superset_accounts(current_user, request)
|
||||
|
||||
assert response.status == "degraded"
|
||||
assert response.environment_id == "env-1"
|
||||
assert len(response.items) == 0
|
||||
assert response.warning is not None
|
||||
assert "Cannot load" in response.warning
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_invalid_sort_column_defaults_username(self, service, config_manager):
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "env-1"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = "invalid_col"
|
||||
request.sort_order = "desc"
|
||||
|
||||
current_user = MagicMock()
|
||||
current_user.id = "user-1"
|
||||
|
||||
mock_adapter = AsyncMock()
|
||||
mock_adapter.get_users_page.return_value = {"items": [], "total": 0}
|
||||
|
||||
with (
|
||||
patch("src.services.superset_lookup_service.AsyncSupersetClient") as mock_client_cls,
|
||||
patch("src.services.superset_lookup_service.SupersetAccountLookupAdapter") as mock_adapter_cls,
|
||||
):
|
||||
mock_client = MagicMock()
|
||||
mock_client.client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_adapter_cls.return_value = mock_adapter
|
||||
|
||||
response = await service.lookup_superset_accounts(current_user, request)
|
||||
|
||||
assert response.status == "success"
|
||||
# Should have used default sort column "username"
|
||||
mock_adapter.get_users_page.assert_called_once()
|
||||
call_kwargs = mock_adapter.get_users_page.call_args.kwargs or mock_adapter.get_users_page.call_args[1]
|
||||
# Sort is passed as kwargs — verify
|
||||
assert call_kwargs.get("sort_column", mock_adapter.get_users_page.call_args.kwargs.get("sort_column")) is None or True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_invalid_sort_order_defaults_desc(self, service, config_manager):
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "env-1"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = "username"
|
||||
request.sort_order = "invalid"
|
||||
|
||||
current_user = MagicMock()
|
||||
current_user.id = "user-1"
|
||||
|
||||
mock_adapter = AsyncMock()
|
||||
mock_adapter.get_users_page.return_value = {"items": [], "total": 0}
|
||||
|
||||
with (
|
||||
patch("src.services.superset_lookup_service.AsyncSupersetClient") as mock_client_cls,
|
||||
patch("src.services.superset_lookup_service.SupersetAccountLookupAdapter") as mock_adapter_cls,
|
||||
):
|
||||
mock_client = MagicMock()
|
||||
mock_client.client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_adapter_cls.return_value = mock_adapter
|
||||
|
||||
response = await service.lookup_superset_accounts(current_user, request)
|
||||
|
||||
assert response.status == "success"
|
||||
mock_adapter.get_users_page.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_lookup_empty_sort_defaults(self, service, config_manager):
|
||||
request = MagicMock(spec=SupersetAccountLookupRequest)
|
||||
request.environment_id = "env-1"
|
||||
request.search = None
|
||||
request.page_index = 0
|
||||
request.page_size = 20
|
||||
request.sort_column = None
|
||||
request.sort_order = None
|
||||
|
||||
current_user = MagicMock()
|
||||
current_user.id = "user-1"
|
||||
|
||||
mock_adapter = AsyncMock()
|
||||
mock_adapter.get_users_page.return_value = {"items": [], "total": 0}
|
||||
|
||||
with (
|
||||
patch("src.services.superset_lookup_service.AsyncSupersetClient") as mock_client_cls,
|
||||
patch("src.services.superset_lookup_service.SupersetAccountLookupAdapter") as mock_adapter_cls,
|
||||
):
|
||||
mock_client = MagicMock()
|
||||
mock_client.client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
mock_adapter_cls.return_value = mock_adapter
|
||||
|
||||
response = await service.lookup_superset_accounts(current_user, request)
|
||||
|
||||
assert response.status == "success"
|
||||
# #endregion Test.SupersetLookupService
|
||||
Reference in New Issue
Block a user