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 .api_views import JobView

api_urlpatterns = [

    url(r'jobs/(?P<pk>[\w,-]+)/$', JobView.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.http import JsonResponse

from django.views.generic import DetailView

from passerelle.base.models import Job

class JobView(DetailView):

    model = Job

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

        job = self.get_object()

        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 api_urlpatterns

    url('^api/',

        decorated_includes(manager_required, include(api_urlpatterns))),

# 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 passerelle.apps.orange.models import OrangeSMSGateway

from passerelle.base.models import Job

from test_manager import login, admin_user

pytestmark = pytest.mark.django_db

@pytest.fixture

def connector(db):

    return OrangeSMSGateway.objects.create(

        slug='my_connector',

        username='jdoe',

        password='secret',

        groupname='group2'

    )

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

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

    assert Job.objects.count() == 0

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

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

    assert resp.location == '/login/?next=%s' % url

    app = login(app)

    assert app.get(url, status=404)

    # registered job

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

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

    assert Job.objects.count() == 1

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

    assert not resp.json['err']

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

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

    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