898 lines
32 KiB
Python
898 lines
32 KiB
Python
"""Unit tests for src/config.py — Settings, validation, and helper functions.
|
|
|
|
Refs #1172
|
|
"""
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_settings(**env_overrides):
|
|
"""Create a fresh Settings instance with isolated env vars."""
|
|
from config import Settings
|
|
|
|
# Prevent Pydantic from reading .env file (local .env pollutes defaults)
|
|
_orig_config = Settings.model_config.copy()
|
|
Settings.model_config["env_file"] = None
|
|
|
|
# Strip keys that might bleed in from the test environment
|
|
clean_env = {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if not k.startswith(
|
|
(
|
|
"OLLAMA_",
|
|
"TIMMY_",
|
|
"AGENT_",
|
|
"DEBUG",
|
|
"GITEA_",
|
|
"GROK_",
|
|
"ANTHROPIC_",
|
|
"SPARK_",
|
|
"MEMORY_",
|
|
"MAX_",
|
|
"DISCORD_",
|
|
"TELEGRAM_",
|
|
"CORS_",
|
|
"TRUSTED_",
|
|
"L402_",
|
|
"LIGHTNING_",
|
|
"REPO_ROOT",
|
|
"RQLITE_",
|
|
"BRAIN_",
|
|
"SELF_MODIFY",
|
|
"WORK_ORDERS",
|
|
"VASSAL_",
|
|
"PAPERCLIP_",
|
|
"OPENFANG_",
|
|
"HERMES_",
|
|
"BACKLOG_",
|
|
"LOOP_QA",
|
|
"FOCUS_",
|
|
"THINKING_",
|
|
"HANDS_",
|
|
"WEEKLY_",
|
|
"AUTORESEARCH_",
|
|
"REWARD_",
|
|
"BROWSER_",
|
|
"GABS_",
|
|
"SCRIPTURE_",
|
|
"MCP_",
|
|
"CHAT_API",
|
|
"CSRF_",
|
|
"ERROR_",
|
|
"DB_",
|
|
"MODERATION_",
|
|
"SOVEREIGNTY_",
|
|
"XAI_",
|
|
"CLAUDE_",
|
|
"FLUX_",
|
|
"IMAGE_",
|
|
"MUSIC_",
|
|
"VIDEO_",
|
|
"CREATIVE_",
|
|
"WAN_",
|
|
"ACE_",
|
|
"GIT_",
|
|
)
|
|
)
|
|
}
|
|
clean_env.update(env_overrides)
|
|
with patch.dict(os.environ, clean_env, clear=True):
|
|
try:
|
|
return Settings()
|
|
finally:
|
|
Settings.model_config.update(_orig_config)
|
|
|
|
|
|
# ── normalize_ollama_url ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestNormalizeOllamaUrl:
|
|
"""normalize_ollama_url replaces localhost with 127.0.0.1."""
|
|
|
|
def test_replaces_localhost(self):
|
|
from config import normalize_ollama_url
|
|
|
|
assert normalize_ollama_url("http://localhost:11434") == "http://127.0.0.1:11434"
|
|
|
|
def test_preserves_ip_address(self):
|
|
from config import normalize_ollama_url
|
|
|
|
assert normalize_ollama_url("http://192.168.1.5:11434") == "http://192.168.1.5:11434"
|
|
|
|
def test_preserves_non_localhost_hostname(self):
|
|
from config import normalize_ollama_url
|
|
|
|
assert normalize_ollama_url("http://ollama.local:11434") == "http://ollama.local:11434"
|
|
|
|
def test_replaces_multiple_occurrences(self):
|
|
from config import normalize_ollama_url
|
|
|
|
result = normalize_ollama_url("http://localhost:11434/localhost")
|
|
assert result == "http://127.0.0.1:11434/127.0.0.1"
|
|
|
|
def test_empty_string(self):
|
|
from config import normalize_ollama_url
|
|
|
|
assert normalize_ollama_url("") == ""
|
|
|
|
def test_127_0_0_1_unchanged(self):
|
|
from config import normalize_ollama_url
|
|
|
|
url = "http://127.0.0.1:11434"
|
|
assert normalize_ollama_url(url) == url
|
|
|
|
|
|
# ── Settings defaults ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSettingsDefaults:
|
|
"""Settings instantiation produces correct defaults."""
|
|
|
|
def test_default_agent_name(self):
|
|
s = _make_settings()
|
|
assert s.agent_name == "Agent"
|
|
|
|
def test_default_ollama_url(self):
|
|
s = _make_settings()
|
|
assert s.ollama_url == "http://localhost:11434"
|
|
|
|
def test_default_ollama_model(self):
|
|
s = _make_settings()
|
|
assert s.ollama_model == "qwen3:14b"
|
|
|
|
def test_default_ollama_fast_model(self):
|
|
s = _make_settings()
|
|
assert s.ollama_fast_model == "qwen3:8b"
|
|
|
|
def test_default_ollama_num_ctx(self):
|
|
s = _make_settings()
|
|
assert s.ollama_num_ctx == 32768
|
|
|
|
def test_default_ollama_max_loaded_models(self):
|
|
s = _make_settings()
|
|
assert s.ollama_max_loaded_models == 2
|
|
|
|
def test_default_debug_false(self):
|
|
s = _make_settings()
|
|
assert s.debug is False
|
|
|
|
def test_default_timmy_env(self):
|
|
s = _make_settings()
|
|
assert s.timmy_env == "development"
|
|
|
|
def test_default_timmy_test_mode_false(self):
|
|
s = _make_settings()
|
|
assert s.timmy_test_mode is False
|
|
|
|
def test_default_spark_enabled(self):
|
|
s = _make_settings()
|
|
assert s.spark_enabled is True
|
|
|
|
def test_default_lightning_backend(self):
|
|
s = _make_settings()
|
|
assert s.lightning_backend == "mock"
|
|
|
|
def test_default_max_agent_steps(self):
|
|
s = _make_settings()
|
|
assert s.max_agent_steps == 10
|
|
|
|
def test_default_memory_prune_days(self):
|
|
s = _make_settings()
|
|
assert s.memory_prune_days == 90
|
|
|
|
def test_default_memory_prune_keep_facts(self):
|
|
s = _make_settings()
|
|
assert s.memory_prune_keep_facts is True
|
|
|
|
def test_default_fallback_models_is_list(self):
|
|
s = _make_settings()
|
|
assert isinstance(s.fallback_models, list)
|
|
assert len(s.fallback_models) > 0
|
|
|
|
def test_default_vision_fallback_models_is_list(self):
|
|
s = _make_settings()
|
|
assert isinstance(s.vision_fallback_models, list)
|
|
assert len(s.vision_fallback_models) > 0
|
|
|
|
def test_default_cors_origins_is_list(self):
|
|
s = _make_settings()
|
|
assert isinstance(s.cors_origins, list)
|
|
assert len(s.cors_origins) > 0
|
|
|
|
def test_default_trusted_hosts_is_list(self):
|
|
s = _make_settings()
|
|
assert isinstance(s.trusted_hosts, list)
|
|
assert "localhost" in s.trusted_hosts
|
|
|
|
def test_default_timmy_model_backend(self):
|
|
s = _make_settings()
|
|
assert s.timmy_model_backend == "ollama"
|
|
|
|
def test_default_grok_enabled_false(self):
|
|
s = _make_settings()
|
|
assert s.grok_enabled is False
|
|
|
|
def test_default_moderation_enabled(self):
|
|
s = _make_settings()
|
|
assert s.moderation_enabled is True
|
|
|
|
def test_default_moderation_threshold(self):
|
|
s = _make_settings()
|
|
assert s.moderation_threshold == 0.8
|
|
|
|
def test_default_telemetry_disabled(self):
|
|
s = _make_settings()
|
|
assert s.telemetry_enabled is False
|
|
|
|
def test_default_db_busy_timeout(self):
|
|
s = _make_settings()
|
|
assert s.db_busy_timeout_ms == 5000
|
|
|
|
def test_default_chat_api_max_body_bytes(self):
|
|
s = _make_settings()
|
|
assert s.chat_api_max_body_bytes == 1_048_576
|
|
|
|
def test_default_csrf_cookie_secure_false(self):
|
|
s = _make_settings()
|
|
assert s.csrf_cookie_secure is False
|
|
|
|
def test_default_self_modify_disabled(self):
|
|
s = _make_settings()
|
|
assert s.self_modify_enabled is False
|
|
|
|
def test_default_vassal_disabled(self):
|
|
s = _make_settings()
|
|
assert s.vassal_enabled is False
|
|
|
|
def test_default_focus_mode(self):
|
|
s = _make_settings()
|
|
assert s.focus_mode == "broad"
|
|
|
|
def test_default_thinking_enabled(self):
|
|
s = _make_settings()
|
|
assert s.thinking_enabled is True
|
|
|
|
def test_default_gitea_url(self):
|
|
s = _make_settings()
|
|
assert s.gitea_url == "http://localhost:3000"
|
|
|
|
def test_default_hermes_enabled(self):
|
|
s = _make_settings()
|
|
assert s.hermes_enabled is True
|
|
|
|
def test_default_scripture_enabled(self):
|
|
s = _make_settings()
|
|
assert s.scripture_enabled is True
|
|
|
|
|
|
# ── normalized_ollama_url property ───────────────────────────────────────────
|
|
|
|
|
|
class TestNormalizedOllamaUrlProperty:
|
|
"""normalized_ollama_url property applies normalize_ollama_url."""
|
|
|
|
def test_default_url_normalized(self):
|
|
s = _make_settings()
|
|
assert "127.0.0.1" in s.normalized_ollama_url
|
|
assert "localhost" not in s.normalized_ollama_url
|
|
|
|
def test_custom_url_with_localhost(self):
|
|
s = _make_settings(OLLAMA_URL="http://localhost:9999")
|
|
assert s.normalized_ollama_url == "http://127.0.0.1:9999"
|
|
|
|
def test_custom_url_without_localhost_unchanged(self):
|
|
s = _make_settings(OLLAMA_URL="http://192.168.1.5:11434")
|
|
assert s.normalized_ollama_url == "http://192.168.1.5:11434"
|
|
|
|
|
|
# ── Env var overrides ─────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSettingsEnvOverrides:
|
|
"""Environment variables override default values."""
|
|
|
|
def test_agent_name_override(self):
|
|
s = _make_settings(AGENT_NAME="Timmy")
|
|
assert s.agent_name == "Timmy"
|
|
|
|
def test_ollama_url_override(self):
|
|
s = _make_settings(OLLAMA_URL="http://10.0.0.1:11434")
|
|
assert s.ollama_url == "http://10.0.0.1:11434"
|
|
|
|
def test_ollama_model_override(self):
|
|
s = _make_settings(OLLAMA_MODEL="llama3.1")
|
|
assert s.ollama_model == "llama3.1"
|
|
|
|
def test_ollama_fast_model_override(self):
|
|
s = _make_settings(OLLAMA_FAST_MODEL="gemma:2b")
|
|
assert s.ollama_fast_model == "gemma:2b"
|
|
|
|
def test_ollama_num_ctx_override(self):
|
|
s = _make_settings(OLLAMA_NUM_CTX="8192")
|
|
assert s.ollama_num_ctx == 8192
|
|
|
|
def test_debug_true_from_string(self):
|
|
s = _make_settings(DEBUG="true")
|
|
assert s.debug is True
|
|
|
|
def test_debug_false_from_string(self):
|
|
s = _make_settings(DEBUG="false")
|
|
assert s.debug is False
|
|
|
|
def test_timmy_env_production(self):
|
|
s = _make_settings(TIMMY_ENV="production")
|
|
assert s.timmy_env == "production"
|
|
|
|
def test_timmy_test_mode_true(self):
|
|
s = _make_settings(TIMMY_TEST_MODE="true")
|
|
assert s.timmy_test_mode is True
|
|
|
|
def test_grok_enabled_override(self):
|
|
s = _make_settings(GROK_ENABLED="true")
|
|
assert s.grok_enabled is True
|
|
|
|
def test_spark_enabled_override(self):
|
|
s = _make_settings(SPARK_ENABLED="false")
|
|
assert s.spark_enabled is False
|
|
|
|
def test_memory_prune_days_override(self):
|
|
s = _make_settings(MEMORY_PRUNE_DAYS="30")
|
|
assert s.memory_prune_days == 30
|
|
|
|
def test_max_agent_steps_override(self):
|
|
s = _make_settings(MAX_AGENT_STEPS="25")
|
|
assert s.max_agent_steps == 25
|
|
|
|
def test_telegram_token_override(self):
|
|
s = _make_settings(TELEGRAM_TOKEN="tg-secret")
|
|
assert s.telegram_token == "tg-secret"
|
|
|
|
def test_discord_token_override(self):
|
|
s = _make_settings(DISCORD_TOKEN="dc-secret")
|
|
assert s.discord_token == "dc-secret"
|
|
|
|
def test_gitea_url_override(self):
|
|
s = _make_settings(GITEA_URL="http://10.0.0.1:3000")
|
|
assert s.gitea_url == "http://10.0.0.1:3000"
|
|
|
|
def test_gitea_repo_override(self):
|
|
s = _make_settings(GITEA_REPO="myorg/myrepo")
|
|
assert s.gitea_repo == "myorg/myrepo"
|
|
|
|
def test_focus_mode_deep(self):
|
|
s = _make_settings(FOCUS_MODE="deep")
|
|
assert s.focus_mode == "deep"
|
|
|
|
def test_thinking_interval_override(self):
|
|
s = _make_settings(THINKING_INTERVAL_SECONDS="60")
|
|
assert s.thinking_interval_seconds == 60
|
|
|
|
def test_hermes_interval_override(self):
|
|
s = _make_settings(HERMES_INTERVAL_SECONDS="60")
|
|
assert s.hermes_interval_seconds == 60
|
|
|
|
def test_vassal_enabled_override(self):
|
|
s = _make_settings(VASSAL_ENABLED="true")
|
|
assert s.vassal_enabled is True
|
|
|
|
def test_self_modify_enabled_override(self):
|
|
s = _make_settings(SELF_MODIFY_ENABLED="true")
|
|
assert s.self_modify_enabled is True
|
|
|
|
def test_moderation_enabled_override(self):
|
|
s = _make_settings(MODERATION_ENABLED="false")
|
|
assert s.moderation_enabled is False
|
|
|
|
def test_l402_hmac_secret_override(self):
|
|
s = _make_settings(L402_HMAC_SECRET="mysecret")
|
|
assert s.l402_hmac_secret == "mysecret"
|
|
|
|
def test_anthropic_api_key_override(self):
|
|
s = _make_settings(ANTHROPIC_API_KEY="sk-ant-abc")
|
|
assert s.anthropic_api_key == "sk-ant-abc"
|
|
|
|
|
|
# ── Type validation ───────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSettingsTypeValidation:
|
|
"""Pydantic correctly parses and validates types from string env vars."""
|
|
|
|
def test_bool_from_1(self):
|
|
s = _make_settings(DEBUG="1")
|
|
assert s.debug is True
|
|
|
|
def test_bool_from_0(self):
|
|
s = _make_settings(DEBUG="0")
|
|
assert s.debug is False
|
|
|
|
def test_int_field_rejects_non_numeric(self):
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
_make_settings(OLLAMA_NUM_CTX="not_a_number")
|
|
|
|
def test_timmy_env_rejects_invalid_literal(self):
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
_make_settings(TIMMY_ENV="staging")
|
|
|
|
def test_timmy_model_backend_rejects_invalid(self):
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
_make_settings(TIMMY_MODEL_BACKEND="openai")
|
|
|
|
def test_timmy_model_backend_accepts_all_valid_values(self):
|
|
for backend in ("ollama", "grok", "claude", "auto"):
|
|
s = _make_settings(TIMMY_MODEL_BACKEND=backend)
|
|
assert s.timmy_model_backend == backend
|
|
|
|
def test_lightning_backend_accepts_mock(self):
|
|
s = _make_settings(LIGHTNING_BACKEND="mock")
|
|
assert s.lightning_backend == "mock"
|
|
|
|
def test_lightning_backend_accepts_lnd(self):
|
|
s = _make_settings(LIGHTNING_BACKEND="lnd")
|
|
assert s.lightning_backend == "lnd"
|
|
|
|
def test_lightning_backend_rejects_invalid(self):
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
_make_settings(LIGHTNING_BACKEND="stripe")
|
|
|
|
def test_focus_mode_rejects_invalid(self):
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
_make_settings(FOCUS_MODE="zen")
|
|
|
|
def test_extra_fields_ignored(self):
|
|
# model_config has extra="ignore"
|
|
s = _make_settings(TOTALLY_UNKNOWN_FIELD="hello")
|
|
assert not hasattr(s, "totally_unknown_field")
|
|
|
|
def test_float_field_moderation_threshold(self):
|
|
s = _make_settings(MODERATION_THRESHOLD="0.95")
|
|
assert s.moderation_threshold == pytest.approx(0.95)
|
|
|
|
def test_float_field_gabs_timeout(self):
|
|
s = _make_settings(GABS_TIMEOUT="10.5")
|
|
assert s.gabs_timeout == pytest.approx(10.5)
|
|
|
|
|
|
# ── Edge cases ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSettingsEdgeCases:
|
|
"""Edge cases: empty strings, boundary values."""
|
|
|
|
def test_empty_string_tokens_stay_empty(self):
|
|
s = _make_settings(TELEGRAM_TOKEN="", DISCORD_TOKEN="")
|
|
assert s.telegram_token == ""
|
|
assert s.discord_token == ""
|
|
|
|
def test_zero_int_fields(self):
|
|
s = _make_settings(OLLAMA_NUM_CTX="0", MEMORY_PRUNE_DAYS="0")
|
|
assert s.ollama_num_ctx == 0
|
|
assert s.memory_prune_days == 0
|
|
|
|
def test_large_int_value(self):
|
|
s = _make_settings(CHAT_API_MAX_BODY_BYTES="104857600")
|
|
assert s.chat_api_max_body_bytes == 104857600
|
|
|
|
def test_negative_int_accepted(self):
|
|
# Pydantic doesn't constrain these to positive by default
|
|
s = _make_settings(MAX_AGENT_STEPS="-1")
|
|
assert s.max_agent_steps == -1
|
|
|
|
def test_empty_api_keys_are_strings(self):
|
|
s = _make_settings()
|
|
assert isinstance(s.anthropic_api_key, str)
|
|
assert isinstance(s.xai_api_key, str)
|
|
assert isinstance(s.gitea_token, str)
|
|
|
|
|
|
# ── _compute_repo_root ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestComputeRepoRoot:
|
|
"""_compute_repo_root auto-detects .git directory."""
|
|
|
|
def test_returns_non_empty_string(self):
|
|
from config import Settings
|
|
|
|
s = Settings()
|
|
result = s._compute_repo_root()
|
|
assert isinstance(result, str)
|
|
assert len(result) > 0
|
|
|
|
def test_explicit_repo_root_returned_directly(self):
|
|
from config import Settings
|
|
|
|
s = Settings()
|
|
s.repo_root = "/tmp/custom-repo"
|
|
assert s._compute_repo_root() == "/tmp/custom-repo"
|
|
|
|
def test_detects_git_directory(self):
|
|
from config import Settings
|
|
|
|
s = Settings()
|
|
result = s._compute_repo_root()
|
|
import os
|
|
|
|
# The detected root should contain a .git directory (or be the cwd fallback)
|
|
assert os.path.isabs(result)
|
|
|
|
|
|
# ── model_post_init / gitea_token file fallback ───────────────────────────────
|
|
|
|
|
|
class TestModelPostInit:
|
|
"""model_post_init resolves gitea_token from file fallback."""
|
|
|
|
def test_gitea_token_from_env(self):
|
|
from config import Settings
|
|
|
|
with patch.dict(os.environ, {"GITEA_TOKEN": "env-token-abc"}, clear=False):
|
|
s = Settings()
|
|
assert s.gitea_token == "env-token-abc"
|
|
|
|
def test_gitea_token_stays_empty_when_no_file(self):
|
|
from config import Settings
|
|
|
|
env = {k: v for k, v in os.environ.items() if k != "GITEA_TOKEN"}
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with patch("os.path.isfile", return_value=False):
|
|
s = Settings()
|
|
assert s.gitea_token == ""
|
|
|
|
def test_gitea_token_read_from_timmy_token_file(self, tmp_path):
|
|
"""model_post_init reads token from .timmy_gitea_token file."""
|
|
from config import Settings
|
|
|
|
token_file = tmp_path / ".timmy_gitea_token"
|
|
token_file.write_text("file-token-xyz\n")
|
|
|
|
env = {k: v for k, v in os.environ.items() if k != "GITEA_TOKEN"}
|
|
with patch.dict(os.environ, env, clear=True):
|
|
s = Settings()
|
|
|
|
# Override repo_root so post_init finds our temp file
|
|
def _fake_root():
|
|
return str(tmp_path)
|
|
|
|
s._compute_repo_root = _fake_root # type: ignore[method-assign]
|
|
# Re-run post_init logic manually since Settings is already created
|
|
s.gitea_token = ""
|
|
repo_root = _fake_root()
|
|
token_path = os.path.join(repo_root, ".timmy_gitea_token")
|
|
if os.path.isfile(token_path):
|
|
s.gitea_token = open(token_path).read().strip() # noqa: SIM115
|
|
assert s.gitea_token == "file-token-xyz"
|
|
|
|
def test_gitea_token_empty_file_stays_empty(self, tmp_path):
|
|
"""Empty token file leaves gitea_token as empty string."""
|
|
token_file = tmp_path / ".timmy_gitea_token"
|
|
token_file.write_text(" \n") # only whitespace
|
|
|
|
from config import Settings
|
|
|
|
env = {k: v for k, v in os.environ.items() if k != "GITEA_TOKEN"}
|
|
with patch.dict(os.environ, env, clear=True):
|
|
s = Settings()
|
|
# Simulate post_init with the tmp dir
|
|
s.gitea_token = ""
|
|
token_path = str(token_file)
|
|
if os.path.isfile(token_path):
|
|
token = open(token_path).read().strip() # noqa: SIM115
|
|
if token:
|
|
s.gitea_token = token
|
|
assert s.gitea_token == ""
|
|
|
|
|
|
# ── check_ollama_model_available ──────────────────────────────────────────────
|
|
|
|
|
|
class TestCheckOllamaModelAvailable:
|
|
"""check_ollama_model_available handles network responses and errors."""
|
|
|
|
def test_returns_false_on_oserror(self):
|
|
from config import check_ollama_model_available
|
|
|
|
with patch("urllib.request.urlopen", side_effect=OSError("Connection refused")):
|
|
assert check_ollama_model_available("llama3.1") is False
|
|
|
|
def test_returns_false_on_value_error(self):
|
|
from config import check_ollama_model_available
|
|
|
|
with patch("urllib.request.urlopen", side_effect=ValueError("Bad JSON")):
|
|
assert check_ollama_model_available("llama3.1") is False
|
|
|
|
def test_returns_true_exact_model_match(self):
|
|
import json
|
|
|
|
from config import check_ollama_model_available
|
|
|
|
response_data = json.dumps({"models": [{"name": "llama3.1:8b-instruct"}]}).encode()
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = response_data
|
|
mock_response.__enter__ = lambda s: s
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_response):
|
|
assert check_ollama_model_available("llama3.1") is True
|
|
|
|
def test_returns_true_startswith_match(self):
|
|
import json
|
|
|
|
from config import check_ollama_model_available
|
|
|
|
response_data = json.dumps({"models": [{"name": "qwen3:14b"}]}).encode()
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = response_data
|
|
mock_response.__enter__ = lambda s: s
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_response):
|
|
# "qwen3" matches "qwen3:14b" via startswith
|
|
assert check_ollama_model_available("qwen3") is True
|
|
|
|
def test_returns_false_when_model_not_found(self):
|
|
import json
|
|
|
|
from config import check_ollama_model_available
|
|
|
|
response_data = json.dumps({"models": [{"name": "qwen2.5:7b"}]}).encode()
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = response_data
|
|
mock_response.__enter__ = lambda s: s
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_response):
|
|
assert check_ollama_model_available("llama3.1") is False
|
|
|
|
def test_returns_false_empty_model_list(self):
|
|
import json
|
|
|
|
from config import check_ollama_model_available
|
|
|
|
response_data = json.dumps({"models": []}).encode()
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = response_data
|
|
mock_response.__enter__ = lambda s: s
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_response):
|
|
assert check_ollama_model_available("llama3.1") is False
|
|
|
|
def test_exact_name_match(self):
|
|
import json
|
|
|
|
from config import check_ollama_model_available
|
|
|
|
response_data = json.dumps({"models": [{"name": "qwen3:14b"}]}).encode()
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = response_data
|
|
mock_response.__enter__ = lambda s: s
|
|
mock_response.__exit__ = MagicMock(return_value=False)
|
|
|
|
with patch("urllib.request.urlopen", return_value=mock_response):
|
|
assert check_ollama_model_available("qwen3:14b") is True
|
|
|
|
|
|
# ── get_effective_ollama_model ────────────────────────────────────────────────
|
|
|
|
|
|
class TestGetEffectiveOllamaModel:
|
|
"""get_effective_ollama_model walks fallback chain."""
|
|
|
|
def test_returns_primary_when_available(self):
|
|
from config import get_effective_ollama_model, settings
|
|
|
|
with patch("config.check_ollama_model_available", return_value=True):
|
|
result = get_effective_ollama_model()
|
|
# Should return whatever the settings primary model is
|
|
assert result == settings.ollama_model
|
|
|
|
def test_falls_back_when_primary_unavailable(self):
|
|
from config import get_effective_ollama_model, settings
|
|
|
|
# Make primary unavailable, but one fallback available
|
|
fallback_target = settings.fallback_models[0]
|
|
|
|
def side_effect(model):
|
|
return model == fallback_target
|
|
|
|
with patch("config.check_ollama_model_available", side_effect=side_effect):
|
|
result = get_effective_ollama_model()
|
|
assert result == fallback_target
|
|
|
|
def test_returns_user_model_when_nothing_available(self):
|
|
from config import get_effective_ollama_model, settings
|
|
|
|
with patch("config.check_ollama_model_available", return_value=False):
|
|
result = get_effective_ollama_model()
|
|
# Last resort: returns user's configured model
|
|
assert result == settings.ollama_model
|
|
|
|
def test_skips_unavailable_fallbacks(self):
|
|
from config import get_effective_ollama_model, settings
|
|
|
|
# Only the last fallback is available
|
|
fallbacks = settings.fallback_models
|
|
last_fallback = fallbacks[-1]
|
|
|
|
def side_effect(model):
|
|
return model == last_fallback
|
|
|
|
with patch("config.check_ollama_model_available", side_effect=side_effect):
|
|
result = get_effective_ollama_model()
|
|
assert result == last_fallback
|
|
|
|
|
|
# ── validate_startup ──────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestValidateStartup:
|
|
"""validate_startup enforces security in production, warns in dev."""
|
|
|
|
def setup_method(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
|
|
def test_skips_in_test_mode(self):
|
|
import config
|
|
|
|
with patch.dict(os.environ, {"TIMMY_TEST_MODE": "1"}):
|
|
config.validate_startup()
|
|
assert config._startup_validated is True
|
|
|
|
def test_dev_mode_does_not_exit(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
env = {k: v for k, v in os.environ.items() if k != "TIMMY_TEST_MODE"}
|
|
env["TIMMY_ENV"] = "development"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
# Should not raise SystemExit
|
|
config.validate_startup()
|
|
assert config._startup_validated is True
|
|
|
|
def test_production_exits_without_l402_hmac_secret(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
with patch.object(config.settings, "timmy_env", "production"):
|
|
with patch.object(config.settings, "l402_hmac_secret", ""):
|
|
with patch.object(config.settings, "l402_macaroon_secret", ""):
|
|
with pytest.raises(SystemExit):
|
|
config.validate_startup(force=True)
|
|
|
|
def test_production_exits_without_l402_macaroon_secret(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
with patch.object(config.settings, "timmy_env", "production"):
|
|
with patch.object(config.settings, "l402_hmac_secret", "present"):
|
|
with patch.object(config.settings, "l402_macaroon_secret", ""):
|
|
with pytest.raises(SystemExit):
|
|
config.validate_startup(force=True)
|
|
|
|
def test_production_exits_with_cors_wildcard(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
with patch.object(config.settings, "timmy_env", "production"):
|
|
with patch.object(config.settings, "l402_hmac_secret", "secret1"):
|
|
with patch.object(config.settings, "l402_macaroon_secret", "secret2"):
|
|
with patch.object(config.settings, "cors_origins", ["*"]):
|
|
with pytest.raises(SystemExit):
|
|
config.validate_startup(force=True)
|
|
|
|
def test_production_passes_with_all_secrets_and_no_wildcard(self):
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
with patch.object(config.settings, "timmy_env", "production"):
|
|
with patch.object(config.settings, "l402_hmac_secret", "secret1"):
|
|
with patch.object(config.settings, "l402_macaroon_secret", "secret2"):
|
|
with patch.object(config.settings, "cors_origins", ["http://localhost:3000"]):
|
|
config.validate_startup(force=True)
|
|
assert config._startup_validated is True
|
|
|
|
def test_idempotent_without_force(self):
|
|
import config
|
|
|
|
config._startup_validated = True
|
|
config.validate_startup()
|
|
assert config._startup_validated is True
|
|
|
|
def test_force_reruns_when_already_validated(self):
|
|
import config
|
|
|
|
config._startup_validated = True
|
|
with patch.dict(os.environ, {"TIMMY_TEST_MODE": "1"}):
|
|
config.validate_startup(force=True)
|
|
# Should have run (and set validated again)
|
|
assert config._startup_validated is True
|
|
|
|
def test_dev_warns_on_cors_wildcard(self, caplog):
|
|
import logging
|
|
|
|
import config
|
|
|
|
config._startup_validated = False
|
|
env = {k: v for k, v in os.environ.items() if k != "TIMMY_TEST_MODE"}
|
|
env["TIMMY_ENV"] = "development"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with patch.object(config.settings, "timmy_env", "development"):
|
|
with patch.object(config.settings, "cors_origins", ["*"]):
|
|
with patch.object(config.settings, "l402_hmac_secret", ""):
|
|
with patch.object(config.settings, "l402_macaroon_secret", ""):
|
|
with caplog.at_level(logging.WARNING, logger="config"):
|
|
config.validate_startup(force=True)
|
|
assert any("CORS" in rec.message for rec in caplog.records)
|
|
|
|
|
|
# ── APP_START_TIME ────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestAppStartTime:
|
|
"""APP_START_TIME is set at module load."""
|
|
|
|
def test_app_start_time_is_datetime(self):
|
|
from datetime import datetime
|
|
|
|
from config import APP_START_TIME
|
|
|
|
assert isinstance(APP_START_TIME, datetime)
|
|
|
|
def test_app_start_time_has_utc_timezone(self):
|
|
from config import APP_START_TIME
|
|
|
|
assert APP_START_TIME.tzinfo is not None
|
|
|
|
def test_app_start_time_is_in_the_past_or_now(self):
|
|
from datetime import UTC, datetime
|
|
|
|
from config import APP_START_TIME
|
|
|
|
assert APP_START_TIME <= datetime.now(UTC)
|
|
|
|
|
|
# ── Module-level singleton ────────────────────────────────────────────────────
|
|
|
|
|
|
class TestSettingsSingleton:
|
|
"""The module-level `settings` singleton is a Settings instance."""
|
|
|
|
def test_settings_is_settings_instance(self):
|
|
from config import Settings, settings
|
|
|
|
assert isinstance(settings, Settings)
|
|
|
|
def test_settings_repo_root_is_set(self):
|
|
from config import settings
|
|
|
|
assert isinstance(settings.repo_root, str)
|
|
|
|
def test_settings_has_expected_defaults(self):
|
|
from config import settings
|
|
|
|
# In test mode these may be overridden, but type should be correct
|
|
assert isinstance(settings.ollama_url, str)
|
|
assert isinstance(settings.debug, bool)
|