    def test_schema_dag(self):
        """
        Unittest schema dag info
        """
        f = towhee.dummy_input[(),('video_url', 'video_id')]()\
            .set_parallel(5).video_decode.ffmpeg[('video_url'), ('video_frame')]()\
            .image_text_embedding.clip_image[('video_frame'), ('embedding')]()\
            .image_decode.test[('video_id', 'video_frame', 'embedding'),()]()\
            .as_function()
        for i in f.dag_info.values():
            if i['op_name'] == 'start':
                expect_input = []
                expect_output = ['video_url', 'video_id']
                self.assertEqual(i['input_info'], expect_input)
                self.assertEqual(i['output_info'], expect_output)
            elif i['op_name'] == 'video-decode/ffmpeg':
                expect_input = [('start', 'video_url')]
                expect_output = ['video_frame']
                self.assertEqual(i['input_info'], expect_input)
                self.assertEqual(i['output_info'], expect_output)
            elif i['op_name'] == 'image-text-embedding/clip-image':
                expect_input = 'video_frame'
                expect_output = ['embedding']
                self.assertEqual(i['input_info'][0][1], expect_input)
                self.assertEqual(i['output_info'], expect_output)
            elif i['op_name'] == 'image-decode/test':
                expect_input_1 = 'video_id'
                expect_input_2 = 'video_frame'
                expect_input_3 = 'embedding'
                expect_output = []
                self.assertEqual(i['input_info'][0][1], expect_input_1)
                self.assertEqual(i['input_info'][1][1], expect_input_2)
                self.assertEqual(i['input_info'][2][1], expect_input_3)
                self.assertEqual(i['output_info'], expect_output)
            elif i['op_name'] == 'end':
                self.assertEqual(i['input_info'], None)
                self.assertEqual(i['output_info'], None)
Unittest schema dag info
tests/testcases/test_dag_info.py