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()