add tests and docs

Author: CJ-Wright

streamz/core.py

@@ -622,6 +622,10 @@ class filter(Stream):
     predicate : function
         The predicate. Should return True or False, where
         True means that the predicate is satisfied.
+    *args :
+        The arguments to pass to the predicate.
+    **kwargs:
+        Keyword arguments to pass to predicate
 
     Examples
     --------
@@ -649,7 +653,6 @@ def update(self, x, who=None):
             return self._emit(x)
 
 
-
 @Stream.register_api()
 class accumulate(Stream):
     """ Accumulate results with previous state

streamz/tests/test_core.py

@@ -83,6 +83,26 @@ def test_filter():
     assert L == [0, 2, 4, 6, 8]
 
 
+def test_filter_args():
+    source = Stream()
+    L = source.filter(lambda x, n: x % n == 0, 2).sink_to_list()
+
+    for i in range(10):
+        source.emit(i)
+
+    assert L == [0, 2, 4, 6, 8]
+
+
+def test_filter_kwargs():
+    source = Stream()
+    L = source.filter(lambda x, n=1: x % n == 0, n=2).sink_to_list()
+
+    for i in range(10):
+        source.emit(i)
+
+    assert L == [0, 2, 4, 6, 8]
+
+
 def test_filter_none():
     source = Stream()
     L = source.filter(None).sink_to_list()