test(gateway): exercise streaming configuration with real YAML

tests/test_gateway_streaming_nested_config.py

@@ -1,42 +1,54 @@
 """Regression test for #25676 — nested gateway.streaming config must be loaded."""
 from pathlib import Path
-from unittest.mock import patch, MagicMock
+from textwrap import dedent
+
+import pytest
 
 
-
-def _load_with_yaml_dict(yaml_dict: dict):
-    """Patch filesystem so load_gateway_config() sees *yaml_dict* as config.yaml."""
+@pytest.fixture
+def load_yaml(tmp_path, monkeypatch):
+    """Exercise the raw loader, not a particular YAML parser implementation."""
     from gateway.config import load_gateway_config
 
-    fake_home = Path("/tmp/fake_hermes_home_25676")
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
 
-    def fake_exists(self):
-        return str(self).endswith("config.yaml")
+    def load(text):
+        (tmp_path / "config.yaml").write_text(dedent(text), encoding="utf-8")
+        return load_gateway_config()
 
-    with patch("gateway.config.get_hermes_home", return_value=fake_home), \
-         patch.object(Path, "exists", fake_exists), \
-         patch("builtins.open", create=True) as mock_file:
-        mock_file.return_value.__enter__ = lambda s: s
-        mock_file.return_value.__exit__ = MagicMock(return_value=False)
-        with patch("yaml.safe_load", return_value=yaml_dict):
-            return load_gateway_config()
+    return load
 
 
 class TestStreamingConfigNested:
-    def test_top_level_streaming(self):
-        cfg = _load_with_yaml_dict({"streaming": {"enabled": True, "transport": "draft"}})
+    def test_top_level_streaming(self, load_yaml):
+        cfg = load_yaml("streaming: {enabled: true, transport: draft}")
         assert cfg.streaming.enabled is True
         assert cfg.streaming.transport == "draft"
 
 
-    def test_top_level_takes_precedence(self):
-        cfg = _load_with_yaml_dict({
-            "streaming": {"enabled": True, "transport": "edit"},
-            "gateway": {"streaming": {"enabled": False, "transport": "draft"}},
-        })
+    def test_top_level_takes_precedence(self, load_yaml):
+        cfg = load_yaml("""
+            streaming: {enabled: true, transport: edit}
+            gateway:
+              streaming: {enabled: false, transport: draft}
+        """)
         assert cfg.streaming.enabled is True
         assert cfg.streaming.transport == "edit"
 
+    def test_nested_only_streaming(self, load_yaml):
+        cfg = load_yaml("""
+            gateway:
+              streaming: {enabled: true, transport: draft}
+        """)
+        assert cfg.streaming.enabled is True
+        assert cfg.streaming.transport == "draft"
+
+    def test_explicit_disable_preserves_transport(self, load_yaml):
+        cfg = load_yaml("streaming: {enabled: false, mode: auto, transport: draft}")
+        assert cfg.streaming.enabled is False
+        assert cfg.streaming.transport == "draft"
+
 
 class TestStreamingModeAlias:
     """``streaming: {mode: ...}`` is an alias that also implies ``enabled``.
@@ -105,10 +117,10 @@ class TestStreamingYamlBooleanQuirk:
         assert sc.transport == "auto"
 
 
-    def test_loader_normalizes_bare_yaml_off(self):
+    def test_loader_normalizes_bare_yaml_off(self, load_yaml):
         """End-to-end through load_gateway_config(): unquoted ``mode: off``
         (a YAML boolean) must keep streaming disabled."""
-        cfg = _load_with_yaml_dict({"streaming": {"mode": False}})
+        cfg = load_yaml("streaming: {mode: off}")
         assert cfg.streaming.enabled is False
         assert cfg.streaming.transport == "off"