110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
# 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 unittest import mock
|
|
|
|
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
|
|
from passerelle.apps.ovh.models import OVHSMSGateway
|
|
from passerelle.base.models import AccessRight, ApiUser, Job
|
|
from tests.test_manager import login
|
|
|
|
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, freezer):
|
|
assert Job.objects.count() == 0
|
|
url = reverse('api-job', kwargs={'pk': 22})
|
|
freezer.move_to('2022-01-01 12:00')
|
|
|
|
# no job
|
|
resp = app.get(url, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'No job found matching the query'
|
|
|
|
job = connector.add_job('send_job', my_parameter='my_value')
|
|
assert Job.objects.count() == 1
|
|
url = reverse('api-job', kwargs={'pk': job.id})
|
|
|
|
# app disabled
|
|
with override_settings(PASSERELLE_APP_OVH_ENABLED=False):
|
|
resp = app.get(url, status=403)
|
|
|
|
# access without permission
|
|
resp = app.get(url, status=403)
|
|
|
|
# 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=403)
|
|
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
|
|
freezer.move_to('2022-01-01 12:01')
|
|
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
|
|
assert 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
|