150 lines
4.7 KiB
Python
150 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import datetime
|
|
import pytest
|
|
|
|
from django.utils.timezone import now
|
|
|
|
from zoo.models import EntitySchema, Entity, Job
|
|
|
|
|
|
@pytest.fixture
|
|
def person_schema(db):
|
|
schema = EntitySchema.objects.create(
|
|
name='person',
|
|
schema={
|
|
'type': 'object',
|
|
'properties': {
|
|
'first_name': {
|
|
'type': 'string',
|
|
},
|
|
'last_name': {
|
|
'type': 'string',
|
|
},
|
|
'address': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'street': {
|
|
'type': 'string',
|
|
},
|
|
}
|
|
},
|
|
'bad_type_property': {
|
|
'type': 'not string type'
|
|
}
|
|
}
|
|
})
|
|
return schema
|
|
|
|
|
|
@pytest.fixture
|
|
def person_entity(person_schema):
|
|
entity = Entity.objects.create(
|
|
schema=person_schema,
|
|
meta={},
|
|
content={
|
|
'first_name': 'Leon',
|
|
'last_name': 'Blum',
|
|
'address': {
|
|
'street': 'Rue du Château',
|
|
}
|
|
})
|
|
return entity
|
|
|
|
|
|
def test_new_schema(person_schema, person_entity):
|
|
qs = Entity.objects.content_search(person_schema, address__street='chateau')
|
|
assert qs.count() == 1
|
|
|
|
|
|
class ActionExample(object):
|
|
def __init__(self, tries=2, content=None):
|
|
self.content = content or {
|
|
'tries': tries,
|
|
'done': False,
|
|
}
|
|
|
|
def do(self, **kwargs):
|
|
if self.content['tries'] == 0:
|
|
self.content['done'] = True
|
|
return Job.STATE_SUCCESS
|
|
self.content['tries'] -= 1
|
|
return Job.STATE_ERROR
|
|
|
|
def to_json(self):
|
|
return self.content
|
|
|
|
@classmethod
|
|
def from_json(cls, content):
|
|
return cls(content=content)
|
|
|
|
|
|
def test_jobs(db):
|
|
assert Job.create(ActionExample()).state == Job.STATE_ERROR
|
|
assert Job.objects.count() == 1
|
|
assert Job.objects.get().content['done'] is False
|
|
assert Job.objects.get().content['tries'] == 1
|
|
assert Job.objects.get().state == Job.STATE_ERROR
|
|
# too fast nothing has been done
|
|
Job.redo()
|
|
assert Job.objects.count() == 1
|
|
assert Job.objects.get().content['done'] is False
|
|
assert Job.objects.get().content['tries'] == 1
|
|
assert Job.objects.get().state == Job.STATE_ERROR
|
|
# first retry
|
|
Job.redo(timestamp=now() + datetime.timedelta(seconds=1))
|
|
assert Job.objects.count() == 1
|
|
assert Job.objects.get().content['done'] is False
|
|
assert Job.objects.get().content['tries'] == 0
|
|
assert Job.objects.get().state == Job.STATE_ERROR
|
|
# second retry
|
|
Job.redo(timestamp=now() + datetime.timedelta(seconds=1))
|
|
assert Job.objects.count() == 1
|
|
assert Job.objects.get().content['done'] is True
|
|
assert Job.objects.get().content['tries'] == 0
|
|
assert Job.objects.get().state == Job.STATE_SUCCESS
|
|
|
|
|
|
def test_demo_schemas_view(app, person_schema):
|
|
resp = app.get('/demo/schemas/', status=200)
|
|
assert resp.html.find('a').text == 'person'
|
|
assert resp.html.find('a')['href'] == '/demo/schemas/person/'
|
|
|
|
|
|
def test_demo_schema_view(app, person_schema):
|
|
resp = app.get('/demo/schemas/person/', {'page': 'not a number'}, status=200)
|
|
assert resp.html.find('h1').text == 'person'
|
|
assert resp.html.find('label', {'for': 'id_first_name'}).name
|
|
assert resp.html.find('label', {'for': 'id_last_name'}).name
|
|
assert resp.html.find('label', {'for': 'id_address__street'}).name
|
|
assert not resp.html.find('label', {'for': 'bad_type_property'})
|
|
assert resp.html.find('input', {'type': 'submit'})['value'] == 'Recherche'
|
|
assert u'Aucune entité trouvée' in resp.text
|
|
|
|
|
|
def test_demo_search_on_schema_view(app, person_entity):
|
|
resp = app.get('/demo/schemas/person/', {'first_name': 'Leo'}, status=200)
|
|
assert u'Aucune entité trouvée' not in resp.text
|
|
assert 'Blum' in resp.html.pre.text
|
|
|
|
resp = app.get('/demo/schemas/person/', {'limit': 'not a float'}, status=200)
|
|
assert u'Aucune entité trouvée' in resp.text
|
|
|
|
|
|
def test_admin_zoo_data_entity_inactive_view(app, admin, nanterre_classic_family):
|
|
url = '/admin/zoo_data/entity/inactive/'
|
|
app.set_user(admin.username)
|
|
resp = app.get(url, status=200)
|
|
assert [x.text for x in resp.html.find_all('h1')] == [
|
|
'Administration technique du RSU', u'Fiches inactives à supprimer']
|
|
|
|
resp = app.get(url+'?recompute', status=302)
|
|
assert resp.location == url
|
|
|
|
# unavailable post query
|
|
resp = app.post(url+'?delete', status=403)
|
|
|
|
resp = app.get(url+'?csv', status=200)
|
|
assert resp.content_type == 'text/csv'
|
|
assert resp.text == 'id,prenoms,nom_d_usage,nom_de_naissance,date_de_naissance,statut_legal,age\r\n'
|