205 lines
8.4 KiB
Python
205 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr
|
|
from pytest import MonkeyPatch
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from modelforge_api.db import get_session
|
|
from modelforge_api.domain.configuration_policy import SERVING_OVERRIDE_KEYS
|
|
from modelforge_api.domain.configuration_reference import SETTING_DOCS, Sensitivity
|
|
from modelforge_api.main import app
|
|
from modelforge_api.persistence.models import Base
|
|
from modelforge_api.services.configuration import build_configuration_catalog
|
|
from modelforge_api.settings import Settings, get_settings
|
|
|
|
OPERATOR_TOKEN = "configuration-test-operator-token" # noqa: S105 - synthetic test credential
|
|
DATABASE_SECRET = (
|
|
"postgresql+psycopg://runtime:do-not-leak@db:5432/modelforge" # noqa: S105
|
|
)
|
|
HF_SECRET = "hf_do-not-leak" # noqa: S105 - synthetic test credential
|
|
REDIS_PASSWORD = "redis-do-not-leak" # noqa: S105 - synthetic test credential
|
|
REDIS_URL = f"redis://cache-user:{REDIS_PASSWORD}@redis.internal:6379/0"
|
|
ALERT_WEBHOOK_SECRET = "https://hooks.example.test/token-that-must-not-leak" # noqa: S105
|
|
|
|
|
|
def _settings() -> Settings:
|
|
return Settings(
|
|
_env_file=None,
|
|
env="test",
|
|
operator_api_key=SecretStr(OPERATOR_TOKEN),
|
|
database_url=DATABASE_SECRET,
|
|
hf_token=SecretStr(HF_SECRET),
|
|
redis_url=REDIS_URL,
|
|
alert_webhook_url=SecretStr(ALERT_WEBHOOK_SECRET),
|
|
log_level="WARNING",
|
|
)
|
|
|
|
|
|
def test_catalog_covers_every_runtime_setting_and_never_serializes_secrets() -> None:
|
|
catalog = build_configuration_catalog(_settings())
|
|
|
|
by_key = {entry.key: entry for entry in catalog.settings}
|
|
assert set(by_key) == set(SETTING_DOCS)
|
|
assert catalog.read_only is False
|
|
assert {entry.key for entry in catalog.settings if entry.editable} == SERVING_OVERRIDE_KEYS
|
|
|
|
secret_keys = {
|
|
key
|
|
for key, documentation in SETTING_DOCS.items()
|
|
if documentation.sensitivity is Sensitivity.SECRET
|
|
}
|
|
assert secret_keys
|
|
assert all(by_key[key].value is None for key in secret_keys)
|
|
assert by_key["database_url"].configured is True
|
|
assert by_key["hf_token"].configured is True
|
|
assert by_key["operator_api_key"].configured is True
|
|
assert by_key["alert_webhook_url"].configured is True
|
|
|
|
payload = catalog.model_dump_json()
|
|
assert DATABASE_SECRET not in payload
|
|
assert HF_SECRET not in payload
|
|
assert OPERATOR_TOKEN not in payload
|
|
assert REDIS_PASSWORD not in payload
|
|
assert ALERT_WEBHOOK_SECRET not in payload
|
|
|
|
|
|
def test_catalog_reports_the_actual_pydantic_source_at_each_priority(
|
|
tmp_path: Path, monkeypatch: MonkeyPatch
|
|
) -> None:
|
|
dotenv = tmp_path / "model-forge.env"
|
|
dotenv.write_text(
|
|
"MODELFORGE_LOG_LEVEL=INFO\n"
|
|
"MODELFORGE_GATEWAY_MAX_BATCH_SIZE=12\n"
|
|
"MODELFORGE_GATEWAY_REQUEST_TIMEOUT_SECONDS=100\n"
|
|
"MODELFORGE_HF_TOKEN=hf_dotenv_secret\n"
|
|
"MODELFORGE_ALERT_WEBHOOK_ALLOWED_HOSTS=\n",
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("MODELFORGE_LOG_LEVEL", "WARNING")
|
|
monkeypatch.setenv("MODELFORGE_GATEWAY_REQUEST_TIMEOUT_SECONDS", "111")
|
|
monkeypatch.setenv("MODELFORGE_HF_TOKEN", "hf_process_secret")
|
|
|
|
settings = Settings(
|
|
_env_file=dotenv,
|
|
env="test",
|
|
gateway_request_timeout_seconds=123,
|
|
)
|
|
catalog = build_configuration_catalog(settings)
|
|
by_key = {entry.key: entry for entry in catalog.settings}
|
|
|
|
assert settings.log_level == "WARNING"
|
|
assert by_key["log_level"].source == "process_environment"
|
|
assert by_key["log_level"].value == "WARNING"
|
|
assert settings.gateway_max_batch_size == 12
|
|
assert by_key["gateway_max_batch_size"].source == "dotenv"
|
|
assert by_key["gateway_max_batch_size"].value == 12
|
|
assert settings.gateway_request_timeout_seconds == 123
|
|
assert by_key["gateway_request_timeout_seconds"].source == "initialization"
|
|
assert by_key["cors_origins"].source == "pydantic_default"
|
|
assert by_key["hf_token"].source == "process_environment"
|
|
assert by_key["hf_token"].value is None
|
|
assert by_key["alert_webhook_allowed_hosts"].source == "dotenv"
|
|
assert by_key["alert_webhook_allowed_hosts"].value == ""
|
|
assert settings.setting_source("hf_token") == "process_environment"
|
|
assert "hf_process_secret" not in repr(settings._setting_sources)
|
|
assert "hf_dotenv_secret" not in repr(settings._setting_sources)
|
|
assert "hf_process_secret" not in catalog.model_dump_json()
|
|
assert "hf_dotenv_secret" not in catalog.model_dump_json()
|
|
|
|
effective = settings.model_copy(update={"gateway_max_batch_size": 24})
|
|
overridden = build_configuration_catalog(effective, overrides={"gateway_max_batch_size": 24})
|
|
overridden_setting = next(item for item in overridden.settings if item.key == "gateway_max_batch_size")
|
|
assert overridden_setting.source == "operator_override"
|
|
assert overridden_setting.value == 24
|
|
|
|
monkeypatch.setenv("MODELFORGE_LOG_LEVEL", "")
|
|
empty_environment_value = Settings(_env_file=dotenv, env="test")
|
|
empty_catalog = build_configuration_catalog(empty_environment_value)
|
|
empty_log_level = next(item for item in empty_catalog.settings if item.key == "log_level")
|
|
assert empty_environment_value.log_level == ""
|
|
assert empty_log_level.source == "process_environment"
|
|
assert empty_log_level.value == ""
|
|
|
|
|
|
def test_catalog_reports_a_secrets_file_source_without_returning_its_value(
|
|
tmp_path: Path, monkeypatch: MonkeyPatch
|
|
) -> None:
|
|
secret_dir = tmp_path / "settings-secrets"
|
|
secret_dir.mkdir()
|
|
(secret_dir / "MODELFORGE_LOG_LEVEL").write_text("ERROR", encoding="utf-8")
|
|
monkeypatch.delenv("MODELFORGE_LOG_LEVEL", raising=False)
|
|
|
|
settings = Settings(_env_file=None, _secrets_dir=secret_dir, env="test")
|
|
catalog = build_configuration_catalog(settings)
|
|
log_level = next(item for item in catalog.settings if item.key == "log_level")
|
|
|
|
assert settings.log_level == "ERROR"
|
|
assert log_level.source == "secrets_file"
|
|
assert log_level.value == "ERROR"
|
|
|
|
|
|
def test_settings_copy_and_mutation_never_keep_a_stale_field_source(
|
|
monkeypatch: MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("MODELFORGE_LOG_LEVEL", "WARNING")
|
|
settings = Settings(_env_file=None, env="test")
|
|
|
|
unchanged_copy = settings.model_copy()
|
|
assert unchanged_copy.setting_source("log_level") == "process_environment"
|
|
|
|
copied_update = settings.model_copy(update={"log_level": "DEBUG"})
|
|
assert copied_update.setting_source("log_level") == "runtime_update"
|
|
assert settings.setting_source("log_level") == "process_environment"
|
|
|
|
settings.log_level = "ERROR"
|
|
assert settings.setting_source("log_level") == "runtime_update"
|
|
|
|
reconstructed = Settings.model_validate(settings.model_dump())
|
|
assert reconstructed.setting_source("log_level") == "initialization"
|
|
|
|
|
|
def test_catalog_exposes_the_queue_timeout_cross_setting_validation() -> None:
|
|
catalog = build_configuration_catalog(_settings())
|
|
by_key = {entry.key: entry for entry in catalog.settings}
|
|
|
|
assert by_key["gateway_queue_timeout_seconds"].validation_rules["less_than_setting"] == (
|
|
"gateway_request_timeout_seconds"
|
|
)
|
|
|
|
|
|
def test_configuration_catalog_endpoint_requires_control_plane_auth_and_stays_secret_safe() -> None:
|
|
settings = _settings()
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
session = Session(engine)
|
|
app.dependency_overrides[get_session] = lambda: session
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
client = TestClient(app)
|
|
try:
|
|
anonymous = client.get("/api/v1/configuration/settings")
|
|
assert anonymous.status_code == 401
|
|
|
|
response = client.get(
|
|
"/api/v1/configuration/settings",
|
|
headers={"X-ModelForge-Admin-Token": OPERATOR_TOKEN},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.text
|
|
assert DATABASE_SECRET not in body
|
|
assert HF_SECRET not in body
|
|
assert OPERATOR_TOKEN not in body
|
|
assert REDIS_PASSWORD not in body
|
|
payload = response.json()
|
|
assert payload["environment"] == "test"
|
|
assert payload["read_only"] is False
|
|
assert {item["key"] for item in payload["settings"]} == set(SETTING_DOCS)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
session.close()
|
|
engine.dispose()
|