Produits Entr'ouvert » Passerelle

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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.conf.urls import url

from .views import JobDetailView

urlpatterns = [

    url(r'jobs/(?P<pk>[\w,-]+)/$', JobDetailView.as_view(), name='api-job'),

]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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.apps import apps

from django.conf import settings

def get_related_app(connector):

    for app in apps.get_app_configs():

        if not hasattr(app, 'get_connector_model'):

            continue

        if app.get_connector_model() == connector.__class__:

            return app

    return None

def app_enabled(app_label):

    return getattr(settings, 'PASSERELLE_APP_%s_ENABLED' % app_label.upper(), True)

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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.http import Http404

from django.http import JsonResponse

from django.views.generic import DetailView

from passerelle.api.utils import get_related_app, app_enabled

from passerelle.base.models import Job

from passerelle.utils import is_authorized

class JobDetailView(DetailView):

    model = Job

    def error(self, message):

        return JsonResponse({'err': 1, 'err_desc': message})

    def get(self, *args, **kwargs):

        try:

            job = self.get_object()

        except Http404 as exc:

            return self.error(str(exc))

        connector = job.resource

        app = get_related_app(connector)

        if not app:

            return self.error('Cannot retrieve job resource connector')

        if not app_enabled(app.label):

            return self.error('Please enable %s' % app.label)

        if not is_authorized(self.request, connector, 'can_access'):

            return self.error('Permission denied')

        data = {

            'id': job.id,

            'resource': job.resource.__class__.__name__,

            'parameters': job.parameters,

            'status': job.status,

            'status_details': job.status_details,

            'update_timestamp': job.update_timestamp,

            'done_timestamp': job.done_timestamp,

        }

        return JsonResponse({'err': 0, 'data': data})

from .api.urls import urlpatterns as api_urls

    url('^api/', include(api_urls)),

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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/>.

import mock

import pytest

from django.core.urlresolvers import reverse

from django.contrib.contenttypes.models import ContentType

from django.test import override_settings

from passerelle.apps.ovh.models import OVHSMSGateway

from passerelle.base.models import ApiUser, AccessRight, Job

from test_manager import login, simple_user, admin_user

pytestmark = pytest.mark.django_db

API_KEY = '1234'

@pytest.fixture

def connector(db):

    connector = OVHSMSGateway.objects.create(

        slug='my_connector',

    )

    apiuser = ApiUser.objects.create(username='me', keytype='API', key=API_KEY)

    obj_type = ContentType.objects.get_for_model(OVHSMSGateway)

    AccessRight.objects.create(codename='can_access', apiuser=apiuser,

                               resource_type=obj_type, resource_pk=connector.pk)

    return connector

@mock.patch('passerelle.sms.models.SMSResource.send_job')

def test_api_jobs(mocked_send_job, app, connector, simple_user, admin_user):

    assert Job.objects.count() == 0

    url = reverse('api-job', kwargs={'pk': 22})

    # no job

    resp = app.get(url, status=200)

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

    assert resp.json['err_desc'] == 'No job found matching the query'

    # no job resource

    job = connector.add_job('send_job', my_parameter='my_value')

    assert Job.objects.count() == 1

    url = reverse('api-job', kwargs={'pk': job.id})

    with mock.patch('passerelle.api.views.get_related_app', return_value=None):

        resp = app.get(url, status=200)

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

    assert resp.json['err_desc'] == 'Cannot retrieve job resource connector'

    # app disabled

    with override_settings(PASSERELLE_APP_OVH_ENABLED=False):

        resp = app.get(url, status=200)

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

    assert resp.json['err_desc'] == 'Please enable ovh'

    # access without permission

    resp = app.get(url, status=200)

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

    assert resp.json['err_desc'] == 'Permission denied'

    # apiuser access

    resp = app.get(url, params={'apikey': API_KEY}, status=200)

    assert not resp.json['err']

    # logged user access

    app = login(app, simple_user.username, password='user')

    resp = app.get(url, status=200)

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

    assert resp.json['err_desc'] == 'Permission denied'

    app = login(app, admin_user.username, password='admin')

    resp = app.get(url, status=200)

    assert not resp.json['err']

    # registered job

    assert resp.json['data']['id'] == job.id

    assert resp.json['data']['resource'] == 'OVHSMSGateway'

    assert resp.json['data']['parameters'] == {'my_parameter': 'my_value'}

    assert resp.json['data']['status'] == 'registered'

    assert resp.json['data']['done_timestamp'] is None

    update_timestamp1 = resp.json['data']['update_timestamp']

    # completed job

    connector.jobs()

    assert mocked_send_job.call_args == mock.call(my_parameter='my_value')

    resp = app.get(url, status=200)

    assert not resp.json['err']

    assert resp.json['data']['id'] == job.id

    assert resp.json['data']['status'] == 'completed'

    assert resp.json['data']['done_timestamp'] is not None

    resp.json['data']['update_timestamp'] < update_timestamp1

    # failed job

    job = connector.add_job('send_job')

    assert Job.objects.count() == 2

    mocked_send_job.side_effect = Exception('my error message')

    connector.jobs()

    url = reverse('api-job', kwargs={'pk': job.id})

    resp = app.get(url, status=200)

    assert not resp.json['err']

    assert resp.json['data']['id'] == job.id

    assert resp.json['data']['status'] == 'failed'

    assert resp.json['data']['status_details'] == {'error_summary': 'Exception: my error message'}

    assert resp.json['data']['done_timestamp'] is not None