test_phrases.py 文件源码

def test_produce_phrases_with_ok_job(self,
                                         cached_phrases,
                                         corpus_base_path,
                                         get_pipeline_for_component,
                                         build_phrases,
                                         get_job_finish_status
                                         ):
        from eea.corpus.processing.phrases.process import produce_phrases
        from pkg_resources import resource_filename

        content = ['hello', 'world']
        env = {'phash_id': 'X', 'file_name': 'x.csv', 'text_column': 'text'}
        base_path = resource_filename('eea.corpus', 'tests/fixtures/')

        corpus_base_path.return_value = base_path
        cached_phrases.return_value = ['something', 'else']
        get_job_finish_status.return_value = True

        stream = produce_phrases(content, env, {})

        assert list(stream) == ['something', 'else']
        assert corpus_base_path.call_count == 1
        assert get_pipeline_for_component.call_count == 0
        assert build_phrases.call_count == 0
        assert cached_phrases.call_count == 1