Support networkx to represent components as a graph

kpops/pipeline_generator/pipeline.py

@@ -6,6 +6,7 @@
 from collections.abc import Iterator
 from contextlib import suppress
 from pathlib import Path
+import matplotlib.pyplot as plt
 
 import networkx as nx
 import yaml

tests/components/test_streams_app.py

@@ -429,3 +429,62 @@ async def test_should_clean_streams_app_and_deploy_clean_up_job_and_delete_clean
                 "test-namespace", self.STREAMS_APP_CLEAN_NAME, dry_run
             ),
         ]
+
+    @pytest.mark.asyncio
+    async def test_get_topics(
+        self, config: PipelineConfig, handlers: ComponentHandlers
+    ):
+        streams_app = StreamsApp(
+            name=self.STREAMS_APP_NAME,
+            config=config,
+            handlers=handlers,
+            **{
+                "namespace": "test-namespace",
+                "app": {
+                    "streams": {"brokers": "fake-broker:9092"},
+                },
+                "from": {
+                    "topics": {
+                        "example-input": {"type": "input"},
+                        "b": {"type": "input"},
+                        "a": {"type": "input"},
+                        "topic-extra2": {"type": "extra", "role": "role2"},
+                        "topic-extra3": {"type": "extra", "role": "role2"},
+                        "topic-extra": {"type": "extra", "role": "role1"},
+                        ".*": {"type": "input-pattern"},
+                        "example.*": {
+                            "type": "extra-pattern",
+                            "role": "another-pattern",
+                        },
+                    }
+                },
+            },
+        )
+        assert streams_app.get_input_topics() == ["example-input", "b", "a"]
+        assert streams_app.get_extra_input_topics() == {
+            "role1": ["topic-extra"],
+            "role2": ["topic-extra2", "topic-extra3"],
+        }
+
+    @pytest.mark.asyncio
+    async def test_get_output_topic(
+        self, config: PipelineConfig, handlers: ComponentHandlers
+    ):
+        streams_app = StreamsApp(
+            name=self.STREAMS_APP_NAME,
+            config=config,
+            handlers=handlers,
+            **{
+                "namespace": "test-namespace",
+                "app": {
+                    "streams": {"brokers": "fake-broker:9092"},
+                },
+                "from": {
+                    "topics": {
+                        "example-input": {"type": "input"},
+                    }
+                },
+                "to": {"topics": {"example-output": {"type": "output"}}},
+            },
+        )
+        assert streams_app.get_output_topic() == "example-output"

tests/pipeline/resources/pipeline-with-loop/defaults.yaml

@@ -19,4 +19,4 @@ streams-app:
     labels:
       pipeline: ${pipeline_name}
     streams:
-      optimizeLeaveGroupBehavior: false
+      optimizeLeaveGroupBehavior: false

tests/pipeline/resources/pipeline-with-loop/pipeline.yaml

@@ -31,4 +31,4 @@
   to:
     topics:
       my-output-topic:
-        type: output
+        type: output