test_config.py

"""Tests for ``paperscout.config`` validation."""

from __future__ import annotations

import pytest
from pydantic import ValidationError

from paperscout.config import (
    ENV_VAR_MAP,
    Settings,
    legacy_env_name,
    override_settings,
    prefixed_env_name,
    settings,
)
from paperscout.errors import ConfigurationError


def test_settings_rejects_blank_slack_when_not_testing(monkeypatch):
    monkeypatch.delenv("_PAPERSCOUT_TESTING", raising=False)
    with pytest.raises(ConfigurationError, match="Slack is not configured"):
        Settings(
            slack_bot_token="",
            slack_signing_secret="",
        )


def test_settings_accepts_slack_when_not_testing(monkeypatch):
    monkeypatch.delenv("_PAPERSCOUT_TESTING", raising=False)
    s = Settings(
        slack_bot_token="xoxb-real",
        slack_signing_secret="not-empty",
    )
    assert s.slack_bot_token == "xoxb-real"


def test_settings_allows_empty_slack_under_testing_flag(monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    s = Settings(slack_bot_token="", slack_signing_secret="")
    assert s.slack_bot_token == ""


def test_prefixed_process_env_loads(monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.setenv("PAPERSCOUT_POLL_INTERVAL_MINUTES", "99")
    s = Settings()
    assert s.poll_interval_minutes == 99


def test_legacy_process_env_fallback(monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.delenv("PAPERSCOUT_POLL_INTERVAL_MINUTES", raising=False)
    monkeypatch.setenv("POLL_INTERVAL_MINUTES", "88")
    s = Settings()
    assert s.poll_interval_minutes == 88


def test_prefixed_wins_over_legacy(monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.setenv("PAPERSCOUT_POLL_INTERVAL_MINUTES", "99")
    monkeypatch.setenv("POLL_INTERVAL_MINUTES", "88")
    s = Settings()
    assert s.poll_interval_minutes == 99


def test_prefixed_process_env_loads_bool(monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.setenv("PAPERSCOUT_ENABLE_ISO_PROBE", "false")
    s = Settings()
    assert s.enable_iso_probe is False


def test_env_mapping_covers_all_fields():
    assert set(ENV_VAR_MAP.keys()) == set(Settings.model_fields.keys())
    db = ENV_VAR_MAP["database_url"]
    assert db.prefixed == "PAPERSCOUT_DATABASE_URL"
    assert db.legacy == "DATABASE_URL"
    for field_name in Settings.model_fields:
        names = ENV_VAR_MAP[field_name]
        assert names.prefixed == prefixed_env_name(field_name)
        assert names.legacy == legacy_env_name(field_name)


def test_prefixed_keys_in_dotenv_file_are_ignored(tmp_path, monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.delenv("PAPERSCOUT_POLL_INTERVAL_MINUTES", raising=False)
    monkeypatch.delenv("POLL_INTERVAL_MINUTES", raising=False)
    env_file = tmp_path / ".env"
    env_file.write_text("PAPERSCOUT_POLL_INTERVAL_MINUTES=77\n")
    s = Settings(_env_file=env_file)
    assert s.poll_interval_minutes == 30


def test_legacy_keys_in_dotenv_file_load(tmp_path, monkeypatch):
    monkeypatch.setenv("_PAPERSCOUT_TESTING", "1")
    monkeypatch.delenv("PAPERSCOUT_POLL_INTERVAL_MINUTES", raising=False)
    monkeypatch.delenv("POLL_INTERVAL_MINUTES", raising=False)
    env_file = tmp_path / ".env"
    env_file.write_text("POLL_INTERVAL_MINUTES=66\n")
    s = Settings(_env_file=env_file)
    assert s.poll_interval_minutes == 66


class TestOverrideSettings:
    def test_override_settings_applies_within_block(self):
        original = settings.poll_interval_minutes
        with override_settings(poll_interval_minutes=99):
            assert settings.poll_interval_minutes == 99
        assert settings.poll_interval_minutes == original

    def test_override_settings_restores_after_block(self):
        original = settings.wg21_index_timeout_s
        with override_settings(wg21_index_timeout_s=42.0):
            assert settings.wg21_index_timeout_s == 42.0
        assert settings.wg21_index_timeout_s == original

    def test_override_settings_restores_on_exception(self):
        original = settings.poll_interval_minutes
        with pytest.raises(RuntimeError, match="boom"):
            with override_settings(poll_interval_minutes=77):
                assert settings.poll_interval_minutes == 77
                raise RuntimeError("boom")
        assert settings.poll_interval_minutes == original

    def test_override_settings_nested(self):
        original = settings.poll_interval_minutes
        with override_settings(poll_interval_minutes=10):
            assert settings.poll_interval_minutes == 10
            with override_settings(poll_interval_minutes=20):
                assert settings.poll_interval_minutes == 20
            assert settings.poll_interval_minutes == 10
        assert settings.poll_interval_minutes == original

    def test_override_settings_rejects_unknown_field(self):
        with pytest.raises(TypeError, match="Unknown settings field"):
            with override_settings(not_a_real_field=1):
                pass

    def test_override_settings_rejects_invalid_value(self):
        with pytest.raises(ValidationError):
            with override_settings(wg21_index_timeout_s=-1):
                pass

    def test_importers_see_override(self):
        from paperscout.config import settings as imported_settings

        assert imported_settings is settings
        original = settings.enable_iso_probe
        with override_settings(enable_iso_probe=not original):
            assert imported_settings.enable_iso_probe is not original
        assert imported_settings.enable_iso_probe == original

    def test_override_settings_rejects_concurrent_threads(self):
        import threading

        import paperscout.config as config_module

        a_inside = threading.Event()
        b_inside = threading.Event()
        a_may_exit = threading.Event()
        errors: list[RuntimeError] = []
        original = settings.poll_interval_minutes

        def thread_a() -> None:
            try:
                with override_settings(poll_interval_minutes=111):
                    a_inside.set()
                    b_inside.wait(timeout=5)
            except RuntimeError as exc:
                errors.append(exc)
            finally:
                a_may_exit.set()

        def thread_b() -> None:
            with override_settings(poll_interval_minutes=222):
                b_inside.set()
                a_may_exit.wait(timeout=5)

        try:
            t_a = threading.Thread(target=thread_a)
            t_b = threading.Thread(target=thread_b)
            t_a.start()
            assert a_inside.wait(timeout=5)
            t_b.start()
            assert b_inside.wait(timeout=5)
            t_a.join(timeout=5)
            t_b.join(timeout=5)
            assert not t_a.is_alive()
            assert not t_b.is_alive()
            assert errors, "expected concurrent override exit to raise RuntimeError"
            assert any("non-LIFO or concurrent" in str(exc) for exc in errors)
        finally:
            config_module._override_stack.clear()
            settings.poll_interval_minutes = original