chrono/tests/api/test_applification_api.py

1070 lines
43 KiB
Python

import io
import json
import re
import tarfile
import pytest
from django.contrib.auth.models import Group
from django.contrib.contenttypes.models import ContentType
from chrono.agendas.models import Agenda, Category, Desk, EventsType, Resource, UnavailabilityCalendar
from chrono.apps.export_import.models import Application, ApplicationElement
from chrono.apps.snapshot.models import AgendaSnapshot
pytestmark = pytest.mark.django_db
def test_object_types(app, user, admin_user):
app.authorization = ('Basic', ('john.doe', 'password'))
app.get('/api/export-import/', status=403)
app.authorization = ('Basic', ('admin', 'admin'))
resp = app.get('/api/export-import/')
assert resp.json == {
'data': [
{
'id': 'agendas',
'text': 'Agendas',
'singular': 'Agenda',
'urls': {'list': 'http://testserver/api/export-import/agendas/'},
},
{
'id': 'agendas_categories',
'minor': True,
'singular': 'Category (agendas)',
'text': 'Categories (agendas)',
'urls': {'list': 'http://testserver/api/export-import/agendas_categories/'},
},
{
'id': 'events_types',
'minor': True,
'singular': 'Events type',
'text': 'Events types',
'urls': {'list': 'http://testserver/api/export-import/events_types/'},
},
{
'id': 'resources',
'minor': True,
'singular': 'Resource',
'text': 'Resources',
'urls': {'list': 'http://testserver/api/export-import/resources/'},
},
{
'id': 'unavailability_calendars',
'minor': True,
'singular': 'Unavailability calendar',
'text': 'Unavailability calendars',
'urls': {'list': 'http://testserver/api/export-import/unavailability_calendars/'},
},
{
'id': 'roles',
'minor': True,
'singular': 'Role',
'text': 'Roles',
'urls': {'list': 'http://testserver/api/export-import/roles/'},
},
]
}
def test_list(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
Agenda.objects.create(label='Event', slug='event', kind='events')
Category.objects.create(slug='cat', label='Category')
Resource.objects.create(slug='foo', label='Foo')
EventsType.objects.create(slug='foo', label='Foo')
UnavailabilityCalendar.objects.create(slug='foo', label='Foo')
group = Group.objects.create(name='group1')
resp = app.get('/api/export-import/agendas/')
assert resp.json == {
'data': [
{
'id': 'event',
'text': 'Event',
'type': 'agendas',
'urls': {
'export': 'http://testserver/api/export-import/agendas/event/',
'dependencies': 'http://testserver/api/export-import/agendas/event/dependencies/',
'redirect': 'http://testserver/api/export-import/agendas/event/redirect/',
},
},
{
'id': 'rdv',
'text': 'Rdv',
'type': 'agendas',
'urls': {
'export': 'http://testserver/api/export-import/agendas/rdv/',
'dependencies': 'http://testserver/api/export-import/agendas/rdv/dependencies/',
'redirect': 'http://testserver/api/export-import/agendas/rdv/redirect/',
},
},
]
}
resp = app.get('/api/export-import/agendas_categories/')
assert resp.json == {
'data': [
{
'id': 'cat',
'text': 'Category',
'type': 'agendas_categories',
'urls': {
'export': 'http://testserver/api/export-import/agendas_categories/cat/',
'dependencies': 'http://testserver/api/export-import/agendas_categories/cat/dependencies/',
'redirect': 'http://testserver/api/export-import/agendas_categories/cat/redirect/',
},
}
]
}
resp = app.get('/api/export-import/resources/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'resources',
'urls': {
'export': 'http://testserver/api/export-import/resources/foo/',
'dependencies': 'http://testserver/api/export-import/resources/foo/dependencies/',
'redirect': 'http://testserver/api/export-import/resources/foo/redirect/',
},
}
]
}
resp = app.get('/api/export-import/events_types/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'events_types',
'urls': {
'export': 'http://testserver/api/export-import/events_types/foo/',
'dependencies': 'http://testserver/api/export-import/events_types/foo/dependencies/',
'redirect': 'http://testserver/api/export-import/events_types/foo/redirect/',
},
}
]
}
resp = app.get('/api/export-import/unavailability_calendars/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'unavailability_calendars',
'urls': {
'export': 'http://testserver/api/export-import/unavailability_calendars/foo/',
'dependencies': 'http://testserver/api/export-import/unavailability_calendars/foo/dependencies/',
'redirect': 'http://testserver/api/export-import/unavailability_calendars/foo/redirect/',
},
}
]
}
resp = app.get('/api/export-import/roles/')
assert resp.json == {
'data': [{'id': group.pk, 'text': 'group1', 'type': 'roles', 'urls': {}, 'uuid': None}]
}
# unknown component type
app.get('/api/export-import/unknown/', status=404)
def test_export_agenda(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
group1 = Group.objects.create(name='group1')
group2 = Group.objects.create(name='group2')
Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings', edit_role=group1, view_role=group2)
resp = app.get('/api/export-import/agendas/rdv/')
assert resp.json['data']['label'] == 'Rdv'
assert resp.json['data']['permissions'] == {'view': 'group2', 'edit': 'group1'}
def test_export_minor_components(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
Category.objects.create(slug='cat', label='Category')
Resource.objects.create(slug='foo', label='Foo')
EventsType.objects.create(slug='foo', label='Foo')
UnavailabilityCalendar.objects.create(slug='foo', label='Foo')
resp = app.get('/api/export-import/agendas_categories/cat/')
assert resp.json['data']['label'] == 'Category'
resp = app.get('/api/export-import/resources/foo/')
assert resp.json['data']['label'] == 'Foo'
resp = app.get('/api/export-import/events_types/foo/')
assert resp.json['data']['label'] == 'Foo'
resp = app.get('/api/export-import/unavailability_calendars/foo/')
assert resp.json['data']['label'] == 'Foo'
# unknown component
app.get('/api/export-import/agendas/foo/', status=404)
# unknown component type
app.get('/api/export-import/unknown/foo/', status=404)
def test_agenda_dependencies_category(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
category = Category.objects.create(slug='cat', label='Category')
Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings', category=category)
resp = app.get('/api/export-import/agendas/rdv/dependencies/')
assert resp.json == {
'data': [
{
'id': 'cat',
'text': 'Category',
'type': 'agendas_categories',
'urls': {
'dependencies': 'http://testserver/api/export-import/agendas_categories/cat/dependencies/',
'export': 'http://testserver/api/export-import/agendas_categories/cat/',
'redirect': 'http://testserver/api/export-import/agendas_categories/cat/redirect/',
},
}
],
'err': 0,
}
def test_agenda_dependencies_resources(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
meetings_agenda = Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
meetings_agenda.resources.add(Resource.objects.create(slug='foo', label='Foo'))
resp = app.get('/api/export-import/agendas/rdv/dependencies/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'resources',
'urls': {
'dependencies': 'http://testserver/api/export-import/resources/foo/dependencies/',
'export': 'http://testserver/api/export-import/resources/foo/',
'redirect': 'http://testserver/api/export-import/resources/foo/redirect/',
},
}
],
'err': 0,
}
def test_agenda_dependencies_unavailability_calendars(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
meetings_agenda = Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
desk = Desk.objects.create(slug='foo', label='Foo', agenda=meetings_agenda)
unavailability_calendar = UnavailabilityCalendar.objects.create(slug='foo', label='Foo')
desk.unavailability_calendars.add(unavailability_calendar)
resp = app.get('/api/export-import/agendas/rdv/dependencies/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'unavailability_calendars',
'urls': {
'dependencies': 'http://testserver/api/export-import/unavailability_calendars/foo/dependencies/',
'export': 'http://testserver/api/export-import/unavailability_calendars/foo/',
'redirect': 'http://testserver/api/export-import/unavailability_calendars/foo/redirect/',
},
}
],
'err': 0,
}
events_agenda = Agenda.objects.create(label='Evt', slug='evt', kind='events')
desk = Desk.objects.create(agenda=events_agenda, slug='_exceptions_holder')
desk.unavailability_calendars.add(unavailability_calendar)
resp = app.get('/api/export-import/agendas/evt/dependencies/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'unavailability_calendars',
'urls': {
'dependencies': 'http://testserver/api/export-import/unavailability_calendars/foo/dependencies/',
'export': 'http://testserver/api/export-import/unavailability_calendars/foo/',
'redirect': 'http://testserver/api/export-import/unavailability_calendars/foo/redirect/',
},
}
],
'err': 0,
}
def test_agenda_dependencies_groups(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
group1 = Group.objects.create(name='group1')
group2 = Group.objects.create(name='group2')
Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings', edit_role=group1, view_role=group2)
resp = app.get('/api/export-import/agendas/rdv/dependencies/')
# note: with hobo.agent.common installed, 'groups' will contain group slugs,
# not group id
assert resp.json == {
'data': [
{'id': group2.id, 'text': group2.name, 'type': 'roles', 'urls': {}, 'uuid': None},
{'id': group1.id, 'text': group1.name, 'type': 'roles', 'urls': {}, 'uuid': None},
],
'err': 0,
}
def test_agenda_dependencies_virtual_agendas(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
rdv1 = Agenda.objects.create(label='Rdv1', slug='rdv1', kind='meetings')
rdv2 = Agenda.objects.create(label='Rdv2', slug='rdv2', kind='meetings')
virt = Agenda.objects.create(label='Virt', slug='virt', kind='virtual')
virt.real_agendas.add(rdv1)
virt.real_agendas.add(rdv2)
resp = app.get('/api/export-import/agendas/virt/dependencies/')
assert resp.json == {
'data': [
{
'id': 'rdv1',
'text': 'Rdv1',
'type': 'agendas',
'urls': {
'dependencies': 'http://testserver/api/export-import/agendas/rdv1/dependencies/',
'export': 'http://testserver/api/export-import/agendas/rdv1/',
'redirect': 'http://testserver/api/export-import/agendas/rdv1/redirect/',
},
},
{
'id': 'rdv2',
'text': 'Rdv2',
'type': 'agendas',
'urls': {
'dependencies': 'http://testserver/api/export-import/agendas/rdv2/dependencies/',
'export': 'http://testserver/api/export-import/agendas/rdv2/',
'redirect': 'http://testserver/api/export-import/agendas/rdv2/redirect/',
},
},
],
'err': 0,
}
def test_agenda_dependencies_events_type(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
events_type = EventsType.objects.create(slug='foo', label='Foo')
events_agenda = Agenda.objects.create(label='Evt', slug='evt', kind='events', events_type=events_type)
Desk.objects.create(agenda=events_agenda, slug='_exceptions_holder')
resp = app.get('/api/export-import/agendas/evt/dependencies/')
assert resp.json == {
'data': [
{
'id': 'foo',
'text': 'Foo',
'type': 'events_types',
'urls': {
'dependencies': 'http://testserver/api/export-import/events_types/foo/dependencies/',
'export': 'http://testserver/api/export-import/events_types/foo/',
'redirect': 'http://testserver/api/export-import/events_types/foo/redirect/',
},
}
],
'err': 0,
}
def test_unknown_compoment_dependencies(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
app.get('/api/export-import/agendas/foo/dependencies/', status=404)
def test_unknown_compoment_type_dependencies(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
app.get('/api/export-import/unknown/foo/dependencies/', status=404)
def test_redirect(app):
agenda = Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
category = Category.objects.create(slug='cat', label='Category')
resource = Resource.objects.create(slug='foo', label='Foo')
events_type = EventsType.objects.create(slug='foo', label='Foo')
unavailability_calendar = UnavailabilityCalendar.objects.create(slug='foo', label='Foo')
redirect_url = f'/api/export-import/agendas/{agenda.slug}/redirect/'
resp = app.get(redirect_url, status=302)
assert resp.location == f'/manage/agendas/{agenda.pk}/'
resp = app.get(redirect_url + '?compare', status=302)
assert resp.location == f'/manage/agendas/{agenda.pk}/'
resp = app.get(redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302)
assert (
resp.location
== f'/manage/agendas/{agenda.pk}/history/compare/?version1=bar&version2=bar&application=foo'
)
redirect_url = f'/api/export-import/agendas_categories/{category.slug}/redirect/'
resp = app.get(redirect_url, status=302)
assert resp.location == '/manage/categories/'
resp = app.get(redirect_url + '?compare', status=302)
assert resp.location == '/manage/categories/'
resp = app.get(redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302)
assert (
resp.location
== f'/manage/category/{category.pk}/history/compare/?version1=bar&version2=bar&application=foo'
)
redirect_url = f'/api/export-import/resources/{resource.slug}/redirect/'
resp = app.get(redirect_url, status=302)
assert resp.location == f'/manage/resource/{resource.pk}/'
resp = app.get(redirect_url + '?compare', status=302)
assert resp.location == f'/manage/resource/{resource.pk}/'
resp = app.get(redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302)
assert (
resp.location
== f'/manage/resource/{resource.pk}/history/compare/?version1=bar&version2=bar&application=foo'
)
redirect_url = f'/api/export-import/events_types/{events_type.slug}/redirect/'
resp = app.get(redirect_url, status=302)
assert resp.location == '/manage/events-types/'
resp = app.get(redirect_url + '?compare', status=302)
assert resp.location == '/manage/events-types/'
resp = app.get(redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302)
assert (
resp.location
== f'/manage/events-type/{events_type.pk}/history/compare/?version1=bar&version2=bar&application=foo'
)
redirect_url = f'/api/export-import/unavailability_calendars/{unavailability_calendar.slug}/redirect/'
resp = app.get(redirect_url, status=302)
assert resp.location == f'/manage/unavailability-calendar/{unavailability_calendar.pk}/'
resp = app.get(redirect_url + '?compare', status=302)
assert resp.location == f'/manage/unavailability-calendar/{unavailability_calendar.pk}/'
resp = app.get(redirect_url + '?compare&version1=bar&version2=bar&application=foo', status=302)
assert (
resp.location
== f'/manage/unavailability-calendar/{unavailability_calendar.pk}/history/compare/?version1=bar&version2=bar&application=foo'
)
# unknown component type
app.get('/api/export-import/unknown/foo/redirect/', status=404)
def create_bundle(app, admin_user, visible=True, version_number='42.0'):
app.authorization = ('Basic', ('admin', 'admin'))
group, _ = Group.objects.get_or_create(name='plop')
category, _ = Category.objects.get_or_create(slug='foo', label='Foo')
meetings_agenda, _ = Agenda.objects.get_or_create(
slug='rdv', label='Rdv', kind='meetings', edit_role=group, category=category
)
resource, _ = Resource.objects.get_or_create(slug='foo', label='Foo')
meetings_agenda.resources.add(resource)
desk, _ = Desk.objects.get_or_create(slug='foo', label='Foo', agenda=meetings_agenda)
unavailability_calendar, _ = UnavailabilityCalendar.objects.get_or_create(slug='foo', label='Foo')
desk.unavailability_calendars.add(unavailability_calendar)
events_type, _ = EventsType.objects.get_or_create(slug='foo', label='Foo')
events_agenda, _ = Agenda.objects.get_or_create(
label='Evt', slug='evt', kind='events', events_type=events_type
)
Desk.objects.get_or_create(agenda=events_agenda, slug='_exceptions_holder')
sub_agenda, _ = Agenda.objects.get_or_create(slug='sub', label='Sub', kind='meetings')
virtual_agenda, _ = Agenda.objects.get_or_create(label='Virt', slug='virt', kind='virtual')
virtual_agenda.real_agendas.add(sub_agenda)
components = [
(meetings_agenda, False),
(category, True),
(resource, True),
(unavailability_calendar, True),
(events_agenda, False),
(events_type, True),
(virtual_agenda, False),
(sub_agenda, True),
]
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
manifest_json = {
'application': 'Test',
'slug': 'test',
'icon': 'foo.png',
'description': 'Foo Bar',
'documentation_url': 'http://foo.bar',
'visible': visible,
'version_number': version_number,
'version_notes': 'foo bar blah',
'elements': [],
}
for component, auto_dependency in components:
manifest_json['elements'].append(
{
'type': component.application_component_type,
'slug': component.slug,
'name': component.label,
'auto-dependency': auto_dependency,
}
)
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
tarinfo = tarfile.TarInfo('manifest.json')
tarinfo.size = len(manifest_fd.getvalue())
tar.addfile(tarinfo, fileobj=manifest_fd)
icon_fd = io.BytesIO(
b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABAQAAAAA3bvkkAAAACklEQVQI12NoAAAAggCB3UNq9AAAAABJRU5ErkJggg=='
)
tarinfo = tarfile.TarInfo('foo.png')
tarinfo.size = len(icon_fd.getvalue())
tar.addfile(tarinfo, fileobj=icon_fd)
for component, _ in components:
component_export = app.get(
'/api/export-import/%s/%s/' % (component.application_component_type, component.slug)
).content
tarinfo = tarfile.TarInfo('%s/%s' % (component.application_component_type, component.slug))
tarinfo.size = len(component_export)
tar.addfile(tarinfo, fileobj=io.BytesIO(component_export))
bundle = tar_io.getvalue()
return bundle
@pytest.fixture
def bundle(app, user):
return create_bundle(app, user)
def test_bundle_import(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
bundles = []
for version_number in ['42.0', '42.1']:
bundles.append(create_bundle(app, admin_user, version_number=version_number))
Agenda.objects.all().delete()
Category.objects.all().delete()
Resource.objects.all().delete()
EventsType.objects.all().delete()
UnavailabilityCalendar.objects.all().delete()
resp = app.post('/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert Agenda.objects.all().count() == 4
assert resp.json['err'] == 0
assert Application.objects.count() == 1
application = Application.objects.latest('pk')
assert application.slug == 'test'
assert application.name == 'Test'
assert application.description == 'Foo Bar'
assert application.documentation_url == 'http://foo.bar'
assert application.version_number == '42.0'
assert application.version_notes == 'foo bar blah'
assert re.match(r'applications/icons/foo(_\w+)?.png', application.icon.name)
assert application.editable is False
assert application.visible is True
assert ApplicationElement.objects.count() == 8
for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
for instance in model.objects.all():
last_snapshot = model.get_snapshot_model().objects.filter(instance=instance).latest('pk')
assert last_snapshot.comment == 'Application (Test)'
assert last_snapshot.application_slug == 'test'
assert last_snapshot.application_version == '42.0'
# check editable flag is kept on install
application.editable = True
application.save()
# create link to element not present in manifest: it should be unlinked
last_agenda = Agenda.objects.latest('pk')
ApplicationElement.objects.create(
application=application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=last_agenda.pk + 1,
)
# check update
resp = app.post('/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', bundles[1])])
assert Agenda.objects.all().count() == 4
assert resp.json['err'] == 0
assert Application.objects.count() == 1
application = Application.objects.latest('pk')
assert application.editable is False
assert ApplicationElement.objects.count() == 8
assert (
ApplicationElement.objects.filter(
application=application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=last_agenda.pk + 1,
).exists()
is False
)
for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
for instance in model.objects.all():
last_snapshot = model.get_snapshot_model().objects.filter(instance=instance).latest('pk')
assert last_snapshot.comment == 'Application (Test)'
assert last_snapshot.application_slug == 'test'
assert last_snapshot.application_version == '42.1'
# bad file format
resp = app.post(
'/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', b'garbage')], status=400
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file'
# missing manifest
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
tarinfo = tarfile.TarInfo('foo.json')
tarinfo.size = len(foo_fd.getvalue())
tar.addfile(tarinfo, fileobj=foo_fd)
resp = app.post(
'/api/export-import/bundle-import/',
upload_files=[('bundle', 'bundle.tar', tar_io.getvalue())],
status=400,
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'
# missing component
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
manifest_json = {
'application': 'Test',
'slug': 'test',
'elements': [{'type': 'agendas', 'slug': 'foo', 'name': 'foo'}],
}
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
tarinfo = tarfile.TarInfo('manifest.json')
tarinfo.size = len(manifest_fd.getvalue())
tar.addfile(tarinfo, fileobj=manifest_fd)
resp = app.post(
'/api/export-import/bundle-import/',
upload_files=[('bundle', 'bundle.tar', tar_io.getvalue())],
status=400,
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file, missing component agendas/foo'
def test_bundle_declare(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
bundle = create_bundle(app, admin_user, visible=False)
resp = app.post('/api/export-import/bundle-declare/', upload_files=[('bundle', 'bundle.tar', bundle)])
assert Agenda.objects.all().count() == 4
assert resp.json['err'] == 0
assert Application.objects.count() == 1
application = Application.objects.latest('pk')
assert application.slug == 'test'
assert application.name == 'Test'
assert application.description == 'Foo Bar'
assert application.documentation_url == 'http://foo.bar'
assert application.version_number == '42.0'
assert application.version_notes == 'foo bar blah'
assert re.match(r'applications/icons/foo(_\w+)?.png', application.icon.name)
assert application.editable is True
assert application.visible is False
assert ApplicationElement.objects.count() == 8
bundle = create_bundle(app, admin_user, visible=True)
# create link to element not present in manifest: it should be unlinked
last_page = Agenda.objects.latest('pk')
ApplicationElement.objects.create(
application=application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=last_page.pk + 1,
)
# and remove agendas to have unknown references in manifest
Agenda.objects.all().delete()
resp = app.post('/api/export-import/bundle-declare/', upload_files=[('bundle', 'bundle.tar', bundle)])
assert Application.objects.count() == 1
application = Application.objects.latest('pk')
assert application.visible is True
assert ApplicationElement.objects.count() == 4 # category, events_type, unavailability_calendar, resource
# bad file format
resp = app.post(
'/api/export-import/bundle-declare/', upload_files=[('bundle', 'bundle.tar', b'garbage')], status=400
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file'
# missing manifest
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
tarinfo = tarfile.TarInfo('foo.json')
tarinfo.size = len(foo_fd.getvalue())
tar.addfile(tarinfo, fileobj=foo_fd)
resp = app.post(
'/api/export-import/bundle-declare/',
upload_files=[('bundle', 'bundle.tar', tar_io.getvalue())],
status=400,
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'
# missing component
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
manifest_json = {
'application': 'Test',
'slug': 'test',
'elements': [{'type': 'agendas', 'slug': 'foo', 'name': 'foo'}],
}
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
tarinfo = tarfile.TarInfo('manifest.json')
tarinfo.size = len(manifest_fd.getvalue())
tar.addfile(tarinfo, fileobj=manifest_fd)
resp = app.post(
'/api/export-import/bundle-declare/',
upload_files=[('bundle', 'bundle.tar', tar_io.getvalue())],
status=400,
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file, missing component agendas/foo'
def test_bundle_unlink(app, admin_user, bundle):
app.authorization = ('Basic', ('admin', 'admin'))
application = Application.objects.create(
name='Test',
slug='test',
version_number='42.0',
)
other_application = Application.objects.create(
name='Other Test',
slug='other-test',
version_number='42.0',
)
agenda = Agenda.objects.latest('pk')
ApplicationElement.objects.create(
application=application,
content_object=agenda,
)
ApplicationElement.objects.create(
application=application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=agenda.pk + 1,
)
ApplicationElement.objects.create(
application=other_application,
content_object=agenda,
)
ApplicationElement.objects.create(
application=other_application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=agenda.pk + 1,
)
assert Application.objects.count() == 2
assert ApplicationElement.objects.count() == 4
app.post('/api/export-import/unlink/', {'application': 'test'})
assert Application.objects.count() == 1
assert ApplicationElement.objects.count() == 2
assert ApplicationElement.objects.filter(
application=other_application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=agenda.pk,
).exists()
assert ApplicationElement.objects.filter(
application=other_application,
content_type=ContentType.objects.get_for_model(Agenda),
object_id=agenda.pk + 1,
).exists()
# again
app.post('/api/export-import/unlink/', {'application': 'test'})
assert Application.objects.count() == 1
assert ApplicationElement.objects.count() == 2
def test_bundle_check(app, admin_user):
app.authorization = ('Basic', ('admin', 'admin'))
bundles = []
for version_number in ['42.0', '42.1']:
bundles.append(create_bundle(app, admin_user, version_number=version_number))
Agenda.objects.all().delete()
Category.objects.all().delete()
Resource.objects.all().delete()
EventsType.objects.all().delete()
UnavailabilityCalendar.objects.all().delete()
incomplete_bundles = []
for manifest_json in [{'slug': 'test'}, {'version_number': '1.0'}]:
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
manifest_fd = io.BytesIO(json.dumps(manifest_json, indent=2).encode())
tarinfo = tarfile.TarInfo('manifest.json')
tarinfo.size = len(manifest_fd.getvalue())
tar.addfile(tarinfo, fileobj=manifest_fd)
incomplete_bundles.append(tar_io.getvalue())
# incorrect bundles, missing information
resp = app.post(
'/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', incomplete_bundles[0])]
)
assert resp.json == {'data': {}}
resp = app.post(
'/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', incomplete_bundles[1])]
)
assert resp.json == {'data': {}}
# not yet imported
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert resp.json == {
'data': {
'differences': [],
'no_history_elements': [],
'unknown_elements': [
{'slug': 'rdv', 'type': 'agendas'},
{'slug': 'foo', 'type': 'categories'},
{'slug': 'foo', 'type': 'resources'},
{'slug': 'foo', 'type': 'unavailability_calendars'},
{'slug': 'evt', 'type': 'agendas'},
{'slug': 'foo', 'type': 'events_types'},
{'slug': 'virt', 'type': 'agendas'},
{'slug': 'sub', 'type': 'agendas'},
],
'legacy_elements': [],
}
}
# import bundle
resp = app.post('/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert Application.objects.count() == 1
assert ApplicationElement.objects.count() == 8
# remove application links
Application.objects.all().delete()
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert resp.json == {
'data': {
'differences': [],
'no_history_elements': [],
'unknown_elements': [],
'legacy_elements': [
{
'slug': 'rdv',
'text': 'Rdv',
'type': 'agendas',
'url': '/api/export-import/agendas/rdv/redirect/',
},
{
'slug': 'foo',
'text': 'Foo',
'type': 'agendas_categories',
'url': '/api/export-import/agendas_categories/foo/redirect/',
},
{
'slug': 'foo',
'text': 'Foo',
'type': 'resources',
'url': '/api/export-import/resources/foo/redirect/',
},
{
'slug': 'foo',
'text': 'Foo',
'type': 'unavailability_calendars',
'url': '/api/export-import/unavailability_calendars/foo/redirect/',
},
{
'slug': 'evt',
'text': 'Evt',
'type': 'agendas',
'url': '/api/export-import/agendas/evt/redirect/',
},
{
'slug': 'foo',
'text': 'Foo',
'type': 'events_types',
'url': '/api/export-import/events_types/foo/redirect/',
},
{
'slug': 'virt',
'text': 'Virt',
'type': 'agendas',
'url': '/api/export-import/agendas/virt/redirect/',
},
{
'slug': 'sub',
'text': 'Sub',
'type': 'agendas',
'url': '/api/export-import/agendas/sub/redirect/',
},
],
}
}
# import bundle again, recreate links
resp = app.post('/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert Application.objects.count() == 1
assert ApplicationElement.objects.count() == 8
# no changes since last import
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# add local changes
snapshots = {}
for model in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]:
for instance in model.objects.all():
old_snapshot = model.get_snapshot_model().objects.filter(instance=instance).latest('pk')
instance.take_snapshot(comment='local changes')
new_snapshot = model.get_snapshot_model().objects.filter(instance=instance).latest('pk')
assert new_snapshot.pk > old_snapshot.pk
snapshots[f'{instance.application_component_type}:{instance.slug}'] = (
instance.pk,
old_snapshot.pk,
new_snapshot.pk,
)
# and check
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[0])])
assert resp.json == {
'data': {
'differences': [
{
'slug': 'rdv',
'type': 'agendas',
'url': 'http://testserver/manage/agendas/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['agendas:rdv'][0],
snapshots['agendas:rdv'][1],
snapshots['agendas:rdv'][2],
),
},
{
'slug': 'foo',
'type': 'agendas_categories',
'url': 'http://testserver/manage/category/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['agendas_categories:foo'][0],
snapshots['agendas_categories:foo'][1],
snapshots['agendas_categories:foo'][2],
),
},
{
'slug': 'foo',
'type': 'resources',
'url': 'http://testserver/manage/resource/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['resources:foo'][0],
snapshots['resources:foo'][1],
snapshots['resources:foo'][2],
),
},
{
'slug': 'foo',
'type': 'unavailability_calendars',
'url': 'http://testserver/manage/unavailability-calendar/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['unavailability_calendars:foo'][0],
snapshots['unavailability_calendars:foo'][1],
snapshots['unavailability_calendars:foo'][2],
),
},
{
'slug': 'evt',
'type': 'agendas',
'url': 'http://testserver/manage/agendas/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['agendas:evt'][0],
snapshots['agendas:evt'][1],
snapshots['agendas:evt'][2],
),
},
{
'slug': 'foo',
'type': 'events_types',
'url': 'http://testserver/manage/events-type/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['events_types:foo'][0],
snapshots['events_types:foo'][1],
snapshots['events_types:foo'][2],
),
},
{
'slug': 'virt',
'type': 'agendas',
'url': 'http://testserver/manage/agendas/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['agendas:virt'][0],
snapshots['agendas:virt'][1],
snapshots['agendas:virt'][2],
),
},
{
'slug': 'sub',
'type': 'agendas',
'url': 'http://testserver/manage/agendas/%s/history/compare/?version1=%s&version2=%s'
% (
snapshots['agendas:sub'][0],
snapshots['agendas:sub'][1],
snapshots['agendas:sub'][2],
),
},
],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# update bundle
resp = app.post('/api/export-import/bundle-import/', upload_files=[('bundle', 'bundle.tar', bundles[1])])
# and check
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[1])])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [],
'legacy_elements': [],
}
}
# snapshots without application info
AgendaSnapshot.objects.update(application_slug=None, application_version=None)
resp = app.post('/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', bundles[1])])
assert resp.json == {
'data': {
'differences': [],
'unknown_elements': [],
'no_history_elements': [
{'slug': 'rdv', 'type': 'agendas'},
{'slug': 'evt', 'type': 'agendas'},
{'slug': 'virt', 'type': 'agendas'},
{'slug': 'sub', 'type': 'agendas'},
],
'legacy_elements': [],
}
}
# bad file format
resp = app.post(
'/api/export-import/bundle-check/', upload_files=[('bundle', 'bundle.tar', b'garbage')], status=400
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file'
# missing manifest
tar_io = io.BytesIO()
with tarfile.open(mode='w', fileobj=tar_io) as tar:
foo_fd = io.BytesIO(json.dumps({'foo': 'bar'}, indent=2).encode())
tarinfo = tarfile.TarInfo('foo.json')
tarinfo.size = len(foo_fd.getvalue())
tar.addfile(tarinfo, fileobj=foo_fd)
resp = app.post(
'/api/export-import/bundle-check/',
upload_files=[('bundle', 'bundle.tar', tar_io.getvalue())],
status=400,
)
assert resp.json['err']
assert resp.json['err_desc'] == 'Invalid tar file, missing manifest'