Python repro for issue nv-morpheus#360

python/tests/test_pipeline.py

@@ -19,7 +19,8 @@
 # from functools import partial
 
 # import numpy as np
-# import pytest
+
+import pytest
 
 import mrc
 import mrc.tests.test_edges_cpp as m
@@ -445,9 +446,50 @@ def on_complete():
     executor.join()
 
 
+def test_segment_init_error():
+    """
+    Test for issue #360
+    """
+
+    def gen_data():
+        yield 1
+
+    def init1(builder: mrc.Builder):
+        source = builder.make_source("source", gen_data)
+        egress = builder.get_egress("b")
+        builder.make_edge(source, egress)
+
+    def init2(builder: mrc.Builder):
+
+        def on_next(input):
+            pass
+
+        ingress = builder.get_ingress("b")
+        sink = builder.make_sink("sink", on_next)
+
+        builder.make_edge(ingress, sink)
+        raise RuntimeError("Test for #360")
+
+    pipe = mrc.Pipeline()
+
+    pipe.make_segment("TestSegment11", [], [("b", int, False)], init1)
+    pipe.make_segment("TestSegment22", [("b", int, False)], [], init2)
+
+    options = mrc.Options()
+
+    executor = mrc.Executor(options)
+    executor.register_pipeline(pipe)
+
+    with pytest.raises(RuntimeError):
+        executor.start()
+
+    executor.join()
+
+
 if (__name__ in ("__main__", )):
     test_dynamic_port_creation_good()
     test_dynamic_port_creation_bad()
     test_ingress_egress_custom_type_construction()
     test_dynamic_port_get_ingress_egress()
     test_dynamic_port_with_type_get_ingress_egress()
+    test_segment_init_error()