Temporarily skip tests due to dask/distributed#8953 (#17472)

Temporarily skip tests failing due to an upstream dask change.

Authors:
  - Lawrence Mitchell (https://github.com/wence-)

Approvers:
  - GALI PREM SAGAR (https://github.com/galipremsagar)
  - Matthew Murray (https://github.com/Matt711)

URL: #17472

Author: wence-

python/custreamz/custreamz/tests/test_dataframes.py

@@ -216,7 +216,13 @@ def test_set_index():
     assert_eq(b[0], df.set_index(df.y + 1))
 
 
-def test_binary_stream_operators(stream):
+def test_binary_stream_operators(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     df = cudf.DataFrame({"x": [1, 2, 3], "y": [4, 5, 6]})
 
     expected = df.x + df.y
@@ -242,7 +248,13 @@ def test_index(stream):
     assert_eq(L[1], df.index + 5)
 
 
-def test_pair_arithmetic(stream):
+def test_pair_arithmetic(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     df = cudf.DataFrame({"x": list(range(10)), "y": [1] * 10})
 
     a = DataFrame(example=df.iloc[:0], stream=stream)
@@ -255,7 +267,13 @@ def test_pair_arithmetic(stream):
     assert_eq(cudf.concat(L), (df.x + df.y) * 2)
 
 
-def test_getitem(stream):
+def test_getitem(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     df = cudf.DataFrame({"x": list(range(10)), "y": [1] * 10})
 
     a = DataFrame(example=df.iloc[:0], stream=stream)
@@ -332,7 +350,13 @@ def test_repr_html(stream):
         assert "1" in html
 
 
-def test_setitem(stream):
+def test_setitem(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     df = cudf.DataFrame({"x": list(range(10)), "y": [1] * 10})
 
     sdf = DataFrame(example=df.iloc[:0], stream=stream)
@@ -356,7 +380,13 @@ def test_setitem(stream):
     assert_eq(L[-1], df.mean())
 
 
-def test_setitem_overwrites(stream):
+def test_setitem_overwrites(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     df = cudf.DataFrame({"x": list(range(10))})
     sdf = DataFrame(example=df.iloc[:0], stream=stream)
     stream = sdf.stream
@@ -413,8 +443,14 @@ def test_setitem_overwrites(stream):
     ],
 )
 def test_rolling_count_aggregations(
-    op, window, m, pre_get, post_get, kwargs, stream
+    request, op, window, m, pre_get, post_get, kwargs, stream
 ):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream) and len(kwargs) == 0,
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     index = pd.DatetimeIndex(
         pd.date_range("2000-01-01", "2000-01-03", freq="1h")
     )
@@ -808,7 +844,13 @@ def test_reductions_with_start_state(stream):
     assert output2[0] == 360
 
 
-def test_rolling_aggs_with_start_state(stream):
+def test_rolling_aggs_with_start_state(request, stream):
+    request.applymarker(
+        pytest.mark.xfail(
+            isinstance(stream, DaskStream),
+            reason="https://github.com/dask/distributed/issues/8953",
+        )
+    )
     example = cudf.DataFrame({"name": [], "amount": []}, dtype="float64")
     sdf = DataFrame(stream, example=example)
     output0 = (