Support key signing and handle dry_run cleaner (#184)

- **Support key signing**
- **Consider dry_run status in dispatch merging**

Author: Marenz

RELEASE_NOTES.md

@@ -6,14 +6,15 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+* The `key` parameter in the `Dispatcher` constructor is now deprecated. Use `auth_key` instead. The `sign_secret` parameter is an additional optional parameter for signing.
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* `dry_run` status is now considered when merging dispatches. Dispatches with different `dry_run` values will no longer be merged, ensuring that dry-run and operational dispatches are handled by separate actors.
+* Two new parameters were added to the `Dispatcher` constructor:
+  * `sign_secret`: A secret key used for signing messages.
+  * `auth_key`: An authentication key for the Dispatch API.
 
 ## Bug Fixes
 
-* The merge by type class now uses the correct logger path.
-* The merge by type was made more robust under heavy load, making sure to use the same `now` for all dispatches that are checked.
-* Fix that the merge filter was using an outdated dispatches dict once fetch() ran.
+<!-- Here goes notable bug fixes that are worth a special mention or explanation -->

pyproject.toml

@@ -40,8 +40,7 @@ dependencies = [
   # plugins.mkdocstrings.handlers.python.import)
   "frequenz-sdk >= 1.0.0-rc2100, < 1.0.0-rc2200",
   "frequenz-channels >= 1.6.1, < 2.0.0",
-  "frequenz-client-dispatch >= 0.11.1, < 0.12.0",
-  "frequenz-client-common >= 0.3.2, < 0.4.0",
+  "frequenz-client-dispatch >= 0.11.2, < 0.12.0",
   "frequenz-client-base >= 0.11.0, < 0.12.0",
 ]
 dynamic = ["version"]
@@ -54,7 +53,7 @@ email = "[email protected]"
 dev-flake8 = [
   "flake8 == 7.3.0",
   "flake8-docstrings == 1.7.0",
-  "flake8-pyproject == 1.2.3", # For reading the flake8 config from pyproject.toml
+  "flake8-pyproject == 1.2.3",  # For reading the flake8 config from pyproject.toml
   "pydoclint == 0.6.6",
   "pydocstyle == 6.3.0",
 ]

src/frequenz/dispatch/_dispatcher.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import logging
+import warnings
 from asyncio import Event
 from datetime import timedelta
 from typing import Awaitable, Callable, Self
@@ -64,7 +65,7 @@ async def run():
             async with Dispatcher(
                 microgrid_id=microgrid_id,
                 server_url=url,
-                key=key
+                auth_key=key
             ) as dispatcher:
                 dispatcher.start_managing(
                     dispatch_type="DISPATCH_TYPE",
@@ -90,7 +91,7 @@ async def run():
             async with Dispatcher(
                 microgrid_id=microgrid_id,
                 server_url=url,
-                key=key
+                auth_key=key
             ) as dispatcher:
                 actor = MagicMock() # replace with your actor
 
@@ -135,7 +136,7 @@ async def run():
             async with Dispatcher(
                 microgrid_id=microgrid_id,
                 server_url=url,
-                key=key,
+                auth_key=key,
             ) as dispatcher:
                 events_receiver = dispatcher.new_lifecycle_events_receiver("DISPATCH_TYPE")
 
@@ -172,7 +173,7 @@ async def run():
             async with Dispatcher(
                 microgrid_id=microgrid_id,
                 server_url=url,
-                key=key,
+                auth_key=key,
             ) as dispatcher:
                 # Create a new dispatch
                 new_dispatch = await dispatcher.client.create(
@@ -205,7 +206,9 @@ def __init__(
         *,
         microgrid_id: MicrogridId,
         server_url: str,
-        key: str,
+        key: str | None = None,
+        auth_key: str | None = None,
+        sign_secret: str | None = None,
         call_timeout: timedelta = timedelta(seconds=60),
         stream_timeout: timedelta = timedelta(minutes=5),
     ):
@@ -214,15 +217,39 @@ def __init__(
         Args:
             microgrid_id: The microgrid id.
             server_url: The URL of the dispatch service.
-            key: The key to access the service.
+            key: The key to access the service, deprecated, use `auth_key` instead.
+            auth_key: The authentication key to access the service.
+            sign_secret: The secret to sign the requests, optional
             call_timeout: The timeout for API calls.
             stream_timeout: The timeout for streaming API calls.
+
+        Raises:
+            ValueError: If both or neither `key` and `auth_key` are provided
         """
         super().__init__()
 
+        if key is not None and auth_key is not None:
+            raise ValueError(
+                "Both 'key' and 'auth_key' are provided, please use only 'auth_key'."
+            )
+
+        if key is None and auth_key is None:
+            raise ValueError(
+                "'auth_key' must be provided to access the dispatch service."
+            )
+
+        if key is not None:
+            auth_key = key
+            warnings.warn(
+                "'key' is deprecated, use 'auth_key' instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
         self._client = DispatchApiClient(
             server_url=server_url,
-            key=key,
+            auth_key=auth_key,
+            sign_secret=sign_secret,
             call_timeout=call_timeout,
             stream_timeout=stream_timeout,
         )
@@ -306,11 +333,11 @@ async def start_managing(
         This also decides how instances are mapped from dispatches to actors:
 
         * [`MergeByType`][frequenz.dispatch.MergeByType] — All dispatches map to
-        one single instance identified by the dispatch type.
+        one single instance identified by the dispatch type and dry_run status.
         * [`MergeByTypeTarget`][frequenz.dispatch.MergeByTypeTarget] — A
-        dispatch maps to an instance identified by the dispatch type and target.
-        So different dispatches with equal type and target will map to the same
-        instance.
+        dispatch maps to an instance identified by the dispatch type, dry_run status
+        and target. So different dispatches with equal type and target will map to
+        the same instance.
         * `None` — No merging, each dispatch maps to a separate instance.
 
         Args:

src/frequenz/dispatch/_merge_strategies.py

@@ -30,7 +30,7 @@ class MergeByType(MergeStrategy):
     @override
     def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return DispatchActorId(_hash_positive(dispatch.type))
+        return DispatchActorId(_hash_positive((dispatch.type, dispatch.dry_run)))
 
     @override
     def filter(
@@ -88,4 +88,6 @@ class MergeByTypeTarget(MergeByType):
     @override
     def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return DispatchActorId(_hash_positive((dispatch.type, tuple(dispatch.target))))
+        return DispatchActorId(
+            _hash_positive((dispatch.type, dispatch.dry_run, tuple(dispatch.target)))
+        )

tests/test_frequenz_dispatch.py

@@ -781,3 +781,78 @@ async def test_at_least_one_running_filter(
     await asyncio.sleep(1)
     stopped_b = await receiver.receive()
     assert not stopped_b.started
+
+
+@pytest.mark.parametrize(
+    "merge_strategy",
+    [
+        MergeByType(),
+        MergeByTypeTarget(),
+    ],
+)
+async def test_dry_run_dispatches_not_merged(
+    fake_time: time_machine.Coordinates,
+    generator: DispatchGenerator,
+    merge_strategy: MergeStrategy,
+) -> None:
+    """Test that dispatches with different dry_run values are not merged."""
+    microgrid_id = MicrogridId(randint(1, 100))
+    client = FakeClient()
+    service = DispatchScheduler(
+        microgrid_id=microgrid_id,
+        client=client,
+    )
+    service.start()
+
+    receiver = await service.new_running_state_event_receiver(
+        "TEST_TYPE", merge_strategy=merge_strategy
+    )
+
+    # Create two dispatches with same type and target, but different dry_run
+    dispatch1 = replace(
+        generator.generate_dispatch(),
+        active=True,
+        duration=timedelta(seconds=10),
+        target=TargetIds(1, 2),
+        start_time=_now() + timedelta(seconds=5),
+        recurrence=RecurrenceRule(),
+        type="TEST_TYPE",
+        dry_run=False,
+    )
+    dispatch2 = replace(
+        dispatch1,
+        dry_run=True,
+    )
+
+    lifecycle_events = service.new_lifecycle_events_receiver("TEST_TYPE")
+
+    await client.create(**to_create_params(microgrid_id, dispatch1))
+    await client.create(**to_create_params(microgrid_id, dispatch2))
+
+    # Wait for both to be registered
+    await lifecycle_events.receive()
+    await lifecycle_events.receive()
+
+    # Move time forward to start both dispatches
+    fake_time.shift(timedelta(seconds=6))
+    await asyncio.sleep(1)
+
+    started1 = await receiver.receive()
+    started2 = await receiver.receive()
+
+    assert started1.started
+    assert started2.started
+    assert started1.dry_run != started2.dry_run
+
+    # Move time forward to the end of the dispatches
+    fake_time.shift(timedelta(seconds=10))
+    await asyncio.sleep(1)
+
+    stopped1 = await receiver.receive()
+    stopped2 = await receiver.receive()
+
+    assert not stopped1.started
+    assert not stopped2.started
+    assert stopped1.dry_run != stopped2.dry_run
+
+    await service.stop()

tests/test_managing_actor.py

@@ -335,20 +335,25 @@ async def test_manage_abstraction(
     class MyFakeClient(FakeClient):
         """Fake client for testing."""
 
+        # pylint: disable=too-many-arguments,unused-argument
         def __init__(
             self,
             *,
             server_url: str,
-            key: str,
+            auth_key: str | None = None,
+            key: str | None = None,
+            sign_secret: str | None = None,
             call_timeout: timedelta,
             stream_timeout: timedelta,
         ) -> None:
             assert server_url
-            assert key
+            assert key or auth_key
             assert call_timeout
             assert stream_timeout
             super().__init__()
 
+        # pylint: enable=too-many-arguments,unused-argument
+
     mid = MicrogridId(1)
 
     # Patch `Client` class in Dispatcher with MyFakeClient