Merge pull request #455 from martindurant/ci_fix

CI fixes

Author: martindurant

.github/workflows/main.yaml

@@ -13,9 +13,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        CONDA_ENV: [py37, py38]
-    env:
-      STREAMZ_LAUNCH_KAFKA: true
+        CONDA_ENV: [py38, py39, py310]
+    # env:
+      # STREAMZ_LAUNCH_KAFKA: true
 
     steps:
       - name: APT

ci/environment-py310.yml

@@ -0,0 +1,27 @@
+name: test_env
+channels:
+  - conda-forge
+  - defaults
+dependencies:
+  - python=3.10
+  - pytest
+  - flake8
+  - black
+  - isort
+  - tornado
+  - toolz
+  - librdkafka
+  - dask
+  - distributed
+  - pandas
+  - python-confluent-kafka
+  - codecov
+  - coverage
+  - networkx
+  - graphviz
+  - pytest-asyncio
+  - python-graphviz
+  - bokeh
+  - ipywidgets
+  - flaky
+  - pytest-cov

ci/environment-py37.yml ci/environment-py39.ymlci/environment-py37.yml renamed to ci/environment-py39.yml

@@ -3,7 +3,7 @@ channels:
   - conda-forge
   - defaults
 dependencies:
-  - python=3.7
+  - python=3.9
   - pytest
   - flake8
   - black

setup.py

@@ -17,7 +17,7 @@
       license='BSD',
       keywords='streams',
       packages=packages + tests,
-      python_requires='>=3.7',
+      python_requires='>=3.8',
       long_description=(open('README.rst').read() if exists('README.rst')
                         else ''),
       install_requires=list(open('requirements.txt').read().strip().split('\n')),

streamz/core.py

@@ -168,7 +168,7 @@ def wrapped(*args, **kwargs):
     def register_plugin_entry_point(cls, entry_point, modifier=identity):
         if hasattr(cls, entry_point.name):
             raise ValueError(
-                f"Can't add {entry_point.name} from {entry_point.module_name} "
+                f"Can't add {entry_point.name} "
                 f"to {cls.__name__}: duplicate method name."
             )
 
@@ -178,7 +178,6 @@ def stub(*args, **kwargs):
             if not issubclass(node, Stream):
                 raise TypeError(
                     f"Error loading {entry_point.name} "
-                    f"from module {entry_point.module_name}: "
                     f"{node.__class__.__name__} must be a subclass of Stream"
                 )
             if getattr(cls, entry_point.name).__name__ == "stub":
@@ -379,13 +378,14 @@ def __str__(self):
     __repr__ = __str__
 
     def _ipython_display_(self, **kwargs):  # pragma: no cover
+        # Since this function is only called by jupyter, this import must succeed
+        from IPython.display import HTML, display
+
         try:
             import ipywidgets
             from IPython.core.interactiveshell import InteractiveShell
             output = ipywidgets.Output(_view_count=0)
         except ImportError:
-            # since this function is only called by jupyter, this import must succeed
-            from IPython.display import display, HTML
             if hasattr(self, '_repr_html_'):
                 return display(HTML(self._repr_html_()))
             else:
@@ -420,7 +420,11 @@ def remove_stream(change):
 
         output.observe(remove_stream, '_view_count')
 
-        return output._ipython_display_(**kwargs)
+        if hasattr(output, "_repr_mimebundle_"):
+            data = output._repr_mimebundle_(**kwargs)
+            return display(data, raw=True)
+        else:
+            return output._ipython_display_(**kwargs)
 
     def _emit(self, x, metadata=None):
         """
@@ -1468,18 +1472,23 @@ class zip(Stream):
 
     def __init__(self, *upstreams, **kwargs):
         self.maxsize = kwargs.pop('maxsize', 10)
-        self.condition = Condition()
+        self._condition = None
         self.literals = [(i, val) for i, val in enumerate(upstreams)
                          if not isinstance(val, Stream)]
 
         self.buffers = {upstream: deque()
                         for upstream in upstreams
                         if isinstance(upstream, Stream)}
-
         upstreams2 = [upstream for upstream in upstreams if isinstance(upstream, Stream)]
 
         Stream.__init__(self, upstreams=upstreams2, **kwargs)
 
+    @property
+    def condition(self):
+        if self._condition is None:
+            self._condition = Condition()
+        return self._condition
+
     def _add_upstream(self, upstream):
         # Override method to handle setup of buffer for new stream
         self.buffers[upstream] = deque()
@@ -1876,7 +1885,7 @@ class latest(Stream):
     _graphviz_shape = 'octagon'
 
     def __init__(self, upstream, **kwargs):
-        self.condition = Condition()
+        self._condition = None
         self.next = []
         self.next_metadata = None
 
@@ -1885,6 +1894,12 @@ def __init__(self, upstream, **kwargs):
 
         self.loop.add_callback(self.cb)
 
+    @property
+    def condition(self):
+        if self._condition is None:
+            self._condition = Condition()
+        return self._condition
+
     def update(self, x, who=None, metadata=None):
         if self.next_metadata:
             self._release_refs(self.next_metadata)

streamz/dataframe/tests/test_dataframes.py

@@ -219,7 +219,7 @@ def test_binary_stream_operators(stream):
 
     a.emit(df)
 
-    assert_eq(b[0], expected)
+    wait_for(lambda: b and b[0].equals(expected), 1)
 
 
 def test_index(stream):
@@ -246,7 +246,7 @@ def test_pair_arithmetic(stream):
     a.emit(df.iloc[:5])
     a.emit(df.iloc[5:])
 
-    assert len(L) == 2
+    wait_for(lambda: len(L) == 2, 1)
     assert_eq(pd.concat(L, axis=0), (df.x + df.y) * 2)
 
 
@@ -259,7 +259,7 @@ def test_getitem(stream):
     a.emit(df.iloc[:5])
     a.emit(df.iloc[5:])
 
-    assert len(L) == 2
+    wait_for(lambda: len(L) == 2, 1)
     assert_eq(pd.concat(L, axis=0), df[df.x > 4])
 
 
@@ -298,6 +298,7 @@ def f(x):
     a.emit(df.iloc[7:])
 
     first = df.iloc[:3]
+    wait_for(lambda: len(L) > 2, 1)
     assert assert_eq(L[0], f(first))
     assert assert_eq(L[-1], f(df))
 
@@ -382,7 +383,7 @@ def test_setitem(stream):
     df['a'] = 10
     df[['c', 'd']] = df[['x', 'y']]
 
-    assert_eq(L[-1], df.mean())
+    wait_for(lambda: L and L[-1].equals(df.mean()), 1)
 
 
 def test_setitem_overwrites(stream):

streamz/plugins.py

@@ -1,6 +1,6 @@
 import warnings
 
-import pkg_resources
+import importlib.metadata
 
 
 def try_register(cls, entry_point, *modifier):
@@ -14,9 +14,10 @@ def try_register(cls, entry_point, *modifier):
 
 
 def load_plugins(cls):
-    for entry_point in pkg_resources.iter_entry_points("streamz.sources"):
+    eps = importlib.metadata.entry_points()
+    for entry_point in eps.get("streamz.sources", []):
         try_register(cls, entry_point, staticmethod)
-    for entry_point in pkg_resources.iter_entry_points("streamz.nodes"):
+    for entry_point in eps.get("streamz.nodes", []):
         try_register(cls, entry_point)
-    for entry_point in pkg_resources.iter_entry_points("streamz.sinks"):
+    for entry_point in eps.get("streamz.sinks", []):
         try_register(cls, entry_point)

streamz/tests/py3_test_core.py

@@ -1,16 +1,16 @@
 # flake8: noqa
+import asyncio
 from time import time
-from distributed.utils_test import loop, inc  # noqa
-from tornado import gen
+from distributed.utils_test import inc  # noqa
 
 from streamz import Stream
 
 
-def test_await_syntax(loop):  # noqa
+def test_await_syntax():  # noqa
     L = []
 
     async def write(x):
-        await gen.sleep(0.1)
+        await asyncio.sleep(0.1)
         L.append(x)
 
     async def f():
@@ -25,4 +25,4 @@ async def f():
         assert 0.2 < stop - start < 0.4
         assert 2 <= len(L) <= 4
 
-    loop.run_sync(f)
+    asyncio.run(f())

streamz/tests/test_core.py

@@ -1,3 +1,4 @@
+import asyncio
 from datetime import timedelta
 from functools import partial
 import itertools
@@ -12,14 +13,15 @@
 
 from tornado.queues import Queue
 from tornado.ioloop import IOLoop
+from tornado import gen
 
 import streamz as sz
 
 from streamz import RefCounter
 from streamz.sources import sink_to_file
 from streamz.utils_test import (inc, double, gen_test, tmpfile, captured_logger,   # noqa: F401
         clean, await_for, metadata, wait_for)  # noqa: F401
-from distributed.utils_test import loop   # noqa: F401
+from distributed.utils_test import loop, loop_in_thread, cleanup   # noqa: F401
 
 
 def test_basic():
@@ -1485,37 +1487,13 @@ def dont_test_stream_kwargs(clean):  # noqa: F811
     sin.emit(1)
 
 
-@pytest.fixture
-def thread(loop):  # noqa: F811
-    from threading import Thread, Event
-    thread = Thread(target=loop.start)
-    thread.daemon = True
-    thread.start()
-
-    event = Event()
-    loop.add_callback(event.set)
-    event.wait()
-
-    return thread
-
-
 def test_percolate_loop_information(clean):  # noqa: F811
     source = Stream()
     assert not source.loop
     s = source.timed_window(0.5)
     assert source.loop is s.loop
 
 
-def test_separate_thread_without_time(loop, thread):  # noqa: F811
-    assert thread.is_alive()
-    source = Stream(loop=loop)
-    L = source.map(inc).sink_to_list()
-
-    for i in range(10):
-        source.emit(i)
-        assert L[-1] == i + 1
-
-
 def test_separate_thread_with_time(clean):  # noqa: F811
     L = []
 

streamz/tests/test_dask.py

@@ -72,10 +72,10 @@ async def test_partition_then_scatter_async(c, s, a, b):
     assert L == [1, 2, 3]
 
 
-def test_partition_then_scatter_sync(loop):
+def test_partition_then_scatter_sync():
     # Ensure partition w/ timeout before scatter works correctly for synchronous
     with cluster() as (s, [a, b]):
-        with Client(s['address'], loop=loop) as client:  # noqa: F841
+        with Client(s['address']) as client:  # noqa: F841
             start = time.monotonic()
             source = Stream()
             L = source.partition(2, timeout=.1).scatter().map(
@@ -164,24 +164,24 @@ async def test_accumulate(c, s, a, b):
     assert L[-1][1] == 3
 
 
-def test_sync(loop):  # noqa: F811
+def test_sync():  # noqa: F811
     with cluster() as (s, [a, b]):
-        with Client(s['address'], loop=loop) as client:  # noqa: F841
+        with Client(s['address']) as client:  # noqa: F841
             source = Stream()
             L = source.scatter().map(inc).gather().sink_to_list()
 
             async def f():
                 for i in range(10):
                     await source.emit(i, asynchronous=True)
 
-            sync(loop, f)
+            sync(client.loop, f)
 
             assert L == list(map(inc, range(10)))
 
 
-def test_sync_2(loop):  # noqa: F811
+def test_sync_2():  # noqa: F811
     with cluster() as (s, [a, b]):
-        with Client(s['address'], loop=loop):  # noqa: F841
+        with Client(s['address']):  # noqa: F841
             source = Stream()
             L = source.scatter().map(inc).gather().sink_to_list()
 
@@ -218,9 +218,9 @@ async def test_buffer(c, s, a, b):
     assert source.loop == c.loop
 
 
-def test_buffer_sync(loop):  # noqa: F811
+def test_buffer_sync():  # noqa: F811
     with cluster() as (s, [a, b]):
-        with Client(s['address'], loop=loop) as c:  # noqa: F841
+        with Client(s['address']) as c:  # noqa: F841
             source = Stream()
             buff = source.scatter().map(slowinc, delay=0.5).buffer(5)
             L = buff.gather().sink_to_list()
@@ -241,10 +241,11 @@ def test_buffer_sync(loop):  # noqa: F811
             assert L == list(map(inc, range(10)))
 
 
+@pytest.mark.asyncio
 @pytest.mark.xfail(reason='')
-async def test_stream_shares_client_loop(loop):  # noqa: F811
+async def test_stream_shares_client_loop():  # noqa: F811
     with cluster() as (s, [a, b]):
-        with Client(s['address'], loop=loop) as client:  # noqa: F841
+        with Client(s['address']) as client:  # noqa: F841
             source = Stream()
             d = source.timed_window('20ms').scatter()  # noqa: F841
             assert source.loop is client.loop