Don't manually tamper with the state

distributed/tests/test_worker_state_machine.py

@@ -427,11 +427,12 @@ async def test_self_denounce_missing_data(c, s, a):
     x = c.submit(inc, 1, key="x")
     await x
 
-    # Manually wipe x from the worker
-    # (Using an endpoint like remove-replicas would inform the scheduler).
-    del a.data["x"]
-    del a.tasks["x"]
-    a.validate_state()
+    # Wipe x from the worker and lose the message that would inform the scheduler
+    a.handle_remove_replicas(keys=["x"], stimulus_id="test")
+    assert a.batched_stream.buffer == [
+        {"key": "x", "stimulus_id": "test", "op": "release-worker-data"}
+    ]
+    a.batched_stream.buffer.clear()
 
     y = c.submit(inc, x, key="y")
     # The scheduler tries computing y, but a responds that x is not available.