fix: validate application configuration contracts

AGENTS.md

@@ -76,6 +76,7 @@ Usage notes:
 - Prefer timezone-aware Pendulum values in Python. Reject ambiguous timestamps, keep timezone assumptions at provider boundaries, and centralize unavoidable provider-specific fallback rules instead of spreading guesses through business logic.
 - Do not retain compatibility paths for abandoned internal formats unless the current requirements explicitly require them.
 - Treat persistence, counters, telemetry, and alert bookkeeping performed while handling an error as secondary operations. Their failure must not replace the original business exception, and logic must not rely on state that was not recorded successfully.
+- Validate configuration at its input boundary without coercing invalid scalar types into strings or other superficially valid values. Reject unknown values for application-owned fixed choices early, while leaving third-party dynamic provider namespaces to their owning SDK instead of duplicating a whitelist.
 - Do not use `typing.cast()` in application or test code. Model type boundaries with protocols, typed test doubles, or runtime narrowing instead of suppressing type mismatches.
 - Keep code comments concise and in English.
 - Preserve compatibility between build and runtime environments rather than assuming copied artifacts are portable across distributions or interpreter builds.

tests/test_cli.py

@@ -12,6 +12,7 @@
 
 from weather_briefing.cli import (
     _LOGGER,
+    PUBLISHER_BUILDERS,
     WEATHER_PROVIDER_BUILDERS,
     _aqicn_provider,
     _briefing_delivery_policy,
@@ -38,6 +39,7 @@
 )
 from weather_briefing.config import Settings
 from weather_briefing.models import LocationSpec, ResolvedLocation
+from weather_briefing.registries import PublisherName, WeatherProviderName
 from weather_briefing.state import SQLiteStateStore
 
 
@@ -1098,8 +1100,9 @@ def test_parse_run_time_returns_now_when_value_is_none(monkeypatch) -> None:
     assert result.timezone_name == "Asia/Shanghai"
 
 
-def test_weather_provider_builders_contains_expected_keys() -> None:
-    assert set(WEATHER_PROVIDER_BUILDERS) == {"qweather", "open-meteo"}
+def test_runtime_provider_builders_cover_declared_configuration_names() -> None:
+    assert set(WEATHER_PROVIDER_BUILDERS) == set(WeatherProviderName)
+    assert set(PUBLISHER_BUILDERS) == set(PublisherName)
 
 
 async def test_daemon_schedules_forecast_and_briefing_without_running_either_immediately(monkeypatch) -> None:

tests/test_config.py

@@ -106,18 +106,18 @@ def test_rss_source_requires_public_display_name(monkeypatch, tmp_path: Path) ->
     )
     monkeypatch.setenv("RSS_SOURCES_FILE", str(source_file))
 
-    with pytest.raises(ConfigurationError, match="public display name"):
+    with pytest.raises(ConfigurationError, match=r"rss-sources\.json\[0\]\.name must be a non-empty string"):
         Settings.from_env()
 
 
 @pytest.mark.parametrize(
     ("source", "message"),
     (
-        ('{"name":"Test","url":"https://example.invalid/feed"}', "must have an id"),
-        ('{"id":null,"name":"Test","url":"https://example.invalid/feed"}', "must have an id"),
-        ('{"id":"test","name":null,"url":"https://example.invalid/feed"}', "public display name"),
-        ('{"id":"test","name":"Test"}', "must have a URL"),
-        ('{"id":"test","name":"Test","url":null}', "must have a URL"),
+        ('{"name":"Test","url":"https://example.invalid/feed"}', "id"),
+        ('{"id":null,"name":"Test","url":"https://example.invalid/feed"}', "id"),
+        ('{"id":"test","name":null,"url":"https://example.invalid/feed"}', "name"),
+        ('{"id":"test","name":"Test"}', "url"),
+        ('{"id":"test","name":"Test","url":null}', "url"),
     ),
 )
 def test_rss_source_rejects_missing_or_null_required_fields(
@@ -131,7 +131,32 @@ def test_rss_source_rejects_missing_or_null_required_fields(
     source_file.write_text(f"[{source}]", encoding="utf-8")
     monkeypatch.setenv("RSS_SOURCES_FILE", str(source_file))
 
-    with pytest.raises(ConfigurationError, match=message):
+    with pytest.raises(
+        ConfigurationError,
+        match=rf"rss-sources\.json\[0\]\.{message} must be a non-empty string",
+    ):
+        Settings.from_env()
+
+
+@pytest.mark.parametrize("field", ("id", "name", "url"))
+@pytest.mark.parametrize("value", (1, ["value"], {"value": "nested"}))
+def test_rss_source_required_fields_reject_non_strings(
+    monkeypatch,
+    tmp_path: Path,
+    field: str,
+    value: object,
+) -> None:
+    _required_environment(monkeypatch)
+    source = {"id": "test", "name": "Test", "url": "https://example.invalid/feed"}
+    source[field] = value
+    source_file = tmp_path / "rss-sources.json"
+    source_file.write_text(json.dumps([source]), encoding="utf-8")
+    monkeypatch.setenv("RSS_SOURCES_FILE", str(source_file))
+
+    with pytest.raises(
+        ConfigurationError,
+        match=rf"rss-sources\.json\[0\]\.{field} must be a non-empty string",
+    ):
         Settings.from_env()
 
 
@@ -286,6 +311,29 @@ def test_location_file_supports_name_coordinates_or_both(monkeypatch, tmp_path:
     assert settings.locations[2].name is None
 
 
+@pytest.mark.parametrize("field", ("id", "name"))
+@pytest.mark.parametrize("value", (1, ["value"], {"value": "nested"}))
+def test_location_string_fields_reject_non_strings(
+    monkeypatch,
+    tmp_path: Path,
+    field: str,
+    value: object,
+) -> None:
+    _required_environment(monkeypatch)
+    location = {"id": "test", "name": "Test"}
+    location[field] = value
+    location_file = tmp_path / "locations.json"
+    location_file.write_text(json.dumps([location]), encoding="utf-8")
+    monkeypatch.setenv("BRIEFING_LOCATIONS_FILE", str(location_file))
+    monkeypatch.setenv("RSS_SOURCES_FILE", str(tmp_path / "rss-sources.json"))
+
+    with pytest.raises(
+        ConfigurationError,
+        match=rf"locations\.json\[0\]\.{field} must be a non-empty string",
+    ):
+        Settings.from_env()
+
+
 def test_rss_source_location_ids_must_reference_configured_locations(monkeypatch, tmp_path: Path) -> None:
     _required_environment(monkeypatch)
     location_file = tmp_path / "locations.json"
@@ -720,6 +768,23 @@ def test_empty_weather_providers_raises_error(self, monkeypatch) -> None:
         with pytest.raises(ConfigurationError, match="WEATHER_PROVIDERS cannot be empty"):
             Settings.from_env()
 
+    def test_unsupported_weather_providers_raise_error(self, monkeypatch) -> None:
+        _required_environment(monkeypatch)
+        monkeypatch.setenv("WEATHER_PROVIDERS", "qweather,typo,unknown")
+
+        with pytest.raises(
+            ConfigurationError,
+            match="WEATHER_PROVIDERS contains unsupported providers: typo, unknown",
+        ):
+            Settings.from_env()
+
+    def test_unsupported_publisher_raises_error(self, monkeypatch) -> None:
+        _required_environment(monkeypatch)
+        monkeypatch.setenv("PUBLISHER", "telegrm")
+
+        with pytest.raises(ConfigurationError, match="PUBLISHER must be one of: stdout, telegram"):
+            Settings.from_env()
+
     def test_invalid_json_in_rss_sources_file_raises_error(self, monkeypatch, tmp_path: Path) -> None:
         _required_environment(monkeypatch)
         source_file = tmp_path / "rss-sources.json"

weather_briefing/cli.py

@@ -39,6 +39,7 @@
 )
 from .models import ResolvedLocation
 from .publishers import DeliveryProvider, RenderedTextDiagnostics, StdoutPublisher, TelegramPublisher
+from .registries import PublisherName, WeatherProviderName
 from .render import PlainTextRenderer, TelegramHTMLRenderer
 from .service import BriefingService
 from .sources import HTTPContextSource, RSSSource
@@ -309,23 +310,47 @@ def _delivery_provider(
     client: httpx.AsyncClient,
     diagnostics: RenderedTextDiagnostics | None = None,
 ) -> DeliveryProvider:
-    if settings.publisher == "stdout":
-        return DeliveryProvider(PlainTextRenderer(), StdoutPublisher(), diagnostics=diagnostics)
-    if settings.publisher == "telegram":
-        if not settings.telegram_bot_token or not settings.telegram_chat_id:
-            raise ValueError("Telegram publisher requires TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID")
-        return DeliveryProvider(
-            TelegramHTMLRenderer(),
-            TelegramPublisher(
-                client,
-                settings.telegram_bot_token,
-                settings.telegram_chat_id,
-                diagnostics,
-            ),
-            single_message_limit=TelegramPublisher.MAX_MESSAGE_LENGTH,
-            diagnostics=diagnostics,
-        )
-    raise ValueError(f"Unsupported publisher: {settings.publisher}")
+    builder = PUBLISHER_BUILDERS.get(settings.publisher)
+    if builder is None:
+        raise ValueError(f"Unsupported publisher: {settings.publisher}")
+    return builder(settings, client, diagnostics)
+
+
+def _build_stdout_publisher(
+    settings: Settings,
+    client: httpx.AsyncClient,
+    diagnostics: RenderedTextDiagnostics | None,
+) -> DeliveryProvider:
+    return DeliveryProvider(PlainTextRenderer(), StdoutPublisher(), diagnostics=diagnostics)
+
+
+def _build_telegram_publisher(
+    settings: Settings,
+    client: httpx.AsyncClient,
+    diagnostics: RenderedTextDiagnostics | None,
+) -> DeliveryProvider:
+    if not settings.telegram_bot_token or not settings.telegram_chat_id:
+        raise ValueError("Telegram publisher requires TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID")
+    return DeliveryProvider(
+        TelegramHTMLRenderer(),
+        TelegramPublisher(
+            client,
+            settings.telegram_bot_token,
+            settings.telegram_chat_id,
+            diagnostics,
+        ),
+        single_message_limit=TelegramPublisher.MAX_MESSAGE_LENGTH,
+        diagnostics=diagnostics,
+    )
+
+
+PUBLISHER_BUILDERS: dict[
+    str,
+    Callable[[Settings, httpx.AsyncClient, RenderedTextDiagnostics | None], DeliveryProvider],
+] = {
+    PublisherName.STDOUT: _build_stdout_publisher,
+    PublisherName.TELEGRAM: _build_telegram_publisher,
+}
 
 
 def _weather_context_provider(
@@ -336,7 +361,7 @@ def _weather_context_provider(
     names = weather_providers_for(location, settings.weather_providers)
     providers: list[WeatherContextProvider] = []
     for name in names:
-        if name == "qweather" and not _qweather_is_configured(settings):
+        if name == WeatherProviderName.QWEATHER and not _qweather_is_configured(settings):
             if settings.weather_providers is not None:
                 raise ValueError("Explicit QWeather provider is missing JWT configuration")
             continue
@@ -345,7 +370,7 @@ def _weather_context_provider(
         raise ValueError("No configured weather provider is available")
     _LOGGER.info(
         "Weather provider order providers=%s",
-        ",".join(name for name in names if name != "qweather" or _qweather_is_configured(settings)),
+        ",".join(name for name in names if name != WeatherProviderName.QWEATHER or _qweather_is_configured(settings)),
     )
     weather_provider: WeatherContextProvider = (
         providers[0] if len(providers) == 1 else FallbackWeatherContextProvider(*providers)
@@ -430,8 +455,8 @@ def _aqicn_provider(settings: Settings, client: httpx.AsyncClient) -> AirQuality
 
 
 WEATHER_PROVIDER_BUILDERS: dict[str, Callable[[Settings, httpx.AsyncClient], WeatherContextProvider]] = {
-    "qweather": _build_qweather,
-    "open-meteo": _build_open_meteo,
+    WeatherProviderName.QWEATHER: _build_qweather,
+    WeatherProviderName.OPEN_METEO: _build_open_meteo,
 }
 
 

weather_briefing/config.py

@@ -18,12 +18,17 @@
 
 from .models import ContextSourceConfig, FeedConfig, LocationSpec, ResolvedLocation
 from .reference_data import reference_string_tuple
+from .registries import PublisherName, WeatherProviderName
 
 
 class ConfigurationError(ValueError):
     """Raised when private runtime configuration is missing or malformed."""
 
 
+SUPPORTED_WEATHER_PROVIDERS = frozenset(WeatherProviderName)
+SUPPORTED_PUBLISHERS = frozenset(PublisherName)
+
+
 @overload
 def _clean_env(value: str) -> str: ...
 
@@ -148,6 +153,20 @@ def _optional_string_array(
     return tuple(entries)
 
 
+def _required_string_field(item: dict[str, Any], field: str, path: str) -> str:
+    value = item.get(field)
+    if not isinstance(value, str) or not value.strip():
+        raise ConfigurationError(f"{path}.{field} must be a non-empty string")
+    return value.strip()
+
+
+def _optional_string_field(item: dict[str, Any], field: str, path: str) -> str | None:
+    value = item.get(field)
+    if value is None:
+        return None
+    return _required_string_field(item, field, path)
+
+
 def _context_source(item: object, index: int) -> ContextSourceConfig:
     if not isinstance(item, dict):
         raise ConfigurationError(f"CONTEXT_SOURCES_JSON[{index}] must be a JSON object")
@@ -168,9 +187,19 @@ def _configured_weather_providers() -> tuple[str, ...] | None:
     providers = tuple(item.strip() for item in configured.split(",") if item.strip())
     if not providers:
         raise ConfigurationError("WEATHER_PROVIDERS cannot be empty")
+    unsupported = sorted(set(providers) - SUPPORTED_WEATHER_PROVIDERS)
+    if unsupported:
+        raise ConfigurationError(f"WEATHER_PROVIDERS contains unsupported providers: {', '.join(unsupported)}")
     return providers
 
 
+def _publisher() -> str:
+    publisher = _clean_env(os.getenv("PUBLISHER", "telegram"))
+    if publisher not in SUPPORTED_PUBLISHERS:
+        raise ConfigurationError(f"PUBLISHER must be one of: {', '.join(sorted(SUPPORTED_PUBLISHERS))}")
+    return publisher
+
+
 def state_path_from_env() -> Path:
     """Return the configured SQLite state path without loading all settings."""
     return Path(_clean_env(os.getenv("BRIEFING_STATE_PATH", "state/weather.sqlite3")))
@@ -190,12 +219,11 @@ def _locations(path: Path) -> tuple[LocationSpec, ...]:
         raise ConfigurationError(f"Configure at least one location in {path}")
     locations: list[LocationSpec] = []
     seen_ids: set[str] = set()
-    for item in items:
-        location_id = str(item.get("id", "")).strip()
-        name_value = item.get("name")
-        name = str(name_value).strip() if name_value is not None else None
-        name = name or None
-        if not location_id or not location_id.replace("-", "").replace("_", "").isalnum():
+    for index, item in enumerate(items):
+        item_path = f"{path}[{index}]"
+        location_id = _required_string_field(item, "id", item_path)
+        name = _optional_string_field(item, "name", item_path)
+        if not location_id.replace("-", "").replace("_", "").isalnum():
             raise ConfigurationError("Location id must use letters, numbers, '-' or '_'")
         if location_id in seen_ids:
             raise ConfigurationError(f"Duplicate location id: {location_id}")
@@ -221,16 +249,11 @@ def _locations(path: Path) -> tuple[LocationSpec, ...]:
 
 def _feeds(path: Path) -> tuple[FeedConfig, ...]:
     feeds: list[FeedConfig] = []
-    for item in _json_file(path):
-        source_id = str(item.get("id") or "").strip()
-        source_name = str(item.get("name") or "").strip()
-        source_url = str(item.get("url") or "").strip()
-        if not source_id:
-            raise ConfigurationError("RSS source must have an id")
-        if not source_name:
-            raise ConfigurationError(f"RSS source {source_id} must have a public display name")
-        if not source_url:
-            raise ConfigurationError(f"RSS source {source_id} must have a URL")
+    for index, item in enumerate(_json_file(path)):
+        item_path = f"{path}[{index}]"
+        source_id = _required_string_field(item, "id", item_path)
+        source_name = _required_string_field(item, "name", item_path)
+        source_url = _required_string_field(item, "url", item_path)
         feeds.append(
             FeedConfig(
                 id=source_id,
@@ -414,7 +437,7 @@ def from_env(cls) -> Settings:
             aqicn_api_token=_clean_env(os.getenv("AQICN_API_TOKEN")) or None,
             aqicn_base_url=_clean_env(os.getenv("AQICN_BASE_URL", "https://api.waqi.info")).rstrip("/"),
             state_path=state_path_from_env(),
-            publisher=_clean_env(os.getenv("PUBLISHER", "telegram")),
+            publisher=_publisher(),
             telegram_bot_token=_clean_env(os.getenv("TELEGRAM_BOT_TOKEN")) or None,
             telegram_chat_id=_clean_env(os.getenv("TELEGRAM_CHAT_ID")) or None,
             rss_max_attempts=_positive_integer("RSS_MAX_ATTEMPTS", 3),

weather_briefing/registries.py

@@ -0,0 +1,17 @@
+"""Names for application-owned provider and publisher adapters."""
+
+from enum import StrEnum
+
+
+class WeatherProviderName(StrEnum):
+    """Identify application-owned weather provider adapters."""
+
+    QWEATHER = "qweather"
+    OPEN_METEO = "open-meteo"
+
+
+class PublisherName(StrEnum):
+    """Identify application-owned delivery provider adapters."""
+
+    STDOUT = "stdout"
+    TELEGRAM = "telegram"