Fix auth for sample upsert stream (#48)

The client currently gets auth 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.

This attaches the same metadata explicitly for the upsert stream call
and adds a regression test covering the streaming path.

Verified with:
- uv run pytest tests/test_marketmetering.py -q
- end-to-end client flow against
market-metering.eu-1.production.api.frequenz.com covering list, create,
update, deactivate, activate, upsert_samples, and stream_samples

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.

pyproject.toml

@@ -183,6 +183,9 @@ testpaths = ["tests", "src"]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 required_plugins = ["pytest-asyncio", "pytest-mock"]
+markers = [
+  "integration: integration tests requiring a running gRPC server",
+]
 
 [tool.mypy]
 explicit_package_bases = true

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,
             ),
         )