Tokenization-related test tweaks (backport from #8185)

distributed/deploy/tests/test_adaptive.py

@@ -203,7 +203,7 @@ async def test_adapt_quickly():
         assert len(adapt.log) == 1
 
         # Scale up when there is plenty of available work
-        futures = client.map(slowinc, range(1000), delay=0.100)
+        futures = client.map(slowinc, range(2, 1002), delay=0.100)
         while len(adapt.log) == 1:
             await asyncio.sleep(0.01)
         assert len(adapt.log) == 2

distributed/shuffle/tests/test_shuffle.py

@@ -2479,17 +2479,14 @@ def test_set_index_with_existing_index(client):
 
 
 def test_sort_values_with_existing_divisions(client):
-    "Regression test for #8165"
+    """Regression test for #8165"""
     df = pd.DataFrame(
         {"a": np.random.randint(0, 3, 20), "b": np.random.randint(0, 3, 20)}
     )
-    ddf = dd.from_pandas(
-        df,
-        npartitions=4,
-    )
+    ddf = dd.from_pandas(df, npartitions=4)
     with dask.config.set({"dataframe.shuffle.method": "p2p"}):
         ddf = ddf.set_index("a").sort_values("b")
-        result = client.compute(ddf, sync=True)
+        result = ddf.compute()
         dd.assert_eq(
             result,
             df.set_index("a").sort_values("b"),

distributed/tests/test_scheduler.py

@@ -624,7 +624,7 @@ def func(first, second):
         secede()
         second.wait()
 
-    fs = c.map(func, [first] * 5, [second] * 5)
+    fs = c.map(func, [first] * 5, [second] * 5, key=[f"x{i}" for i in range(5)])
     await async_poll_for(lambda: a.state.executing, timeout=5)
 
     await first.set()
@@ -2641,7 +2641,7 @@ async def test_adaptive_target(c, s, a, b):
         assert s.adaptive_target() == 1
 
         # Long task
-        x = c.submit(slowinc, 1, delay=0.5)
+        x = c.submit(slowinc, 1, delay=0.4)
         while x.key not in s.tasks:
             await asyncio.sleep(0.01)
         assert s.adaptive_target(target_duration=".1s") == 1  # still one