0001-opendatasoft-manage-HTTP-errors-56882.patch

Nicolas Roche, 13 septembre 2021 18:18

Voir les différences: en ligne côte à côte

Subject: [PATCH] opendatasoft: manage HTTP errors (#56882)

 passerelle/apps/opendatasoft/models.py | 23 ++++++++++++++----
 tests/test_opendatasoft.py             | 33 ++++++++++++++++++++++++++
 tests/utils.py                         |  4 ++--
 3 files changed, 53 insertions(+), 7 deletions(-)

     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     from django.db import models
     from django.shortcuts import get_object_or_404
     from django.urls import reverse
     from django.utils.six.moves.urllib import parse as urlparse
     from django.utils.translation import ugettext_lazy as _
     from requests import RequestException
     from passerelle.base.models import BaseQuery, BaseResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.jsonresponse import APIError
     from passerelle.utils.templates import render_to_string, validate_template
     class OpenDataSoft(BaseResource):
-...
             elif sort:
                 params['sort'] = sort
             if self.api_key:
                 params['apikey'] = self.api_key
             if limit:
                 params['rows'] = limit
             params.update(urlparse.parse_qs(filter_expression))
             result_response = self.requests.get(url, params=params)
             err_desc = result_response.json().get('error')
             if err_desc:
                 raise APIError(err_desc, http_status=200)
             try:
                 response = self.requests.get(url, params=params)
             except RequestException as e:
                 raise APIError('OpenDataSoft error: %s' % e)
             try:
                 json_response = response.json()
             except ValueError:
                 json_response = None
             if json_response and json_response.get('error'):
                 raise APIError(json_response.get('error'))
             try:
                 response.raise_for_status()
             except RequestException as e:
                 raise APIError('OpenDataSoft error: %s' % e)
             if not json_response:
                 raise APIError('OpenDataSoft error: bad JSON response')
             result = []
             for record in result_response.json().get('records'):
             for record in json_response.get('records'):
                 data = {}
                 for key, value in record.get('fields').items():
                     if key in ('id', 'text'):
                         key = 'original_%s' % key
                     data[key] = value
                 data['id'] = record.get('recordid')
                 data['text'] = render_to_string(text_template, data).strip()
                 result.append(data)

     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     import json
     import mock
     import pytest
     import utils
     from django.contrib.auth.models import User
     from requests.exceptions import ConnectionError
     from test_manager import login
     from passerelle.apps.opendatasoft.models import OpenDataSoft, Query
     from passerelle.utils import import_site
     pytestmark = pytest.mark.django_db
-...
+    )
     @pytest.fixture
     def connector():
         return utils.setup_access_rights(
             OpenDataSoft.objects.create(
                 slug='my_connector',
                 service_url='http://www.example.net',
                 api_key='my_secret',
+            )
+        )
     @pytest.fixture
     def query(connector):
         return Query.objects.create(
-...
         mocked_get.return_value = utils.FakedResponse(content=json.dumps(content), status_code=200)
         resp = app.get(endpoint, params=params, status=200)
         assert resp.json['data'][0]['original_id'] == 'original id'
         assert resp.json['data'][0]['original_text'] == 'original text'
         assert (
             resp.json['data'][0]['text']
             == "7cafcd5c692773e8b863587b2d38d6be82e023d8 - original id - original text"
+        )
     def test_call_search_errors(app, connector):
         endpoint = utils.generic_endpoint_url('opendatasoft', 'search', slug=connector.slug)
         assert endpoint == '/opendatasoft/my_connector/search'
         url = connector.service_url + '/api/records/1.0/search/'
         # Connection error
         exception = ConnectionError('Remote end closed connection without response')
         with utils.mock_url(url=url, exception=exception):
             resp = app.get(endpoint)
         assert resp.json['err']
         assert resp.json['err_desc'] == 'OpenDataSoft error: Remote end closed connection without response'
         # API error, provides HTTP status code
         with utils.mock_url(url=url, response='{"error": "Unknown dataset: foo"}', status_code=404):
             resp = app.get(endpoint)
         assert resp.json['err']
         assert resp.json['err_desc'] == 'Unknown dataset: foo'
         # HTTP error
         with utils.mock_url(url=url, response='not a json content', reason='Not Found', status_code=404):
             resp = app.get(endpoint)
         assert resp.json['err']
         assert 'OpenDataSoft error: 404 Client Error: Not Found' in resp.json['err_desc']
         # bad JSON response
         with utils.mock_url(url=url, response='not a json content', status_code=200):
             resp = app.get(endpoint)
         assert resp.json['err']
         assert resp.json['err_desc'] == 'OpenDataSoft error: bad JSON response'

     class FakedResponse(mock.Mock):
         headers = {}
         def json(self):
             return json_loads(self.content)
     def mock_url(url=None, response='', status_code=200, headers=None, exception=None):
     def mock_url(url=None, response='', status_code=200, headers=None, reason=None, exception=None):
         urlmatch_kwargs = {}
         if url:
             parsed = urlparse.urlparse(url)
             if parsed.netloc:
                 urlmatch_kwargs['netloc'] = parsed.netloc
             if parsed.path:
                 urlmatch_kwargs['path'] = parsed.path
         if not isinstance(response, str):
             response = json.dumps(response)
         @httmock.remember_called
         @httmock.urlmatch(**urlmatch_kwargs)
         def mocked(url, request):
             if exception:
                 raise exception
             return httmock.response(status_code, response, headers, request=request)
             return httmock.response(status_code, response, headers, reason, request=request)
         return httmock.HTTMock(mocked)
     def make_resource(model_class, **kwargs):
         resource = model_class.objects.create(**kwargs)
         setup_access_rights(resource)
         return resource
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-opendatasoft-manage-HTTP-errors-56882.patch