[fjetter]

Apparently some of these tests are not rock solid.

E.g. in #6166

https://github.com/dask/distributed/runs/6101034638?check_suite_focus=true

Based on the traceback, it looks like the mock.patch was not properly removed after leaving the context manager. To rule this out and see if there is something else ongoing, I removed the mocks in favour of some event/lock/subclass magic.

I realise that these tests are not great and are extremely difficult to maintain. We can have a conversation about removing some of them but I would like to not have this conversation right now. I consider this new approach easier to reason about and at least slightly easier to maintain.

As part of #5896 I very much hope that we can remove some, if not all of these tests and replace them with some much easier unit tests. I would like to have the conversation about removing code once we are in a position to replace them with something simpler. This is not in scope for this PR

Traceback

    @gen_cluster(client=True)
    async def test_flight_cancelled_error(c, s, a, b):
        """One worker with one thread. We provoke an flight->cancelled transition
        and let the task err."""
        lock = asyncio.Lock()
        await lock.acquire()
    
        async def wait_and_raise(*args, **kwargs):
            async with lock:
                raise RuntimeError()
    
        with mock.patch.object(
            distributed.worker,
            "get_data_from_worker",
            side_effect=wait_and_raise,
        ):
            fut1 = c.submit(inc, 1, workers=[a.address], allow_other_workers=True)
            fut2 = c.submit(inc, fut1, workers=[b.address])
    
            await wait_for_state(fut1.key, "flight", b)
            fut2.release()
            fut1.release()
            await wait_for_state(fut1.key, "cancelled", b)
    
        lock.release()
        # At this point we do not fetch the result of the future since the future
        # itself would raise a cancelled exception. At this point we're concerned
        # about the worker. The task should transition over error to be eventually
        # forgotten since we no longer hold a ref.
        while fut1.key in b.tasks:
            await asyncio.sleep(0.01)
    
        # Everything should still be executing as usual after this
>       assert await c.submit(sum, c.map(inc, range(10))) == sum(map(inc, range(10)))

distributed\tests\test_cancelled_state.py:247: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
distributed\client.py:294: in _result
    raise exc.with_traceback(tb)
C:\Miniconda3\envs\dask-distributed\lib\unittest\mock.py:2164: in _execute_mock_call
    result = await effect(*args, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   raise RuntimeError()
E   RuntimeError

distributed\tests\test_cancelled_state.py:223: RuntimeError
---------------------------- Captured stdout call -----------------------------
Dumped cluster state to test_cluster_dump\test_flight_cancelled_error.yaml
---------------------------- Captured stderr call -----------------------------
2022-04-20 19:43:01,398 - distributed.worker - ERROR - 
Traceback (most recent call last):
  File "d:\a\distributed\distributed\distributed\worker.py", line 3037, in gather_dep
    response = await get_data_from_worker(
  File "C:\Miniconda3\envs\dask-distributed\lib\unittest\mock.py", line 2164, in _execute_mock_call
    result = await effect(*args, **kwargs)
  File "D:\a\distributed\distributed\distributed\tests\test_cancelled_state.py", line 223, in wait_and_raise
    raise RuntimeError()
RuntimeError
2022-04-20 19:43:01,400 - distributed.utils - ERROR - 
Traceback (most recent call last):
  File "d:\a\distributed\distributed\distributed\utils.py", line 693, in log_errors
    yield
  File "d:\a\distributed\distributed\distributed\worker.py", line 3037, in gather_dep
    response = await get_data_from_worker(
  File "C:\Miniconda3\envs\dask-distributed\lib\unittest\mock.py", line 2164, in _execute_mock_call
    result = await effect(*args, **kwargs)
  File "D:\a\distributed\distributed\distributed\tests\test_cancelled_state.py", line 223, in wait_and_raise
    raise RuntimeError()
RuntimeError
2022-04-20 19:43:01,401 - tornado.application - ERROR - Exception in callback functools.partial(<bound method IOLoop._discard_future_result of <tornado.platform.asyncio.AsyncIOLoop object at 0x000002DC80D8D430>>, <Task finished name='Task-74891' coro=<Worker.gather_dep() done, defined at d:\a\distributed\distributed\distributed\worker.py:2981> exception=RuntimeError()>)
Traceback (most recent call last):
  File "C:\Miniconda3\envs\dask-distributed\lib\site-packages\tornado\ioloop.py", line 741, in _run_callback
    ret = callback()
  File "C:\Miniconda3\envs\dask-distributed\lib\site-packages\tornado\ioloop.py", line 765, in _discard_future_result
    future.result()
  File "d:\a\distributed\distributed\distributed\worker.py", line 3037, in gather_dep
    response = await get_data_from_worker(
  File "C:\Miniconda3\envs\dask-distributed\lib\unittest\mock.py", line 2164, in _execute_mock_call
    result = await effect(*args, **kwargs)
  File "D:\a\distributed\distributed\distributed\tests\test_cancelled_state.py", line 223, in wait_and_raise
    raise RuntimeError()
RuntimeError

[github-actions]

Unit Test Results

       15 files  ±    0         15 suites  ±0   6h 58m 32s ⏱️ + 5m 49s
  2 727 tests ±    0    2 646 ✔️ +    1    81 💤 +  1  0 ❌  - 2 
20 510 runs  +323  19 524 ✔️ +295  986 💤 +30  0 ❌  - 2 

Results for commit a7c0df0. ± Comparison against base commit 6947873.

[mrocklin]

Woo! +1 on using subclasses over mocks. FWIW I find them to be both more understandable and less magical.