Fix auth for sample upsert stream

The client got authentication and signing metadata from
frequenz-client-base only for unary-unary and unary-stream RPCs.
UpsertMarketLocationSamplesStream is a streaming upsert call, so it
missed the key, timestamp, nonce, and signature metadata and the
service rejected it with a missing-signature authentication error.

Attach the same metadata explicitly for the upsert stream call and add
a regression test covering the streaming path.

Signed-off-by: Mathias L. Baumann <mathias.baumann@frequenz.com>

Author: Marenz

RELEASE_NOTES.md

@@ -53,3 +53,4 @@ Initial release of the Frequenz Market Metering API client for Python.
 ## Bug Fixes
 
 - `update_market_location()`: Add missing `expected_revision` parameter required for optimistic concurrency control.
+- `upsert_samples()`: Attach auth and signing metadata to the streaming upsert RPC so authenticated sample upserts work against services that require signed requests.

src/frequenz/client/marketmetering/_client.py

@@ -5,6 +5,10 @@
 
 from __future__ import annotations
 
+import hmac
+import secrets
+import time
+from base64 import urlsafe_b64encode
 from datetime import datetime, timedelta
 from typing import AsyncIterator, cast
 
@@ -152,6 +156,33 @@ def stub(self) -> marketmetering_pb2_grpc.MarketMeteringServiceStub:
             raise ClientNotConnected(server_url=self.server_url, operation="stub")
         return self._stub
 
+    def _metadata(self, method: str) -> tuple[tuple[str, str | bytes], ...] | None:
+        """Build request metadata for RPCs not covered by client-base interceptors."""
+        if self._auth_key is None:
+            return None
+
+        metadata: list[tuple[str, str | bytes]] = [("key", self._auth_key)]
+        if self._sign_secret is None:
+            return tuple(metadata)
+
+        ts = str(int(time.time())).encode()
+        nonce = urlsafe_b64encode(secrets.token_bytes(16))
+
+        digest = hmac.new(self._sign_secret.encode(), digestmod="sha256")
+        digest.update(self._auth_key.encode())
+        digest.update(ts)
+        digest.update(nonce)
+        digest.update(method.encode())
+
+        metadata.extend(
+            [
+                ("ts", ts),
+                ("nonce", nonce),
+                ("sig", urlsafe_b64encode(digest.digest()).rstrip(b"=")),
+            ]
+        )
+        return tuple(metadata)
+
     async def create_market_location(
         self,
         *,
@@ -336,6 +367,7 @@ async def request_generator() -> (
             AsyncIterator[pb.UpsertMarketLocationSamplesStreamResponse],
             self.stub.UpsertMarketLocationSamplesStream(
                 request_generator(),  # type: ignore[arg-type]
+                metadata=self._metadata("UpsertMarketLocationSamplesStream"),
                 timeout=self._stream_timeout_seconds,
             ),
         )

tests/test_client.py

@@ -551,9 +551,9 @@ async def test_yields_parsed_series(self) -> None:
         response = pb.ReceiveMarketLocationSamplesStreamResponse(series=[series_pb])
 
         # Mock the streaming call to return an async iterator.
-        async def mock_stream() -> AsyncIterator[
-            pb.ReceiveMarketLocationSamplesStreamResponse
-        ]:
+        async def mock_stream() -> (
+            AsyncIterator[pb.ReceiveMarketLocationSamplesStreamResponse]
+        ):
             yield response
 
         client.stub.ReceiveMarketLocationSamplesStream = MagicMock(
@@ -579,9 +579,9 @@ async def test_sends_time_filter(self) -> None:
         """Test that start_time and end_time are sent."""
         client = _make_client()
 
-        async def mock_stream() -> AsyncIterator[
-            pb.ReceiveMarketLocationSamplesStreamResponse
-        ]:
+        async def mock_stream() -> (
+            AsyncIterator[pb.ReceiveMarketLocationSamplesStreamResponse]
+        ):
             return
             yield  # make it an async generator  # noqa: RET504
 
@@ -638,9 +638,9 @@ async def test_yields_parsed_results(self) -> None:
             sample=sample_pb,
         )
 
-        async def mock_stream() -> AsyncIterator[
-            pb.UpsertMarketLocationSamplesStreamResponse
-        ]:
+        async def mock_stream() -> (
+            AsyncIterator[pb.UpsertMarketLocationSamplesStreamResponse]
+        ):
             yield upsert_response
 
         client.stub.UpsertMarketLocationSamplesStream = MagicMock(
@@ -666,9 +666,9 @@ async def mock_stream() -> AsyncIterator[
             samples=[sample],
         )
 
-        async def input_stream() -> AsyncIterator[
-            tuple[MarketLocationRef, MarketLocationSeries]
-        ]:
+        async def input_stream() -> (
+            AsyncIterator[tuple[MarketLocationRef, MarketLocationSeries]]
+        ):
             yield (ml_ref, series)
 
         results = []

tests/test_integration.py

@@ -27,6 +27,8 @@
         uv run pytest -m integration
 """
 
+import os
+import socket
 from collections.abc import AsyncIterator
 
 import grpc
@@ -52,9 +54,22 @@
 pytestmark = pytest.mark.integration
 
 
+def _service_available() -> bool:
+    """Check whether the local integration test service is reachable."""
+    with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as sock:
+        sock.settimeout(0.5)
+        return sock.connect_ex(("::1", 50051, 0, 0)) == 0
+
+
 @pytest.fixture
 async def client() -> AsyncIterator[MarketMeteringApiClient]:
     """Create a connected client for testing."""
+    if os.environ.get("CI") == "true":
+        pytest.skip("integration tests are not run in CI")
+
+    if not _service_available():
+        pytest.skip("integration test service is not running on [::1]:50051")
+
     c = MarketMeteringApiClient(
         server_url=SERVICE_URL,
         auth_key=AUTH_KEY,

tests/test_marketmetering.py

@@ -3,8 +3,13 @@
 
 """Tests for the Market Metering client."""
 
-from datetime import timedelta
+from collections.abc import AsyncIterator
+from datetime import datetime, timedelta, timezone
+from typing import Any, cast
 
+import pytest
+
+from frequenz.client.marketmetering import MarketMeteringApiClient
 from frequenz.client.marketmetering.types import (
     DataQuality,
     DownsamplingMethod,
@@ -13,6 +18,8 @@
     MarketLocationId,
     MarketLocationIdType,
     MarketLocationRef,
+    MarketLocationSample,
+    MarketLocationSeries,
     MetricType,
     MetricUnit,
     ResamplingMethod,
@@ -117,3 +124,96 @@ def test_resampling_options_with_resolution(self) -> None:
         """Test ResamplingOptions with resolution."""
         options = ResamplingOptions(resolution=TimeResolution.MIN_15)
         assert options.resolution == TimeResolution.MIN_15
+
+
+class _UpsertStub:
+    """A fake stub capturing upsert stream call arguments."""
+
+    def __init__(self) -> None:
+        self.metadata: tuple[tuple[str, str | bytes], ...] | None = None
+        self.requests: list[object] = []
+
+    # gRPC-generated method names use UpperCamelCase and this stub mirrors that.
+    # pylint: disable=invalid-name,missing-function-docstring
+    async def UpsertMarketLocationSamplesStream(  # noqa: N802
+        self,
+        request_iterator: AsyncIterator[object],
+        *,
+        metadata: tuple[tuple[str, str | bytes], ...] | None = None,
+        timeout: float | None = None,
+    ) -> AsyncIterator[object]:
+        del timeout
+        self.metadata = metadata
+        async for request in request_iterator:
+            self.requests.append(request)
+        items: tuple[object, ...] = ()
+        for item in items:
+            yield item
+
+
+class TestClientMethods:
+    """Tests for client RPC helpers."""
+
+    @pytest.mark.asyncio
+    async def test_upsert_samples_adds_stream_metadata(self) -> None:
+        """Test that bidi upsert includes auth and signing metadata."""
+        client = MarketMeteringApiClient(
+            server_url="grpc://example.com",
+            auth_key="test-key",
+            sign_secret="test-secret",
+            connect=False,
+        )
+        stub = _UpsertStub()
+        setattr(client, "_stub", cast(Any, stub))
+        setattr(client, "_channel", cast(Any, object()))
+
+        market_location_ref = MarketLocationRef(
+            enterprise_id=42,
+            market_area=MarketArea.EU_DE,
+            market_location_id=MarketLocationId(
+                value="DE01234567890",
+                type=MarketLocationIdType.MALO_ID,
+            ),
+        )
+        series = MarketLocationSeries(
+            market_location_ref=market_location_ref,
+            direction=EnergyFlowDirection.IMPORT,
+            metric_type=MetricType.ACTIVE_ENERGY,
+            metric_unit=MetricUnit.KWH,
+            resolution=TimeResolution.MIN_15,
+            samples=[],
+        )
+        sample = MarketLocationSample(
+            sample_time=datetime.now(timezone.utc),
+            value=1.0,
+            quality=DataQuality.MEASURED,
+            revision=1,
+            update_time=None,
+            resampling_method=ResamplingMethod.UNSPECIFIED,
+        )
+
+        async def sample_generator() -> (
+            AsyncIterator[tuple[MarketLocationRef, MarketLocationSeries]]
+        ):
+            yield (
+                market_location_ref,
+                MarketLocationSeries(
+                    market_location_ref=series.market_location_ref,
+                    direction=series.direction,
+                    metric_type=series.metric_type,
+                    metric_unit=series.metric_unit,
+                    resolution=series.resolution,
+                    samples=[sample],
+                ),
+            )
+
+        assert [
+            result async for result in client.upsert_samples(sample_generator())
+        ] == []
+        assert stub.metadata is not None
+        metadata = dict(stub.metadata)
+        assert metadata["key"] == "test-key"
+        assert metadata["ts"]
+        assert metadata["nonce"]
+        assert metadata["sig"]
+        assert len(stub.requests) == 1