some initial tests

Author: sanderegg

services/dask-sidecar/tests/unit/test_utils_dask.py

@@ -40,25 +40,32 @@
 def test_publish_event(
     dask_client: distributed.Client, job_id: str, task_owner: TaskOwner
 ):
-    dask_pub = distributed.Pub("some_topic", client=dask_client)
-    dask_sub = distributed.Sub("some_topic", client=dask_client)
     event_to_publish = TaskProgressEvent(
         job_id=job_id,
         msg="the log",
         progress=1,
         task_owner=task_owner,
     )
-    publish_event(dask_pub=dask_pub, event=event_to_publish)
+    dask_client.log_event("some_topic", event_to_publish.model_dump_json())
+    # publish_event(dask_pub=dask_pub, event=event_to_publish)
 
     # NOTE: this tests runs a sync dask client,
     # and the CI seems to have sometimes difficulties having this run in a reasonable time
     # hence the long time out
-    message = dask_sub.get(timeout=DASK_TESTING_TIMEOUT_S)
-    assert message is not None
-    assert isinstance(message, str)
-    received_task_log_event = TaskProgressEvent.model_validate_json(message)
+    events = dask_client.get_events("some_topic")
+    assert events is not None
+    assert isinstance(events, tuple)
+    assert len(events) == 1
+    assert isinstance(events[0], tuple)
+    received_task_log_event = TaskProgressEvent.model_validate_json(events[0][1])
     assert received_task_log_event == event_to_publish
 
+    # message = dask_sub.get(timeout=DASK_TESTING_TIMEOUT_S)
+    # assert message is not None
+    # assert isinstance(message, str)
+    # received_task_log_event = TaskProgressEvent.model_validate_json(message)
+    # assert received_task_log_event == event_to_publish
+
 
 async def test_publish_event_async(
     async_dask_client: distributed.Client, job_id: str, task_owner: TaskOwner