add no node and multiples nodes unit test for ExtractTxtOp

tests/op/extract/load/test_txt_op.py

@@ -1,8 +1,8 @@
 import unittest
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 from uniflow.node import Node
-from uniflow.op.extract.load.txt_op import ExtractTxtOp, ProcessTxtOp
+from uniflow.op.extract.load.txt_op import ExtractTxtOp
 
 
 class TestExtractTxtOp(unittest.TestCase):
@@ -12,48 +12,62 @@ def setUp(self):
     @patch(
         "uniflow.op.extract.load.txt_op.read_file", return_value="mocked file content"
     )
-    def test_call(self, mock_read_file):
+    def test_call_with_empty_node(self, mock_read_file):
         # arrange
-        node = Node(name="node1", value_dict={"filename": "mocked_file_path"})
+        nodes = []
 
         # act
-        output_nodes = self.extract_txt_op([node])
+        output_nodes = self.extract_txt_op(nodes)
 
         # assert
-        mock_read_file.assert_called_once_with("mocked_file_path")
-        self.assertEqual(len(output_nodes), 1)
-        self.assertEqual(output_nodes[0].value_dict["text"], "mocked file content")
+        mock_read_file.assert_not_called()
+        self.assertEqual(len(output_nodes), 0)
 
+    @patch(
+        "uniflow.op.extract.load.txt_op.read_file", return_value="mocked file content"
+    )
+    def test_call_with_node_without_filename(self, mock_read_file):
+        # arrange
+        node = Node(name="node1", value_dict={})
 
-class TestProcessTxtOp(unittest.TestCase):
-    def setUp(self):
-        self.process_txt_op = ProcessTxtOp(name="process_txt_op")
-
-    def test_empty_input(self):
-        node = Node(name="node1", value_dict={"text": ""})
-
-        output_nodes = self.process_txt_op([node])
+        # act
+        with self.assertRaises(KeyError):
+            self.extract_txt_op([node])
 
-        self.assertEqual(len(output_nodes), 1)
-        self.assertEqual(output_nodes[0].value_dict["text"][0], "")
+        # assert
+        mock_read_file.assert_not_called()
 
-    def test_whitespace_input(self):
-        node = Node(name="node1", value_dict={"text": " \n \n "})
+    @patch(
+        "uniflow.op.extract.load.txt_op.read_file", return_value="mocked file content"
+    )
+    def test_call_with_node(self, mock_read_file):
+        # arrange
+        node = Node(name="node1", value_dict={"filename": "mocked_file_path"})
 
-        output_nodes = self.process_txt_op([node])
+        # act
+        output_nodes = self.extract_txt_op([node])
 
+        # assert
+        mock_read_file.assert_called_once_with("mocked_file_path")
         self.assertEqual(len(output_nodes), 1)
-        self.assertEqual(output_nodes[0].value_dict["text"][0], "")
+        self.assertEqual(output_nodes[0].value_dict["text"], "mocked file content")
 
-    def test_call(self):
-        node = Node(
-            name="node1", value_dict={"text": "This is a test\nThis is another test"}
-        )
+    @patch(
+        "uniflow.op.extract.load.txt_op.read_file", return_value="mocked file content"
+    )
+    def test_call_with_multiple_nodes(self, mock_read_file):
+        # arrange
+        node1 = Node(name="node1", value_dict={"filename": "mocked_file_path1"})
+        node2 = Node(name="node2", value_dict={"filename": "mocked_file_path2"})
+        nodes = [node1, node2]
 
-        output_nodes = self.process_txt_op([node])
+        # act
+        output_nodes = self.extract_txt_op(nodes)
 
-        self.assertEqual(len(output_nodes), 1)
-        self.assertEqual(
-            output_nodes[0].value_dict["text"],
-            ["This is a test", "This is another test"],
+        # assert
+        mock_read_file.assert_has_calls(
+            [call("mocked_file_path1"), call("mocked_file_path2")], any_order=True
         )
+        self.assertEqual(len(output_nodes), 2)
+        self.assertEqual(output_nodes[0].value_dict["text"], "mocked file content")
+        self.assertEqual(output_nodes[1].value_dict["text"], "mocked file content")

uniflow/op/extract/load/txt_op.py

@@ -1,7 +1,6 @@
 """Extract txt op."""
 
 import copy
-import re
 from typing import Sequence
 
 from uniflow.node import Node
@@ -33,30 +32,3 @@ def __call__(self, nodes: Sequence[Node]) -> Sequence[Node]:
                 )
             )
         return output_nodes
-
-
-class ProcessTxtOp(Op):
-    """Process txt Op Class."""
-
-    def __call__(self, nodes: Sequence[Node]) -> Sequence[Node]:
-        """Run Model Op.
-
-        Args:
-            nodes (Sequence[Node]): Nodes to run.
-
-        Returns:
-            Sequence[Node]: Nodes after running.
-        """
-        output_nodes = []
-        for node in nodes:
-            value_dict = copy.deepcopy(node.value_dict)
-            text = value_dict["text"]
-            text = re.split(r"\s*\n\s*", text.strip())
-            output_nodes.append(
-                Node(
-                    name=self.unique_name(),
-                    value_dict={"text": text},
-                    prev_nodes=[node],
-                )
-            )
-        return output_nodes