52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
import utils
|
|
|
|
from passerelle.base.models import Job, SkipJob, ResourceLog
|
|
from .test_base_adresse import base_adresse, StreetModel
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_jobs(mocked_get, app, base_adresse):
|
|
filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.bz2')
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
|
|
job = base_adresse.add_job('update_streets_data')
|
|
assert job.status == 'registered'
|
|
|
|
base_adresse.jobs()
|
|
assert Job.objects.get(id=job.id).status == 'completed'
|
|
assert StreetModel.objects.count() == 3
|
|
|
|
StreetModel.objects.all().delete()
|
|
|
|
ResourceLog.objects.all().delete()
|
|
job = base_adresse.add_job('update_streets_data')
|
|
mocked_get.side_effect = Exception('hello')
|
|
base_adresse.jobs()
|
|
assert Job.objects.get(id=job.id).status == 'failed'
|
|
assert Job.objects.get(id=job.id).status_details == {'error_summary': 'Exception: hello'}
|
|
assert ResourceLog.objects.all().count() == 1
|
|
assert ResourceLog.objects.all()[0].message == 'error running update_streets_data job (hello)'
|
|
|
|
job = base_adresse.add_job('update_streets_data')
|
|
mocked_get.side_effect = SkipJob()
|
|
base_adresse.jobs()
|
|
assert Job.objects.get(id=job.id).status == 'registered'
|
|
|
|
# don't run jobs if connector is down
|
|
StreetModel.objects.all().delete()
|
|
with mock.patch('passerelle.apps.base_adresse.models.BaseAdresse.down') as down:
|
|
down.side_effect = lambda: True
|
|
mocked_get.return_value = utils.FakedResponse(content=open(filepath).read(), status_code=200)
|
|
job = base_adresse.add_job('update_streets_data')
|
|
assert job.status == 'registered'
|
|
|
|
base_adresse.jobs()
|
|
assert Job.objects.get(id=job.id).status == 'registered'
|