=================================== FAILURES ===================================
________________________________ test_publisher ________________________________
[gw0] linux -- Python 3.9.17 /opt/hostedtoolcache/Python/3.9.17/x64/bin/python

flow = functools.partial(<function _make_flow at 0x7f89e56bea60>, PosixPath('/home/runner/cylc-run'), PosixPath('/home/runner/cylc-run/cit-20230815T191710 0935/integration.test_publisher/test_publisher'))
scheduler = <function _make_scheduler.<locals>.__make_scheduler at 0x7f89e4f0e790>
run = functools.partial(<function _run_flow at 0x7f89e56bedc0>, <_pytest.logging.LogCaptureFixture object at 0x7f89e4bf2fa0>)
one_conf = {'scheduler': {'allow implicit tasks': True}, 'scheduling': {'graph': {'R1': 'one'}}}
port_range = (43001, 43101)

    async def test_publisher(flow, scheduler, run, one_conf, port_range):
        """It should publish deltas when the flow starts."""
        id_ = flow(one_conf)
        schd = scheduler(id_, paused_start=False)
        async with run(schd):
            # create a subscriber
            subscriber = WorkflowSubscriber(
                schd.workflow,
                host=schd.host,
                port=schd.server.pub_port,
                topics=[b'workflow']
            )
    
            async with timeout(2):
                # wait for the first delta from the workflow
>               btopic, msg = await subscriber.socket.recv_multipart()
E               asyncio.exceptions.CancelledError

tests/integration/test_publisher.py:40: CancelledError

During handling of the above exception, another exception occurred:

flow = functools.partial(<function _make_flow at 0x7f89e56bea60>, PosixPath('/home/runner/cylc-run'), PosixPath('/home/runner/cylc-run/cit-20230815T191710 0935/integration.test_publisher/test_publisher'))
scheduler = <function _make_scheduler.<locals>.__make_scheduler at 0x7f89e4f0e790>
run = functools.partial(<function _run_flow at 0x7f89e56bedc0>, <_pytest.logging.LogCaptureFixture object at 0x7f89e4bf2fa0>)
one_conf = {'scheduler': {'allow implicit tasks': True}, 'scheduling': {'graph': {'R1': 'one'}}}
port_range = (43001, 43101)

    async def test_publisher(flow, scheduler, run, one_conf, port_range):
        """It should publish deltas when the flow starts."""
        id_ = flow(one_conf)
        schd = scheduler(id_, paused_start=False)
        async with run(schd):
            # create a subscriber
            subscriber = WorkflowSubscriber(
                schd.workflow,
                host=schd.host,
                port=schd.server.pub_port,
                topics=[b'workflow']
            )
    
            async with timeout(2):
                # wait for the first delta from the workflow
>               btopic, msg = await subscriber.socket.recv_multipart()

tests/integration/test_publisher.py:40: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/hostedtoolcache/Python/3.9.17/x64/lib/python3.9/site-packages/async_timeout/__init__.py:141: in __aexit__
    self._do_exit(exc_type)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <async_timeout.Timeout object at 0x7f89e4a89590>
exc_type = <class 'asyncio.exceptions.CancelledError'>

    def _do_exit(self, exc_type: Optional[Type[BaseException]]) -> None:
        if exc_type is asyncio.CancelledError and self._state == _State.TIMEOUT:
            assert self._task is not None
            _uncancel_task(self._task)
            self._timeout_handler = None
            self._task = None
>           raise asyncio.TimeoutError
E           asyncio.exceptions.TimeoutError

/opt/hostedtoolcache/Python/3.9.17/x64/lib/python3.9/site-packages/async_timeout/__init__.py:228: TimeoutError