test_harvester.py 文件源码

def test_harvest_before_download_null_url_stops_gather_stage(self):

        plugin = p.get_plugin('test_rdf_harvester')

        source_url = 'http://return.none'

        # Mock the GET request to get the file
        httpretty.register_uri(httpretty.GET, source_url,
                               body=self.rdf_content,
                               content_type=self.rdf_content_type)

        # The harvester will try to do a HEAD request first so we need to mock
        # this as well
        httpretty.register_uri(httpretty.HEAD, source_url,
                               status=405,
                               content_type=self.rdf_content_type)

        harvest_source = self._create_harvest_source(source_url)
        self._create_harvest_job(harvest_source['id'])
        self._run_jobs(harvest_source['id'])
        self._gather_queue(1)

        eq_(plugin.calls['before_download'], 1)

        # Run the jobs to mark the previous one as Finished
        self._run_jobs()

        # Check that the file was not requested
        assert 'return.none' not in httpretty.last_request().headers['host']

        # Get the harvest source with the udpated status
        harvest_source = h.call_action('harvest_source_show',
                                       id=harvest_source['id'])

        last_job_status = harvest_source['status']['last_job']

        eq_(last_job_status['status'], 'Finished')

        eq_(last_job_status['stats']['added'], 0)