Repository: just-work/fffw

Issue:
Copy codec bug
```
source - split(1) - split(3) - Copy x 3
```

In this case Copy codec is connected to split(1) instead of source, followed by some errors.

```
        src = edge.input
        # Ensure that edge is connected to a source with only split filters
        # in between.
        source_edge = edge
        while isinstance(src, filters.Split):
            source_edge = src.input
            src = source_edge.input
        if not isinstance(src, base.Source):
            raise ValueError('copy codec can be connected only to source')
        src = edge.input
        if isinstance(src, filters.Split):
            # Remove current edge from filter graph
            src.disconnect(edge)
        return super().connect_edge(source_edge)
```
Source edge must be used instead of current one.

Discussion:

Patch:
diff --git a/fffw/encoding/codecs.py b/fffw/encoding/codecs.py
index cfe68f9..4857c38 100644
--- a/fffw/encoding/codecs.py
+++ b/fffw/encoding/codecs.py
@@ -82,6 +82,53 @@ class Copy(outputs.Codec):
 
         :returns: edge pointing to an input stream
         """
+        # Running parent method for side effects like stream validation, like if
+        # Copy is compatible with filter graph.
+        super().connect_edge(edge)
+
+        # Ensure that between source stream and copy codec there is no
+        # processing filters. Only split filter is allowed.
+        src = self._validate_filter_chain(edge)
+
+        if edge.input is src:
+            # Copy codec is being connected directly to Source, no more actions
+            # are needed.
+            return edge
+
+        # There are some Splits between Source and Copy. Current edge is not
+        # needed anymore because new Edge will be added directly to the Source.
+        # Recursively removing it from Splits chain.
+        self._remove_edge(edge)
+        # Connecting Copy to a Source directly using new node.
+        src.connect_dest(self)
+        return self.edge
+
+    def _remove_edge(self, edge: base.Edge) -> None:
+        """
+        Remove edge mentions from graph and current instance like it never
+        existed.
+
+        This method is used for reconnecting Copy codec from an end of Split
+        filter chain directly to Source.
+        """
+        # Remove and edge from existing Split filter chain
+        split = edge.input
+        if not isinstance(split, filters.Split):  # pragma: no cover
+            # Method is only called from connect_edge() in case of split
+            # filter presence.
+            raise TypeError("Can't disconnect and edge from real filter")
+        split.disconnect(edge)
+        # As the Edge is thrown away, forgot about it.
+        self._edge = None
+
+    @staticmethod
+    def _validate_filter_chain(edge: base.Edge) -> base.Source:
+        """
+        Ensures that Copy codec is being connected to a filter chain that
+        contains only Split filters.
+
+        :returns: Source stream passed to Copy codec.
+        """
         src = edge.input
         # Ensure that edge is connected to a source with only split filters
         # in between.
@@ -89,8 +136,4 @@ class Copy(outputs.Codec):
             src = src.input.input
         if not isinstance(src, base.Source):
             raise ValueError('copy codec can be connected only to source')
-        src = edge.input
-        if isinstance(src, filters.Split):
-            # Remove current edge from filter graph
-            edge = src.disconnect(edge)
-        return super().connect_edge(edge)
+        return src
diff --git a/fffw/graph/base.py b/fffw/graph/base.py
index fe5b40d..6881fde 100644
--- a/fffw/graph/base.py
+++ b/fffw/graph/base.py
@@ -93,9 +93,11 @@ class Dest(Traversable):
         :return Edge: connected edge
         """
         if not isinstance(edge, Edge):
-            raise ValueError("Only edge allowed")
+            raise TypeError("Only edge allowed")
         if self._edge is not None:
-            raise RuntimeError("Dest is already connected to %s" % self._edge)
+            raise RuntimeError("Dest input edge is already connected")
+        if edge.output is not self:
+            raise ValueError("Edge output is connected to another dest")
         self._edge = edge
         return edge
 
@@ -392,7 +394,9 @@ class Node(Traversable, abc.ABC):
         :returns: connected edge
         """
         if not isinstance(edge, Edge):
-            raise ValueError("only edge allowed")
+            raise TypeError("only edge allowed")
+        if edge.output is not self:
+            raise ValueError("Edge output is connected to another node")
         self.inputs[self.inputs.index(None)] = edge
         return edge
 


Test patch:
diff --git a/tests/test_graph.py b/tests/test_graph.py
index c6edf1b..983a7e2 100644
--- a/tests/test_graph.py
+++ b/tests/test_graph.py
@@ -7,6 +7,7 @@ from fffw.encoding import inputs, outputs, codecs
 from fffw.encoding.complex import FilterComplex
 from fffw.encoding.filters import *
 from fffw.graph import *
+from fffw.graph import base
 from fffw.wrapper import param
 
 
@@ -35,6 +36,71 @@ class FdkAAC(codecs.AudioCodec):
         return replace(ensure_audio(*metadata), bitrate=self.bitrate)
 
 
+class SourceImpl(base.Source):
+
+    @property
+    def name(self) -> str:  # pragma: no cover
+        return ''
+
+
+class NodeImpl(base.Node):
+
+    @property
+    def args(self) -> str:  # pragma: no cover
+        return ''
+
+
+class DestImpl(base.Dest):
+    pass
+
+
+class GraphBaseTestCase(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.source = SourceImpl(VIDEO)
+        self.node = NodeImpl()
+        self.another = NodeImpl()
+        self.dest = DestImpl()
+        self.source_edge = base.Edge(self.source, self.node)
+        self.inter_edge = base.Edge(self.node, self.another)
+        self.dest_edge = base.Edge(self.another, self.dest)
+
+    def test_node_connect_edge_validation(self):
+        """
+        Checks edge validation for Node.
+        """
+
+        with self.subTest("only edge allowed"):
+            with self.assertRaises(TypeError):
+                self.node.connect_edge(object())  # type: ignore
+
+        with self.subTest("edge output cross-link"):
+            with self.assertRaises(ValueError):
+                self.node.connect_edge(self.dest_edge)
+
+        with self.subTest("success"):
+            self.node.connect_edge(self.source_edge)
+
+    def test_dest_connect_edge_validation(self):
+        """
+        Checks edge validation for Dest.
+        """
+        with self.subTest("only edge allowed"):
+            with self.assertRaises(TypeError):
+                self.dest.connect_edge(object())  # type: ignore
+
+        with self.subTest("edge output cross-link"):
+            with self.assertRaises(ValueError):
+                self.dest.connect_edge(self.source_edge)
+
+        with self.subTest("success"):
+            self.dest.connect_edge(self.dest_edge)
+
+        with self.subTest("slot is busy"):
+            with self.assertRaises(RuntimeError):
+                self.dest.connect_edge(self.dest_edge)
+
+
 class FilterGraphBaseTestCase(TestCase):
 
     def setUp(self) -> None:
@@ -322,9 +388,9 @@ class FilterGraphTestCase(FilterGraphBaseTestCase):
         vs2 | c
         vs3 | c
         expected = (
-            deepcopy(vs1.meta.scenes) +
-            deepcopy(vs2.meta.scenes) +
-            deepcopy(vs3.meta.scenes)
+                deepcopy(vs1.meta.scenes) +
+                deepcopy(vs2.meta.scenes) +
+                deepcopy(vs3.meta.scenes)
         )
         assert len(expected) == 3
         current_duration = TS(0)
@@ -523,12 +589,41 @@ class CopyCodecTestCase(FilterGraphBaseTestCase):
         s1 = split
         s2 = split | Scale(1920, 1080)
 
-        s1 > codecs.Copy(kind=VIDEO)
+        copy = s1 > codecs.Copy(kind=VIDEO)
 
         # one output left
         self.assertListEqual(split.outputs, [s2.input])
         # split is disabled because of single output
         self.assertFalse(split.enabled)
+        # copy codec is connected to source
+        self.assertIs(copy.edge.input, self.source.video)
+
+    def test_split_disconnect_transient(self):
+        """
+        With multiple splits, copy codec is being disconnected from all of them.
+        """
+        video = self.source.video
+        inter = video | Split(VIDEO, output_count=1)
+        split = inter | Split(VIDEO, output_count=2)
+        s1 = split
+        s2 = split | Scale(1920, 1080)
+
+        copy = s1 > codecs.Copy(kind=VIDEO)
+
+        # one output left
+        self.assertListEqual(split.outputs, [s2.input])
+        # split is disabled because of single output
+        self.assertFalse(split.enabled)
+
+        # intermediate split is still connected to another split
+        self.assertIs(inter.output.output, split)
+        # copy codec is connected to source
+        self.assertIs(copy.edge.input, video)
+        # source is still connected to split
+        edges = video._outputs
+        expected = [copy.edge, inter.input]
+        self.assertEqual(len(edges), 2)
+        self.assertSetEqual(set(edges), set(expected))
 
     def test_split_disconnect_on_single_output(self):
         """