squash warngings, add coverage

Author: Martin Durant

setup.cfg

@@ -39,3 +39,8 @@ max-line-length = 120
 
 [bdist_wheel]
 universal=1
+
+[tool:pytest]
+markers:
+  network: Test requires an internet connection
+  slow: Skipped unless --runslow passed

streamz/core.py

@@ -23,7 +23,7 @@
 
 try:
     from distributed.client import default_client as _dask_default_client
-except ImportError:
+except ImportError:  # pragma: no cover
     _dask_default_client = None
 
 from collections.abc import Iterable
@@ -356,8 +356,6 @@ def __str__(self):
                     s = str(at)
                 elif hasattr(at, '__name__'):
                     s = getattr(self, m).__name__
-                elif hasattr(at.__class__, '__name__'):
-                    s = getattr(self, m).__class__.__name__
                 else:
                     s = None
             if s:

streamz/tests/test_core.py

@@ -407,6 +407,7 @@ def test_timed_window_ref_counts():
     _ = source.timed_window(0.01)
 
     ref1 = RefCounter()
+    assert str(ref1) == "<RefCounter count=0>"
     source.emit(1, metadata=[{'ref': ref1}])
     assert ref1.count == 1
     yield gen.sleep(0.05)
@@ -417,6 +418,13 @@ def test_timed_window_ref_counts():
     assert ref2.count == 1
 
 
+def test_mixed_async():
+    s1 = Stream(asynchronous=False)
+    with pytest.raises(ValueError):
+        Stream(asynchronous=True, upstream=s1)
+
+
+
 @gen_test()
 def test_timed_window_metadata():
     source = Stream()

streamz/tests/test_dask.py

@@ -91,6 +91,7 @@ def test_partition_then_scatter_sync(loop):
             assert L == [1, 2, 3]
 
 
+@gen_cluster(client=True)
 def test_non_unique_emit(c, s, a, b):
     """Regression for https://github.com/python-streamz/streams/issues/397
 
@@ -191,7 +192,7 @@ def test_sync_2(loop):  # noqa: F811
             assert L == list(map(inc, range(10)))
 
 
-@gen_cluster(client=True, ncores=[('127.0.0.1', 1)] * 2)
+@gen_cluster(client=True, nthreads=[('127.0.0.1', 1)] * 2)
 def test_buffer(c, s, a, b):
     source = Stream(asynchronous=True)
     L = source.scatter().map(slowinc, delay=0.5).buffer(5).gather().sink_to_list()