Config hygiene: validate parameter, flag conflicting flags, dedup source list

backend/config.py

@@ -439,6 +439,54 @@ def validate(self):
             self.warn(f"Invalid end date {self.end_date}")
             sys.exit(2)
 
+        if not self._validate_parameter():
+            self.warn(
+                f"Unknown parameter {self.parameter!r}. "
+                f"Valid parameters: {sorted(PARAMETER_SOURCE_MAP)}"
+            )
+            sys.exit(2)
+
+        # Advisory only: these states are accepted (the code picks one) but are
+        # almost always a mistake, so surface them instead of failing silently.
+        self._warn_spatial_exclusivity()
+        self._warn_output_mode_exclusivity()
+
+    def _validate_parameter(self):
+        # An empty parameter is valid: sites-only flows don't need one. A set
+        # parameter must be one the source map knows, otherwise no source can
+        # ever be resolved for it.
+        if self.parameter:
+            return self.parameter in PARAMETER_SOURCE_MAP
+        return True
+
+    def _warn_spatial_exclusivity(self):
+        # bbox/county/wkt are resolved with inconsistent precedence across
+        # bbox_bounding_points (bbox first) and bounding_wkt (wkt first), so
+        # setting more than one silently does different things in different code
+        # paths. Exactly one (or none, meaning statewide) is intended.
+        set_filters = [n for n in ("bbox", "county", "wkt") if getattr(self, n)]
+        if len(set_filters) > 1:
+            self.warn(
+                f"Multiple spatial filters set ({', '.join(set_filters)}); set "
+                "exactly one — resolution precedence differs between code paths."
+            )
+
+    def _warn_output_mode_exclusivity(self):
+        modes = [
+            n
+            for n in (
+                "output_summary",
+                "output_timeseries_unified",
+                "output_timeseries_separated",
+            )
+            if getattr(self, n)
+        ]
+        if len(modes) > 1:
+            self.warn(
+                f"Multiple output modes set ({', '.join(modes)}); only the first "
+                "is used at dump time. Set exactly one."
+            )
+
     def _extract_date(self, d):
         if d:
             for fmt in (

orchestration/resources/die_config.py

@@ -1,7 +1,7 @@
 import os
 from typing import Optional
 import dagster as dg
-from backend.config import Config
+from backend.config import Config, SOURCE_KEYS
 
 
 class DIEConfigResource(dg.ConfigurableResource):
@@ -62,14 +62,10 @@ def get_config(self, product: dict, parameter: Optional[str] = None) -> Config:
             payload["wkt"] = None
 
         if sources_spec.get("include"):
-            # NOTE: must stay in sync with backend.config.SOURCE_KEYS — an
-            # include-list product silently drops any source missing here.
-            all_sources = [
-                "bernco", "bor", "cabq", "ebid", "nmbgmr_amp",
-                "nmed_dwb", "nmose_isc_seven_rivers", "nmose_pod",
-                "nmose_roswell", "nwis", "pvacd", "wqp",
-            ]
-            for s in all_sources:
+            # Enable only the included sources. Derived from the backend's
+            # canonical source list so a new source can't be silently dropped
+            # from an include-list product.
+            for s in SOURCE_KEYS:
                 payload[f"use_source_{s}"] = s in sources_spec["include"]
         elif sources_spec.get("exclude"):
             for s in sources_spec["exclude"]:

tests/test_config_validation.py

@@ -0,0 +1,60 @@
+"""Tests for Config.validate() parameter check and advisory exclusivity guards."""
+import pytest
+
+from backend.config import Config, PARAMETER_SOURCE_MAP, SOURCE_KEYS
+
+
+def _cfg(**attrs):
+    c = Config()
+    for k, v in attrs.items():
+        setattr(c, k, v)
+    return c
+
+
+class TestParameterValidation:
+    def test_valid_parameter_passes(self):
+        c = _cfg(parameter="waterlevels")
+        c.validate()  # no exit
+
+    def test_empty_parameter_ok(self):
+        # sites-only flows carry no parameter
+        c = _cfg(parameter="")
+        c.validate()
+
+    def test_unknown_parameter_exits(self):
+        c = _cfg(parameter="not_a_real_parameter")
+        with pytest.raises(SystemExit):
+            c.validate()
+
+    def test_validate_parameter_helper(self):
+        assert _cfg(parameter="arsenic")._validate_parameter() is True
+        assert _cfg(parameter="")._validate_parameter() is True
+        assert _cfg(parameter="bogus")._validate_parameter() is False
+
+
+class TestExclusivityGuards:
+    def test_single_spatial_filter_no_warn(self, capsys):
+        # county only — should not trip the multi-filter advisory
+        c = _cfg(parameter="waterlevels", county="Bernalillo")
+        c._warn_spatial_exclusivity()  # must not raise
+
+    def test_multiple_spatial_filters_advisory(self):
+        c = _cfg(county="Bernalillo", wkt="POLYGON((0 0,0 1,1 1,1 0,0 0))")
+        # advisory only — does not raise/exit
+        c._warn_spatial_exclusivity()
+
+    def test_multiple_output_modes_advisory(self):
+        c = _cfg(output_summary=True, output_timeseries_unified=True)
+        c._warn_output_mode_exclusivity()  # advisory, no raise
+
+    def test_validate_passes_with_one_mode(self):
+        c = _cfg(parameter="waterlevels", output_summary=True)
+        c.validate()
+
+
+class TestSourceKeysCanonical:
+    def test_source_keys_cover_parameter_map(self):
+        # every agency referenced by the parameter map is a real source key
+        for entry in PARAMETER_SOURCE_MAP.values():
+            for agency in entry["agencies"]:
+                assert agency in SOURCE_KEYS