test.py 文件源码

def test_providing_res_with_url_calls_datapusher_correctly(self):
        config['datapusher.url'] = 'http://datapusher.ckan.org'
        httpretty.HTTPretty.register_uri(
            httpretty.HTTPretty.POST,
            'http://datapusher.ckan.org/job',
            content_type='application/json',
            body=json.dumps({'job_id': 'foo', 'job_key': 'bar'}))

        package = model.Package.get('annakarenina')

        tests.call_action_api(
            self.app,
            'datastore_create',
            apikey=self.sysadmin_user.apikey,
            resource=dict(package_id=package.id, url='demo.ckan.org')
        )

        assert len(package.resources) == 4, len(package.resources)
        resource = package.resources[3]
        data = json.loads(httpretty.last_request().body)
        assert data['metadata']['resource_id'] == resource.id, data
        assert not data['metadata'].get('ignore_hash'), data
        assert data['result_url'].endswith('/action/datapusher_hook'), data
        assert data['result_url'].startswith('http://'), data