fix(async-watcher): forward auth metadata to GetConfig and Subscribe

AsyncConfigClient cannot use channel interceptors, so it injects auth
metadata per-call via self._metadata(). AsyncConfigWatcher was not
receiving this metadata, causing both the initial GetConfig snapshot
and the Subscribe streaming call to omit auth headers. Auth-enabled
servers rejected the Subscribe with UNAUTHENTICATED.

Added a metadata parameter to AsyncConfigWatcher.__init__ and thread
it through both _load_snapshot and _subscribe_loop. Updated
AsyncConfigClient.watch() to pass self._metadata(). Added a unit test
asserting that both gRPC calls receive the correct metadata tuples.

Co-Authored-By: Claude <noreply@anthropic.com>

Closes #48

Author: zeevdr

sdk/src/opendecree/async_client.py

@@ -373,4 +373,4 @@ def watch(self, tenant_id: str) -> AsyncConfigWatcher:
         """
         from opendecree.async_watcher import AsyncConfigWatcher
 
-        return AsyncConfigWatcher(self._stub, self._pb2, tenant_id, self._timeout)
+        return AsyncConfigWatcher(self._stub, self._pb2, tenant_id, self._timeout, self._metadata())

sdk/src/opendecree/async_watcher.py

@@ -129,11 +129,19 @@ class AsyncConfigWatcher:
     auto-starts on enter, auto-stops on exit.
     """
 
-    def __init__(self, stub: Any, pb2: Any, tenant_id: str, timeout: float) -> None:
+    def __init__(
+        self,
+        stub: Any,
+        pb2: Any,
+        tenant_id: str,
+        timeout: float,
+        metadata: list[tuple[str, str]] | None = None,
+    ) -> None:
         self._stub = stub
         self._pb2 = pb2
         self._tenant_id = tenant_id
         self._timeout = timeout
+        self._metadata: list[tuple[str, str]] = metadata or []
         self._fields: dict[str, AsyncWatchedField] = {}  # type: ignore[type-arg]
         self._task: asyncio.Task | None = None  # type: ignore[type-arg]
         self._stopped = False
@@ -193,6 +201,7 @@ async def _load_snapshot(self) -> None:
         resp = await self._stub.GetConfig(
             self._pb2.GetConfigRequest(tenant_id=self._tenant_id),
             timeout=self._timeout,
+            metadata=self._metadata,
         )
         all_values = process_get_all_response(resp)
         for path, watched in self._fields.items():
@@ -211,6 +220,7 @@ async def _subscribe_loop(self) -> None:
                         tenant_id=self._tenant_id,
                         field_paths=field_paths,
                     ),
+                    metadata=self._metadata,
                 )
                 backoff = _RECONNECT_INITIAL
 

sdk/tests/test_async_watcher.py

@@ -270,3 +270,35 @@ async def test_non_retryable_error_stops_loop(self):
         assert w._task is not None
         assert w._task.done()
         await w.stop()
+
+    @pytest.mark.asyncio
+    async def test_auth_metadata_forwarded(self):
+        """metadata= is passed to both GetConfig and Subscribe."""
+        stub = MagicMock()
+        pb2 = MagicMock()
+
+        mock_resp = MagicMock()
+        mock_resp.config.values = []
+        stub.GetConfig = AsyncMock(return_value=mock_resp)
+
+        auth_meta = [("x-subject", "svc"), ("x-role", "superadmin")]
+        w = AsyncConfigWatcher(stub, pb2, "t1", timeout=5.0, metadata=auth_meta)
+        w.field("fee", float, default=0.0)
+
+        async def empty_stream():
+            return
+            yield
+
+        stub.Subscribe.return_value = empty_stream()
+
+        await w.start()
+        await asyncio.sleep(0.05)
+        await w.stop()
+
+        stub.GetConfig.assert_awaited_once()
+        _, get_kwargs = stub.GetConfig.call_args
+        assert get_kwargs.get("metadata") == auth_meta
+
+        stub.Subscribe.assert_called_once()
+        _, sub_kwargs = stub.Subscribe.call_args
+        assert sub_kwargs.get("metadata") == auth_meta