Repository: LibertyAces/BitSwanPump

Issue:
[MetricsCounter] Metric "error" is never incremented
There is unreachable code in https://github.com/LibertyAces/BitSwanPump/blob/372a59237e57089e44954a69fecc8c0f886291e3/bspump/pipeline.py#L176

Discussion:

Patch:
diff --git a/bspump/pipeline.py b/bspump/pipeline.py
index d4cab73d..ef75d40f 100644
--- a/bspump/pipeline.py
+++ b/bspump/pipeline.py
@@ -172,7 +172,7 @@ They are simply passed as an list of sources to a pipeline `build()` method.
 				self.MetricsCounter.add('warning', 1)
 				self.PubSub.publish("bspump.pipeline.warning!", pipeline=self)
 				return
-
+			else:
 				self.MetricsCounter.add('error', 1)
 
 			if (self._error is not None):


Test patch:
diff --git a/test/__init__.py b/test/__init__.py
index fe9ab179..218cc96f 100644
--- a/test/__init__.py
+++ b/test/__init__.py
@@ -7,4 +7,5 @@ from .file import *
 from .filter import *
 from .matrix import *
 from .test_config_defaults import *
+from .test_metrics_service import *
 
diff --git a/test/test_metrics_service.py b/test/test_metrics_service.py
new file mode 100644
index 00000000..601685ce
--- /dev/null
+++ b/test/test_metrics_service.py
@@ -0,0 +1,77 @@
+import bspump.common
+from bspump import Processor, Pipeline
+from bspump.trigger import PubSubTrigger
+from bspump.unittest import UnitTestSource, UnitTestSink
+
+
+class RaisingProcessor(Processor):
+
+    def process(self, context, event):
+        if "warning" in event or "error" in event:
+            raise Exception()
+        return event
+
+
+class WarningPipeline(Pipeline):
+
+    def __init__(self, app, id=None, config=None):
+
+        super().__init__(app, id, config)
+        self.PubSub.subscribe("bspump.pipeline.cycle_end!", self._on_finished)
+        self.Source = UnitTestSource(app, self).on(
+                PubSubTrigger(app, "unittest.go!", self.PubSub)
+            )
+        self.Processor = RaisingProcessor(app, self)
+        self.Sink = UnitTestSink(app, self)
+        self.build(
+            self.Source,
+            self.Processor,
+            self.Sink
+        )
+
+    def catch_error(self, exception, event):
+        if event == "warning":
+            return False
+        return super().catch_error(exception, event)
+
+    def _on_finished(self, event_name, pipeline):
+        self.App.stop()
+
+
+class TestMetricsService(bspump.unittest.ProcessorTestCase):
+
+    def _metric_flush(self, _, __, counter):
+        if counter.get("error"):
+            self.ErrorCount += counter["error"]
+        if counter.get("warning"):
+            self.WarningCount += counter["warning"]
+
+    def test_metrics_service(self):
+        self.ErrorCount = 0
+        self.WarningCount = 0
+
+        self.App.PubSub.subscribe(
+            "Application.Metrics.Flush!",
+            self._metric_flush
+        )
+
+        svc = self.App.get_service("bspump.PumpService")
+
+        self.Pipeline = WarningPipeline(self.App)
+
+        self.Pipeline.Source.Input = [
+            (None, "ok"),
+            (None, "warning"),
+            (None, "error"),
+        ]
+        self.Pipeline.PubSub.publish("unittest.go!")
+        svc.add_pipeline(self.Pipeline)
+        self.App.run()
+
+        self.assertEqual(
+            [({}, "ok")],
+            self.Pipeline.Sink.Output
+        )
+        self.assertEqual(1, self.WarningCount)
+        self.assertEqual(1, self.ErrorCount)
+