tests.py 文件源码

def test_consume_feed_task_chaining():
    """ Consume test rss feed through connected fetcher and parser and publisher tasks.
        Make sure the parser have created the RssNotification instances, and
        all the messages were sent.

        It should be run from under the django shell/script.
    """
    delete_test_rss_feed_notifications()

    task_flow = chain(fetch.s(), parse.s(), publish.s())
    task_flow.delay(_rss_file_path()).get()