test_runningjob.py 文件源码

def test_stdout_chunk_zero_size(self, get_stdout, get_info_for_rj):
        """Test RunningJob() stdout_chunk_zero_size"""

        get_info_for_rj.return_value = dict([('stdout_size', 0)])
        get_stdout.return_value = "line1\nline2\nline3\nline4\nline5\nline6"

        running_job = pygenie.jobs.RunningJob('1234-stdout-chunk',
                                              info={'status': 'SUCCEEDED'})

        chunk = running_job.stdout_chunk(10, offset=0)

        assert_equals(
            [],
            get_stdout.call_args_list
        )

        assert_equals(chunk, None)