Produits Entr'ouvert » Passerelle

# Generated by Django 2.2.26 on 2022-05-12 15:02

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('arcgis', '0006_auto_20200401_1025'),

    ]

    operations = [

        migrations.AddField(

            model_name='arcgis',

            name='token_password',

            field=models.CharField(

                blank=True, max_length=128, verbose_name='Password of the user who wants to get a token'

            ),

        ),

        migrations.AddField(

            model_name='arcgis',

            name='token_username',

            field=models.CharField(

                blank=True, max_length=128, verbose_name='User name of the user who wants to get a token'

            ),

        ),

    ]

    token_username = models.CharField(

        max_length=128, verbose_name=_('User name of the user who wants to get a token'), blank=True

    )

    token_password = models.CharField(

        max_length=128, verbose_name=_('Password of the user who wants to get a token'), blank=True

    )

    def request(self, url, params=None, data=None, query_type='query'):

        if data:

            response = self.requests.post(url, params=params, data=data)

        else:

            response = self.requests.get(url, params=params, data=data)

        if response.status_code // 100 != 2:

            raise ArcGISError('ArcGIS/%s returned status code %s' % (query_type, response.status_code))

        try:

            response = response.json()

        except ValueError:

            raise ArcGISError('ArcGIS/%s returned invalid JSON content: %r' % (query_type, response.content))

        if not isinstance(response, dict):

            raise ArcGISError('ArcGIS/%s not returned a dict: %r' % (query_type, response))

        if 'error' in response:

            err_desc = response['error'].get('message') or ('unknown ArcGIS/%s error' % query_type)

            raise ArcGISError(err_desc, data=response)

        return response

    def generate_token(self):

        if not self.token_username and not self.token_password:

            return None

        url = urlparse.urljoin(self.base_url, 'info')

        info = self.request(url, params={'f': 'json'}, query_type='token')

        token_url = info.get('authInfo', {}).get('tokenServicesUrl')

        if not token_url:

            raise ArcGISError('ArcGIS/token responded no authInfo/tokenServicesUrl in info: %r' % info)

        response = self.request(

            token_url,

            data={

                'username': self.token_username,

                'password': self.token_password,

                'client': 'referer',

                'referer': urlparse.urljoin(self.base_url, 'services'),

                'f': 'json',

            },

            query_type='token',

        )

        if 'token' not in response:

            raise ArcGISError('ArcGIS/token returned no token: %r' % response)

        return response['token']

        # add a token if applicable

        if 'token' not in params:

            token = self.generate_token()

            if token is not None:

                params['token'] = token

        infos = self.request(url, params=params)

INFO = '''{

    "authInfo": {

        "isTokenBasedSecurity": true,

        "tokenServicesUrl": "https://arcgis/portal/sharing/rest/generateToken"

    },

    "currentVersion": 10.81,

    "fullVersion": "10.8.1",

    "owningSystemUrl": "https://arcgis/portal",

    "secureSoapUrl": null,

    "soapUrl": "https://arcgis/arcgis/services"

}'''

TOKEN = '''{

    "expires": 1649761676673,

    "ssl": true,

    "token": "tok42"

}'''

def test_arcgis_with_token_query(app, arcgis):

    endpoint = tests.utils.generic_endpoint_url('arcgis', 'featureservice-query', slug=arcgis.slug)

    assert endpoint == '/arcgis/test/featureservice-query'

    # open access

    api = ApiUser.objects.create(username='all', keytype='', key='')

    obj_type = ContentType.objects.get_for_model(arcgis)

    AccessRight.objects.create(

        codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=arcgis.pk

    )

    arcgis.token_username = 'tokenuser'

    arcgis.token_password = 'tokenpass'

    arcgis.save()

    params = {'folder': 'fold', 'service': 'serv', 'layer': '42'}

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        with mock.patch('passerelle.utils.Request.post') as requests_post:

            requests_get.side_effect = [

                tests.utils.FakedResponse(content=INFO, status_code=200),

                tests.utils.FakedResponse(content=FEATURES, status_code=200),

            ]

            requests_post.side_effect = [

                tests.utils.FakedResponse(content=TOKEN, status_code=200),

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 2  # info + featureservice-query

            assert requests_post.call_count == 1  # token

            assert (

                requests_get.call_args[0][0]

                == 'https://arcgis.example.net/services/fold/serv/FeatureServer/42/query'

            )

            args = requests_get.call_args[1]['params']

            assert args['f'] == 'json'

            assert args['outFields'] == '*'

            assert args['token'] == 'tok42'  # token added in query

            assert 'data' in resp.json

            assert resp.json['err'] == 0

            assert len(resp.json['data']) == 2

            assert resp.json['data'][0]['id'] == '11'

            assert resp.json['data'][0]['text'] == '11'

            # bad info/token responses

            requests_get.reset_mock()

            requests_post.reset_mock()

            requests_get.side_effect = [

                tests.utils.FakedResponse(content='{}', status_code=200),

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 1

            assert requests_post.call_count == 0

            assert resp.json['err'] == 1

            assert resp.json['err_desc'].startswith(

                'ArcGIS/token responded no authInfo/tokenServicesUrl in info'

            )

            requests_get.reset_mock()

            requests_get.side_effect = [

                tests.utils.FakedResponse(content='CRASH', status_code=500),

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 1

            assert requests_post.call_count == 0

            assert resp.json['err'] == 1

            assert resp.json['err_desc'] == 'ArcGIS/token returned status code 500'

            requests_get.reset_mock()

            requests_get.side_effect = [

                tests.utils.FakedResponse(content='not json', status_code=200),

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 1

            assert requests_post.call_count == 0

            assert resp.json['err'] == 1

            assert resp.json['err_desc'].startswith('ArcGIS/token returned invalid JSON content:')

            requests_get.reset_mock()

            requests_get.side_effect = [

                tests.utils.FakedResponse(content='"not a dict"', status_code=200),

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 1

            assert requests_post.call_count == 0

            assert resp.json['err'] == 1

            assert resp.json['err_desc'].startswith('ArcGIS/token not returned a dict:')

            requests_get.reset_mock()

            requests_post.reset_mock()

            requests_get.side_effect = [

                tests.utils.FakedResponse(content=INFO, status_code=200),

            ]

            requests_post.side_effect = [

                tests.utils.FakedResponse(content="{}", status_code=200),  # no token

            ]

            resp = app.get(endpoint, params=params, status=200)

            assert requests_get.call_count == 1

            assert requests_post.call_count == 1

            assert resp.json['err'] == 1

            assert resp.json['err_desc'].startswith('ArcGIS/token returned no token:')