Update frequenz dependencies (#23)

Author: Marenz

mkdocs.yml

@@ -114,8 +114,8 @@ plugins:
           import:
             # See https://mkdocstrings.github.io/python/usage/#import for details
             - https://docs.python.org/3/objects.inv
-            - https://frequenz-floss.github.io/frequenz-channels-python/v0.16/objects.inv
-            - https://frequenz-floss.github.io/frequenz-client-dispatch-python/v0.1/objects.inv
+            - https://frequenz-floss.github.io/frequenz-channels-python/v1.0/objects.inv
+            - https://frequenz-floss.github.io/frequenz-client-dispatch-python/v0.2/objects.inv
             - https://frequenz-floss.github.io/frequenz-sdk-python/v1.0-pre/objects.inv
             - https://grpc.github.io/grpc/python/objects.inv
             - https://typing-extensions.readthedocs.io/en/stable/objects.inv

pyproject.toml

@@ -31,14 +31,12 @@ dependencies = [
   # Make sure to update the version for cross-referencing also in the
   # mkdocs.yml file when changing the version here (look for the config key
   # plugins.mkdocstrings.handlers.python.import)
-  "frequenz-sdk == v1.0.0-rc5",
-  "frequenz-channels == 1.0.0b2",
-  "frequenz-dispatch-client @ git+https://github.com/frequenz-floss/frequenz-client-dispatch-python.git@00dcb3c",
-  "frequenz-api-dispatch @ git+https://github.com/frequenz-floss/frequenz-api-dispatch.git@223315c",
-#  "frequenz-api-dispatch >= 0.13.0, < 0.14",
-  "frequenz-client-base >= 0.2.1, < 0.4.0",
-  # Directly use unreleased commit until the first release
-  "frequenz-client-common @ git+https://github.com/frequenz-floss/frequenz-client-common-python.git@5e6e7b7",
+  "frequenz-sdk == v1.0.0-rc6",
+  "frequenz-channels == 1.0.0",
+  "frequenz-api-dispatch >= 0.13.0, < 0.14",
+  "frequenz-client-dispatch == 0.2.0",
+  "frequenz-client-base >= 0.3.0, < 0.4.0",
+  "frequenz-client-common >= 0.1.0, < 0.2.0",
 ]
 dynamic = ["version"]
 
@@ -90,6 +88,8 @@ dev-pytest = [
   "pytest-asyncio == 0.23.6",
   "async-solipsism == 0.6",
   "time-machine == 2.14.1",
+  # Currently pinned due to v6.1.0 breaking our code
+  "sybil == 6.0.3",
 ]
 dev = [
   "frequenz-dispatch[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest]",

src/frequenz/dispatch/_dispatcher.py

@@ -13,22 +13,22 @@
 from frequenz.dispatch._event import DispatchEvent
 from frequenz.dispatch.actor import DispatchingActor
 
-ReceivedT = TypeVar("ReceivedT")
+ReceivedT_co = TypeVar("ReceivedT_co", covariant=True)
 """The type being received."""
 
 
-class ReceiverFetcher(Protocol[ReceivedT]):
+class ReceiverFetcher(Protocol[ReceivedT_co]):
     """An interface that just exposes a `new_receiver` method."""
 
     @abc.abstractmethod
     def new_receiver(
-        self, name: str | None = None, maxsize: int = 50
-    ) -> Receiver[ReceivedT]:
+        self, *, name: str | None = None, limit: int = 50
+    ) -> Receiver[ReceivedT_co]:
         """Get a receiver from the channel.
 
         Args:
             name: A name to identify the receiver in the logs.
-            maxsize: The maximum size of the receiver.
+            limit: The maximum size of the receiver.
 
         Returns:
             A receiver instance.
@@ -107,8 +107,8 @@ def __init__(
             grpc_channel: The gRPC channel.
             svc_addr: The service address.
         """
-        self._ready_channel = Broadcast[Dispatch]("ready_dispatches")
-        self._updated_channel = Broadcast[DispatchEvent]("new_dispatches")
+        self._ready_channel = Broadcast[Dispatch](name="ready_dispatches")
+        self._updated_channel = Broadcast[DispatchEvent](name="new_dispatches")
         self._actor = DispatchingActor(
             microgrid_id,
             grpc_channel,

src/frequenz/dispatch/actor.py

@@ -11,7 +11,7 @@
 import grpc.aio
 from dateutil import rrule
 from frequenz.channels import Sender
-from frequenz.channels.util import Timer
+from frequenz.channels.timer import SkipMissedAndDrift, Timer
 from frequenz.client.dispatch import Client
 from frequenz.client.dispatch.types import Dispatch, Frequency, Weekday
 from frequenz.sdk.actor import Actor
@@ -94,7 +94,7 @@ def __init__(
         self._microgrid_id = microgrid_id
         self._updated_dispatch_sender = updated_dispatch_sender
         self._ready_dispatch_sender = ready_dispatch_sender
-        self._poll_timer = Timer.timeout(poll_interval)
+        self._poll_timer = Timer(poll_interval, SkipMissedAndDrift())
 
     async def _run(self) -> None:
         """Run the actor."""

tests/test_frequenz_dispatch.py

@@ -7,13 +7,12 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from random import randint
-from typing import AsyncIterator, Iterator, TypeVar
+from typing import AsyncIterator, Iterator
 from unittest.mock import MagicMock
 
 import async_solipsism
 import time_machine
 from frequenz.channels import Broadcast, Receiver
-from frequenz.channels._broadcast import Sender
 from frequenz.client.dispatch.test.client import FakeClient, to_create_params
 from frequenz.client.dispatch.test.generator import DispatchGenerator
 from frequenz.client.dispatch.types import Dispatch, Frequency
@@ -63,32 +62,16 @@ class ActorTestEnv:
 @fixture
 async def actor_env() -> AsyncIterator[ActorTestEnv]:
     """Return an actor test environment."""
-    T = TypeVar("T")
-
-    class YieldingSender(Sender[T]):
-        """A sender that yields after sending.
-
-        For testing we want to manipulate the time after a call to send.
-
-        The normal sender normally doesn't yield/await, robbing us of the
-        opportunity to manipulate the time.
-        """
-
-        async def send(self, msg: T) -> None:
-            """Send the value and yield."""
-            await super().send(msg)
-            await asyncio.sleep(1)
-
-    updated_dispatches = Broadcast[DispatchEvent]("updated_dispatches")
-    ready_dispatches = Broadcast[Dispatch]("ready_dispatches")
+    updated_dispatches = Broadcast[DispatchEvent](name="updated_dispatches")
+    ready_dispatches = Broadcast[Dispatch](name="ready_dispatches")
     microgrid_id = randint(1, 100)
 
     actor = DispatchingActor(
         microgrid_id=microgrid_id,
         grpc_channel=MagicMock(),
         svc_addr="localhost",
-        updated_dispatch_sender=YieldingSender(updated_dispatches),
-        ready_dispatch_sender=YieldingSender(ready_dispatches),
+        updated_dispatch_sender=updated_dispatches.new_sender(),
+        ready_dispatch_sender=ready_dispatches.new_sender(),
     )
 
     client = FakeClient()
@@ -144,9 +127,12 @@ async def _test_new_dispatch_created(
         case Deleted(dispatch) | Updated(dispatch):
             assert False, "Expected a created event"
         case Created(dispatch):
-            sample.update_time = dispatch.update_time
-            sample.create_time = dispatch.create_time
-            sample.id = dispatch.id
+            sample = replace(
+                sample,
+                update_time=dispatch.update_time,
+                create_time=dispatch.create_time,
+                id=dispatch.id,
+            )
             assert dispatch == sample
 
     return sample
@@ -159,16 +145,19 @@ async def test_existing_dispatch_updated(
 ) -> None:
     """Test that an existing dispatch is updated."""
     sample = generator.generate_dispatch(actor_env.microgrid_id)
-    sample.active = False
-    sample.recurrence.frequency = Frequency.DAILY
+    sample = replace(
+        sample,
+        active=False,
+        recurrence=replace(sample.recurrence, frequency=Frequency.DAILY),
+    )
 
     fake_time.shift(timedelta(seconds=1))
 
-    await _test_new_dispatch_created(actor_env, sample)
+    sample = await _test_new_dispatch_created(actor_env, sample)
     fake_time.shift(timedelta(seconds=1))
 
     await actor_env.client.update(
-        sample.id,
+        dispatch_id=sample.id,
         new_fields={
             "active": True,
             "recurrence.frequency": Frequency.UNSPECIFIED,
@@ -181,11 +170,13 @@ async def test_existing_dispatch_updated(
         case Created(dispatch) | Deleted(dispatch):
             assert False, "Expected an updated event"
         case Updated(dispatch):
-            sample.update_time = dispatch.update_time
-            sample.active = True
-            sample.recurrence = replace(
-                sample.recurrence, frequency=Frequency.UNSPECIFIED
+            sample = replace(
+                sample,
+                active=True,
+                recurrence=replace(sample.recurrence, frequency=Frequency.UNSPECIFIED),
+                update_time=dispatch.update_time,
             )
+
             assert dispatch == sample
 
 
@@ -197,7 +188,7 @@ async def test_existing_dispatch_deleted(
     """Test that an existing dispatch is deleted."""
     sample = generator.generate_dispatch(actor_env.microgrid_id)
 
-    await _test_new_dispatch_created(actor_env, sample)
+    sample = await _test_new_dispatch_created(actor_env, sample)
 
     await actor_env.client.delete(sample.id)
     fake_time.shift(timedelta(seconds=1))
@@ -207,7 +198,7 @@ async def test_existing_dispatch_deleted(
         case Created(dispatch) | Updated(dispatch):
             assert False, "Expected a deleted event"
         case Deleted(dispatch):
-            sample.create_time = dispatch.create_time
+            sample = replace(sample, update_time=dispatch.update_time)
             assert dispatch == sample