busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
52a987e415cb49fc9dc158c2090476938d7b9392
ss-tools/backend/tests/api/test_settings_consolidated.py
busya 52a987e415 feat(settings): expose tunable runtime settings with server-side validation 
Move hardcoded constants into GlobalSettings and surface them in the
Settings UI: task retention, auth rate limit, assistant history retention,
translate baseline expiry, default environment, and extended logging fields.

- consolidated settings API: new fields in GET/PATCH with re-validation
  through GlobalSettings (422 on out-of-range instead of silent persist)
- rate limiter policy read live from settings with 60s cache + lock-free
  fast path; cache invalidated centrally in ConfigManager on auth policy
  change (covers PATCH /settings/global and /consolidated)
- shared settings_provider.get_global_settings() replaces three copies of
  the fallback pattern; scheduler baseline fallback derives from model
  default
- remove dead GlobalSettings fields (pagination_limit, ff_dataset_*,
  LLM_*_RETENTION_DAYS, GLOBAL_VALIDATION_WORKER_LIMIT, AppAsyncRuntimeConfig)
- SystemSettings blocks save on out-of-range values; LoggingSettings gains
  max_bytes/backup_count/agent_view/hide_routine_infra/log_level_for_agents;
  EnvironmentsTab gains default environment selector
- tests: rate limiter settings-driven policy, consolidated PATCH 422 paths,
  System tab save-blocking UX test
2026-08-02 23:51:32 +07:00

658 lines
28 KiB
Python
Raw Blame History

#region Test.SettingsConsolidated.TestSettingsConsolidated [C:3] [TYPE Module] [SEMANTICS test,settings,connections]
# @BRIEF Tests for GET/PATCH /api/settings/consolidated — connections typed-model enforcement.
# @RELATION BINDS_TO -> [get_consolidated_settings, update_consolidated_settings]
# @RELATION DEPENDS_ON -> [EXT:FastAPI:TestClient]
# @TEST_CONTRACT [GlobalSettings.connections] -> [ConsolidatedSettingsResponse | 200]
# @TEST_EDGE: dict_in_connections -> GET survives list[dict] in connections (no .model_dump crash)
# @TEST_EDGE: model_in_connections -> GET works with typed DatabaseConnection models
# @TEST_EDGE: empty_connections -> GET returns empty list
# @TEST_EDGE: patch_saves_as_model -> PATCH stores DatabaseConnection instances, not raw dicts
# @TEST_EDGE: patch_masked_password -> PATCH preserves existing password on "********"
# @TEST_EDGE: patch_new_password -> PATCH encrypts new passwords via ConnectionService
# @TEST_EDGE: unauthorized -> 403 when user lacks admin role
# @TEST_INVARIANT connections_typed -> VERIFIED_BY: patch_saves_as_model, patch_masked_password
# @INVARIANT current_settings.connections must always be list[DatabaseConnection], never list[dict]
# Set required env vars before ANY app imports — crash-early guard for AuthConfig()
import os
import tempfile
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing")
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:test_auth_settings_consolidated")
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:test_main_settings_consolidated")
os.environ.setdefault("DEV_MODE", "true")
import pytest
from unittest.mock import MagicMock
from fastapi.testclient import TestClient
from src.app import app
from src.core.config_models import (
DatabaseConnection,
FeaturesConfig,
GlobalSettings,
LoggingConfig,
)
from src.dependencies import (
get_config_manager,
get_current_user,
)
# ── Shared mocks ───────────────────────────────────────────────────────
mock_user = MagicMock()
mock_user.username = "testuser"
mock_user.roles = []
admin_role = MagicMock()
admin_role.name = "Admin"
admin_role.permissions = []
mock_user.roles.append(admin_role)
non_admin_role = MagicMock()
non_admin_role.name = "User"
non_admin_role.permissions = []
non_admin_role.is_admin = False
def _make_conn_dict(**overrides) -> dict:
"""Build a raw connection dict as the frontend would send it."""
return {
"id": overrides.get("id", "conn-1"),
"name": overrides.get("name", "Test DB"),
"host": overrides.get("host", "localhost"),
"port": overrides.get("port", 5432),
"database": overrides.get("database", "testdb"),
"username": overrides.get("username", "admin"),
"password": overrides.get("password", "secret"),
"dialect": overrides.get("dialect", "postgresql"),
"extra_params": overrides.get("extra_params", {}),
"pool_size": overrides.get("pool_size", 5),
}
def _make_conn_model(**overrides) -> DatabaseConnection:
"""Build a typed DatabaseConnection model."""
return DatabaseConnection(**_make_conn_dict(**overrides))
# ── Fixtures ───────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def mock_deps():
"""Override FastAPI dependencies for consolidated-settings routes.
get_current_user returns an Admin user by default — permission checks
pass through the real has_permission function.
Sets up a real GlobalSettings so typed-field assignment is testable.
"""
mock_user.roles = [admin_role]
# Real GlobalSettings with known state
real_settings = GlobalSettings()
real_settings.features = FeaturesConfig(translate=True, migration=False)
real_settings.logging = LoggingConfig()
# AppConfig mock — .settings is a real GlobalSettings;
# .environments is empty to avoid MagicMock noise in responses
app_config = MagicMock()
app_config.settings = real_settings
app_config.environments = []
mock_config_manager = MagicMock()
mock_config_manager.get_config.return_value = app_config
app.dependency_overrides[get_config_manager] = lambda: mock_config_manager
app.dependency_overrides[get_current_user] = lambda: mock_user
yield {
"config": mock_config_manager,
"settings": real_settings,
}
app.dependency_overrides.clear()
client = TestClient(app)
# ── GET /api/settings/consolidated ─────────────────────────────────────
#region Test.SettingsConsolidated.TestGetConsolidatedWithModelConnections [C:2] [TYPE Function]
# @BRIEF GET returns 200 when connections contains DatabaseConnection models.
# @TEST_INVARIANT connections_typed -> VERIFIED_BY: test_get_consolidated_with_model_connections
def test_get_consolidated_with_model_connections(mock_deps):
"""DatabaseConnection models in settings.connections → 200, all passwords masked."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = [
_make_conn_model(id="c1", name="Alpha", password="real-pwd-1"),
_make_conn_model(id="c2", name="Beta", password="real-pwd-2"),
]
response = client.get("/api/settings/consolidated")
assert response.status_code == 200
data = response.json()
assert len(data["connections"]) == 2
for entry in data["connections"]:
assert entry["password"] == "********", "Passwords must be masked in response"
assert data["connections"][0]["name"] == "Alpha"
assert data["connections"][1]["name"] == "Beta"
assert data["features"]["translate"] is True
assert data["features"]["migration"] is False
#endregion Test.SettingsConsolidated.TestGetConsolidatedWithModelConnections
#region Test.SettingsConsolidated.TestGetConsolidatedWithDictConnections [C:2] [TYPE Function]
# @BRIEF GET handles defensive case where connections contains raw dicts from previous corruption.
# @TEST_EDGE: dict_in_connections -> GET does not crash
def test_get_consolidated_with_dict_connections(mock_deps):
"""Raw dicts in settings.connections → 200 (no .model_dump crash)."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = [ # ← dicts, not DatabaseConnection models
{"id": "c1", "name": "Alpha", "password": "p1", "host": "h1", "port": 5432, "database": "d1", "username": "u1", "dialect": "postgresql"},
{"id": "c2", "name": "Beta", "password": "p2", "host": "h2", "port": 5432, "database": "d2", "username": "u2", "dialect": "postgresql"},
]
response = client.get("/api/settings/consolidated")
assert response.status_code == 200
data = response.json()
assert len(data["connections"]) == 2
for entry in data["connections"]:
assert entry["password"] == "********"
assert data["connections"][0]["name"] == "Alpha"
#endregion Test.SettingsConsolidated.TestGetConsolidatedWithDictConnections
#region Test.SettingsConsolidated.TestGetConsolidatedEmptyConnections [C:2] [TYPE Function]
# @BRIEF GET returns empty list when no connections configured.
# @TEST_EDGE: empty_connections -> GET returns []
def test_get_consolidated_empty_connections(mock_deps):
"""Empty connections list → 200, connections=[], nothing crashes."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = []
response = client.get("/api/settings/consolidated")
assert response.status_code == 200
data = response.json()
assert data["connections"] == []
#endregion Test.SettingsConsolidated.TestGetConsolidatedEmptyConnections
#region Test.SettingsConsolidated.TestGetConsolidatedUnauthorized [C:2] [TYPE Function]
# @BRIEF GET returns 403 when user lacks admin:settings:READ permission.
# @TEST_EDGE: unauthorized -> 403
def test_get_consolidated_unauthorized():
"""Non-admin user → 403."""
mock_user.roles = [non_admin_role]
response = client.get("/api/settings/consolidated")
assert response.status_code == 403
#endregion Test.SettingsConsolidated.TestGetConsolidatedUnauthorized
# ── PATCH /api/settings/consolidated ───────────────────────────────────
#region Test.SettingsConsolidated.TestPatchConnectionsSavesAsModel [C:2] [TYPE Function]
# @BRIEF PATCH with connections dicts stores them as DatabaseConnection models.
# @TEST_INVARIANT connections_typed -> VERIFIED_BY: test_patch_connections_saves_as_model
def test_patch_connections_saves_as_model(mock_deps):
"""After PATCH, current_settings.connections contains DatabaseConnection instances, not dicts."""
mock_config = mock_deps["config"]
payload = {
"connections": [
_make_conn_dict(id="c1", name="Prod DB", password="p@ss"),
]
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
# Verify via update_global_settings call — connections must be typed
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert len(updated_settings.connections) == 1
conn = updated_settings.connections[0]
assert isinstance(conn, DatabaseConnection), (
f"Expected DatabaseConnection, got {type(conn).__name__}"
)
assert conn.id == "c1"
assert conn.name == "Prod DB"
assert conn.host == "localhost"
assert conn.dialect == "postgresql"
#endregion Test.SettingsConsolidated.TestPatchConnectionsSavesAsModel
#region Test.SettingsConsolidated.TestPatchConnectionsMaskedPassword [C:2] [TYPE Function]
# @BRIEF PATCH with "********" password preserves the existing encrypted password.
# @TEST_EDGE: patch_masked_password -> existing password not overwritten
def test_patch_connections_masked_password(mock_deps):
"""Masked password "********" → keeps encrypted password from existing config."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = [
_make_conn_model(id="c-exist", name="Existing", password="encrypted-vault-value"),
]
mock_config = mock_deps["config"]
payload = {
"connections": [
_make_conn_dict(id="c-exist", name="Existing", password="********"),
]
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
conn = updated_settings.connections[0]
assert conn.password == "encrypted-vault-value", (
"Must preserve existing password when masked password sent"
)
#endregion Test.SettingsConsolidated.TestPatchConnectionsMaskedPassword
#region Test.SettingsConsolidated.TestPatchConnectionsNewPasswordEncrypted [C:2] [TYPE Function]
# @BRIEF PATCH with a new plaintext password encrypts it via ConnectionService.
# @TEST_EDGE: patch_new_password -> encrypts via ConnectionService._encrypt_password
def test_patch_connections_new_password_encrypted(mock_deps):
"""New plaintext password → ConnectionService._encrypt_password called."""
settings: GlobalSettings = mock_deps["settings"]
# Existing connection with an encrypted password
settings.connections = [
_make_conn_model(id="c-old", name="Old Conn", password="stale-encrypted"),
]
mock_config = mock_deps["config"]
payload = {
"connections": [
_make_conn_dict(id="c-old", name="Old Conn", password="new-plain-password"),
]
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
# Password is NOT "********" and NOT "stale-encrypted" — it goes through
# ConnectionService._encrypt_password which (without real EncryptionManager)
# returns the password as-is or encrypts it. At minimum we verify it changed.
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
conn = updated_settings.connections[0]
assert conn.password != "********", "New password must not be masked"
assert conn.password != "stale-encrypted", "New password must replace old encrypted value"
assert conn.name == "Old Conn"
#endregion Test.SettingsConsolidated.TestPatchConnectionsNewPasswordEncrypted
#region Test.SettingsConsolidated.TestPatchConnectionsMultipleAtOnce [C:2] [TYPE Function]
# @BRIEF PATCH processes multiple connections in one request.
def test_patch_connections_multiple_at_once(mock_deps):
"""Three connections in one PATCH → all stored as DatabaseConnection models."""
mock_config = mock_deps["config"]
payload = {
"connections": [
_make_conn_dict(id="c1", name="Alpha", password="p1"),
_make_conn_dict(id="c2", name="Beta", password="p2"),
_make_conn_dict(id="c3", name="Gamma", password="p3"),
]
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert len(updated_settings.connections) == 3
for conn in updated_settings.connections:
assert isinstance(conn, DatabaseConnection)
assert [c.name for c in updated_settings.connections] == ["Alpha", "Beta", "Gamma"]
#endregion Test.SettingsConsolidated.TestPatchConnectionsMultipleAtOnce
#region Test.SettingsConsolidated.TestPatchConnectionsEmptyList [C:2] [TYPE Function]
# @BRIEF PATCH with empty connections list clears the list.
def test_patch_connections_empty_list(mock_deps):
"""Empty connections list in PATCH → connections cleared."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = [
_make_conn_model(id="c1", name="To Delete", password="secret"),
]
mock_config = mock_deps["config"]
payload = {"connections": []}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert updated_settings.connections == []
#endregion Test.SettingsConsolidated.TestPatchConnectionsEmptyList
#region Test.SettingsConsolidated.TestPatchSettingsWithoutConnections [C:2] [TYPE Function]
# @BRIEF PATCH without connections only updates other settings (features, timezone).
def test_patch_settings_without_connections(mock_deps):
"""No 'connections' key in payload → connections left untouched."""
settings: GlobalSettings = mock_deps["settings"]
settings.connections = [
_make_conn_model(id="c1", name="Existing Conn", password="secret"),
]
mock_config = mock_deps["config"]
payload = {
"features": {"translate": False, "migration": True},
"app_timezone": "Asia/Tokyo",
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
# Connections preserved
assert len(updated_settings.connections) == 1
assert updated_settings.connections[0].name == "Existing Conn"
# Features updated
assert updated_settings.features.translate is False
assert updated_settings.features.migration is True
# Timezone updated
assert updated_settings.app_timezone == "Asia/Tokyo"
#endregion Test.SettingsConsolidated.TestPatchSettingsWithoutConnections
#region Test.SettingsConsolidated.TestPatchUnauthorized [C:2] [TYPE Function]
# @BRIEF PATCH returns 403 when user lacks admin:settings:WRITE permission.
# @TEST_EDGE: unauthorized -> 403
def test_patch_unauthorized():
"""Non-admin user → 403."""
mock_user.roles = [non_admin_role]
payload = {"features": {"translate": False}}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 403
#endregion Test.SettingsConsolidated.TestPatchUnauthorized
# ── Edge-case patches for 98%+ coverage ──────────────────────────────────────
#region Test.SettingsConsolidated.TestPatchSettingsLlm [C:2]
# @BRIEF PATCH with llm section updates LLM settings (covers line 569).
def test_patch_settings_llm(mock_deps):
"""Update LLM settings via consolidated PATCH."""
mock_config = mock_deps["config"]
payload = {
"llm": {
"default_provider": "openai",
"providers": [
{"name": "openai", "api_key": "sk-new-key", "base_url": "https://custom.api.com", "default_model": "gpt-4o"},
],
}
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert updated_settings.llm["default_provider"] == "openai"
#endregion Test.SettingsConsolidated.TestPatchSettingsLlm
#region Test.SettingsConsolidated.TestPatchSettingsLogging [C:2]
# @BRIEF PATCH with logging section updates LoggingConfig (covers line 573).
def test_patch_settings_logging(mock_deps):
"""Update logging settings."""
mock_config = mock_deps["config"]
payload = {
"logging": {
"level": "DEBUG",
"format": "json",
}
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert updated_settings.logging.level == "DEBUG"
#endregion Test.SettingsConsolidated.TestPatchSettingsLogging
#region Test.SettingsConsolidated.TestPatchSettingsStorage [C:2]
# @BRIEF PATCH with storage section validates path (covers lines 577-581).
def test_patch_settings_storage(mock_deps):
"""Update storage settings with valid path."""
mock_config = mock_deps["config"]
mock_config.validate_path.return_value = (True, "")
with tempfile.TemporaryDirectory() as tmp:
payload = {
"storage": {
"root_path": tmp,
}
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
#endregion Test.SettingsConsolidated.TestPatchSettingsStorage
#region Test.SettingsConsolidated.TestPatchSettingsStorageInvalidPath [C:2]
# @BRIEF PATCH with invalid storage path returns 400 (covers line 579-580).
def test_patch_settings_storage_invalid_path(mock_deps):
"""Invalid storage path returns 400."""
mock_config = mock_deps["config"]
mock_config.validate_path.return_value = (False, "Path is not writable")
payload = {
"storage": {
"root_path": "/invalid/path",
}
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 400
assert "Path is not writable" in response.json()["detail"]
#endregion Test.SettingsConsolidated.TestPatchSettingsStorageInvalidPath
#region Test.SettingsConsolidated.TestPatchSettingsNotifications [C:2]
# @BRIEF PATCH with notifications section updates the payload (covers lines 590-592).
def test_patch_settings_notifications(mock_deps):
"""Update notifications settings."""
mock_config = mock_deps["config"]
mock_config.get_payload.return_value = {}
payload = {
"notifications": {
"email": {"enabled": True, "smtp_host": "smtp.example.com"},
}
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.save_config.assert_called_once()
args, _ = mock_config.save_config.call_args
assert "notifications" in args[0]
assert args[0]["notifications"]["email"]["enabled"] is True
#endregion Test.SettingsConsolidated.TestPatchSettingsNotifications
#region Test.SettingsConsolidated.TestPatchSettingsTimezoneInvalid [C:2]
# @BRIEF PATCH with invalid timezone returns 422 (covers line 599).
def test_patch_settings_timezone_invalid(mock_deps):
"""Invalid IANA timezone returns 422."""
payload = {
"app_timezone": "Invalid/Timezone",
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 422
assert "Invalid IANA timezone" in response.json()["detail"]
#endregion Test.SettingsConsolidated.TestPatchSettingsTimezoneInvalid
#region Test.SettingsConsolidated.TestPatchSettingsAllowedLanguages [C:2]
# @BRIEF PATCH with allowed_languages updates the list (covers line 605).
def test_patch_settings_allowed_languages(mock_deps):
"""Update allowed_languages list."""
mock_config = mock_deps["config"]
payload = {
"allowed_languages": ["en", "fr", "de"],
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert updated_settings.allowed_languages == ["en", "fr", "de"]
#endregion Test.SettingsConsolidated.TestPatchSettingsAllowedLanguages
#region Test.SettingsConsolidated.TestGetConsolidatedNotificationsPayload [C:2]
# @BRIEF GET consolidated includes notifications from AppConfigRecord (covers line 483).
def test_get_consolidated_notifications_payload(mock_deps):
"""GET includes notifications payload from DB config record."""
settings: GlobalSettings = mock_deps["settings"]
response = client.get("/api/settings/consolidated")
assert response.status_code == 200
data = response.json()
assert "notifications" in data
#endregion Test.SettingsConsolidated.TestGetConsolidatedNotificationsPayload
#region Test.SettingsConsolidated.TestPatchConnectionsExistingDicts [C:2]
# @BRIEF PATCH connections handles existing connections as dicts (covers line 550-551).
def test_patch_connections_existing_dicts(mock_deps):
"""When current_settings.connections contains dicts, handles gracefully."""
settings: GlobalSettings = mock_deps["settings"]
# Set existing connections as raw dicts (corrupted state)
settings.connections = [
{"id": "c-exist", "name": "Existing", "password": "encrypted-pwd",
"host": "localhost", "port": 5432, "database": "db1", "username": "u1",
"dialect": "postgresql", "extra_params": {}, "pool_size": 5},
]
mock_config = mock_deps["config"]
payload = {
"connections": [
{"id": "c-exist", "name": "Existing", "password": "********",
"host": "localhost", "port": 5432, "database": "db1", "username": "u1",
"dialect": "postgresql", "extra_params": {}, "pool_size": 5},
]
}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_config.update_global_settings.assert_called_once()
args, _ = mock_config.update_global_settings.call_args
updated_settings: GlobalSettings = args[0]
assert len(updated_settings.connections) == 1
conn = updated_settings.connections[0]
assert isinstance(conn, DatabaseConnection), "Dicts must be converted to DatabaseConnection"
assert conn.password == "encrypted-pwd", "Must preserve existing encrypted password"
#endregion Test.SettingsConsolidated.TestPatchConnectionsExistingDicts
#region Test.SettingsConsolidated.TestNormalizeUrlEdgeCases [C:2]
# @BRIEF Test URL normalization edge cases (covers lines 63, 67).
def test_normalize_url_with_api_v1_suffix():
"""URL ending with /api/v1 is stripped."""
from src.api.routes.settings import _normalize_superset_env_url
result = _normalize_superset_env_url("https://example.com/api/v1")
assert result == "https://example.com"
assert not result.endswith("/api/v1")
def test_normalize_url_without_scheme():
"""URL without scheme gets https:// prepended."""
from src.api.routes.settings import _normalize_superset_env_url
result = _normalize_superset_env_url("example.com")
assert result == "https://example.com"
def test_normalize_url_empty_string():
"""Empty string returns empty string."""
from src.api.routes.settings import _normalize_superset_env_url
result = _normalize_superset_env_url("")
assert result == ""
def test_normalize_url_already_normalized():
"""Already normalized URL passes through."""
from src.api.routes.settings import _normalize_superset_env_url
result = _normalize_superset_env_url("https://example.com")
assert result == "https://example.com"
#endregion Test.SettingsConsolidated.TestNormalizeUrlEdgeCases
#region Test.SettingsConsolidated.TestGetConnectionServiceDependency [C:2]
# @BRIEF _get_connection_service returns ConnectionService (covers line 751).
def test_get_connection_service_dependency():
"""_get_connection_service returns ConnectionService with config_manager."""
from src.api.routes.settings import _get_connection_service
from src.core.config_manager import ConfigManager
from src.core.connection_service import ConnectionService
cm = MagicMock(spec=ConfigManager)
result = _get_connection_service(config_manager=cm)
assert isinstance(result, ConnectionService)
#endregion Test.SettingsConsolidated.TestGetConnectionServiceDependency
#region Test.SettingsConsolidated.TestPatchOutOfRangeRejected [C:2] [TYPE Function]
# @BRIEF Out-of-range auth/task values via consolidated PATCH are rejected with 422.
# @TEST_EDGE: out_of_range_rejected -> raw dict assignment must not bypass Field ge/le
def test_patch_out_of_range_rejected(mock_deps):
"""auth_max_attempts=5000 must be rejected — otherwise the rate limiter would
silently accept it, the persisted config would fail validation on restart and
fall back to defaults (environment loss)."""
payload = {"auth_max_attempts": 5000}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 422
# Nothing may be persisted when the patch is invalid
mock_deps["config"].update_global_settings.assert_not_called()
#endregion Test.SettingsConsolidated.TestPatchOutOfRangeRejected
#region Test.SettingsConsolidated.TestPatchInRangeAccepted [C:2] [TYPE Function]
# @BRIEF In-range values still pass through and reach update_global_settings.
def test_patch_in_range_accepted(mock_deps):
"""auth_max_attempts=25 is within 1..100 and must be applied."""
payload = {"auth_max_attempts": 25}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 200
mock_deps["config"].update_global_settings.assert_called_once()
args, _ = mock_deps["config"].update_global_settings.call_args
assert args[0].auth_max_attempts == 25
#endregion Test.SettingsConsolidated.TestPatchInRangeAccepted
#region Test.SettingsConsolidated.TestPatchZeroTaskRetentionRejected [C:2] [TYPE Function]
# @BRIEF task_retention_days=0 would delete all tasks on next cleanup — must be rejected.
# @TEST_EDGE: zero_retention_rejected -> lower bound enforced
def test_patch_zero_task_retention_rejected(mock_deps):
"""task_retention_days=0 violates ge=1 and must 422."""
payload = {"task_retention_days": 0}
response = client.patch("/api/settings/consolidated", json=payload)
assert response.status_code == 422
mock_deps["config"].update_global_settings.assert_not_called()
#endregion Test.SettingsConsolidated.TestPatchZeroTaskRetentionRejected
#endregion Test.SettingsConsolidated.TestSettingsConsolidated
Reference in New Issue View Git Blame Copy Permalink