avoid plugins overriding existing methods

Author: roveo

streamz/core.py

@@ -306,6 +306,12 @@ def wrapped(*args, **kwargs):
 
     @classmethod
     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"to {cls.__name__}: duplicate method name."
+            )
+
         def stub(*args, **kwargs):
             """ Entrypoints-based streamz plugin. Will be loaded on first call. """
             node = entry_point.load()

streamz/plugins.py

@@ -1,10 +1,22 @@
+import warnings
+
 import pkg_resources
 
 
+def try_register(cls, entry_point, *modifier):
+    try:
+        cls.register_plugin_entry_point(entry_point, *modifier)
+    except ValueError:
+        warnings.warn(
+            f"Can't add {entry_point.name} from {entry_point.module_name}: "
+            "name collision with existing stream node."
+        )
+
+
 def load_plugins(cls):
     for entry_point in pkg_resources.iter_entry_points("streamz.sources"):
-        cls.register_plugin_entry_point(entry_point, staticmethod)
+        try_register(cls, entry_point, staticmethod)
     for entry_point in pkg_resources.iter_entry_points("streamz.nodes"):
-        cls.register_plugin_entry_point(entry_point)
+        try_register(cls, entry_point)
     for entry_point in pkg_resources.iter_entry_points("streamz.sinks"):
-        cls.register_plugin_entry_point(entry_point)
+        try_register(cls, entry_point)

streamz/tests/test_plugins.py

@@ -41,7 +41,7 @@ class test_source(Source):
     assert inspect.isfunction(Stream().from_test)
 
 
-def test_register_plugin_entry_point_raises():
+def test_register_plugin_entry_point_raises_type():
     class invalid_node:
         pass
 
@@ -51,3 +51,10 @@ class invalid_node:
 
     with pytest.raises(TypeError):
         Stream.test()
+
+
+def test_register_plugin_entry_point_raises_duplicate_name():
+    entry_point = MockEntryPoint("map", None)
+
+    with pytest.raises(ValueError):
+        Stream.register_plugin_entry_point(entry_point)