api: add module with applification API (#82198)
gitea/chrono/pipeline/head This commit looks good
Details
gitea/chrono/pipeline/head This commit looks good
Details
This commit is contained in:
parent
a25a8e6ef1
commit
cba5520541
|
@ -438,6 +438,12 @@ class Agenda(models.Model):
|
|||
raise ValueError()
|
||||
return gcd
|
||||
|
||||
def get_dependencies(self):
|
||||
yield self.view_role
|
||||
yield self.edit_role
|
||||
if self.kind == 'virtual':
|
||||
yield from self.real_agendas.all()
|
||||
|
||||
def export_json(self):
|
||||
agenda = {
|
||||
'label': self.label,
|
||||
|
|
|
@ -0,0 +1,219 @@
|
|||
# chrono - content management system
|
||||
# Copyright (C) 2016-2023 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 io
|
||||
import json
|
||||
import tarfile
|
||||
|
||||
from django.contrib.auth.models import Group
|
||||
from django.shortcuts import get_object_or_404, redirect
|
||||
from django.urls import reverse
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from rest_framework import permissions
|
||||
from rest_framework.generics import GenericAPIView
|
||||
from rest_framework.response import Response
|
||||
|
||||
from chrono.agendas.models import Agenda
|
||||
from chrono.apps.export_import.models import Application, ApplicationElement
|
||||
from chrono.manager.utils import import_site
|
||||
|
||||
|
||||
class Index(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
response = [
|
||||
{
|
||||
'id': 'agendas',
|
||||
'text': _('Agendas'),
|
||||
'singular': _('Agenda'),
|
||||
'urls': {
|
||||
'list': request.build_absolute_uri(reverse('api-export-import-agendas-list')),
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
return Response({'data': response})
|
||||
|
||||
|
||||
index = Index.as_view()
|
||||
|
||||
|
||||
def get_agenda_bundle_entry(request, agenda, order):
|
||||
return {
|
||||
'id': str(agenda.slug),
|
||||
'text': agenda.label,
|
||||
'type': 'agendas',
|
||||
'urls': {
|
||||
'export': request.build_absolute_uri(
|
||||
reverse('api-export-import-agenda-export', kwargs={'slug': str(agenda.slug)})
|
||||
),
|
||||
'dependencies': request.build_absolute_uri(
|
||||
reverse('api-export-import-agenda-dependencies', kwargs={'slug': str(agenda.slug)})
|
||||
),
|
||||
'redirect': request.build_absolute_uri(
|
||||
reverse('api-export-import-agenda-redirect', kwargs={'slug': str(agenda.slug)})
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class ListAgendas(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def get(self, request, *args, **kwargs):
|
||||
response = [
|
||||
get_agenda_bundle_entry(request, x, i) for i, x in enumerate(Agenda.objects.order_by('slug'))
|
||||
]
|
||||
return Response({'data': response})
|
||||
|
||||
|
||||
list_agendas = ListAgendas.as_view()
|
||||
|
||||
|
||||
class ExportAgenda(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def get(self, request, slug, *args, **kwargs):
|
||||
serialisation = Agenda.objects.get(slug=slug).export_json()
|
||||
return Response({'data': serialisation})
|
||||
|
||||
|
||||
export_agenda = ExportAgenda.as_view()
|
||||
|
||||
|
||||
class AgendaDependencies(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def get(self, request, slug, *args, **kwargs):
|
||||
agenda = Agenda.objects.get(slug=slug)
|
||||
|
||||
def dependency_dict(element):
|
||||
if isinstance(element, Group):
|
||||
return {
|
||||
'id': element.role.slug if hasattr(element, 'role') else element.id,
|
||||
'text': element.name,
|
||||
'type': 'roles',
|
||||
'urls': {},
|
||||
}
|
||||
elif isinstance(element, Agenda):
|
||||
return get_agenda_bundle_entry(request, element, 0)
|
||||
return element
|
||||
|
||||
dependencies = [dependency_dict(x) for x in agenda.get_dependencies() if x]
|
||||
return Response({'err': 0, 'data': dependencies})
|
||||
|
||||
|
||||
agenda_dependencies = AgendaDependencies.as_view()
|
||||
|
||||
|
||||
def agenda_redirect(request, slug):
|
||||
agenda = get_object_or_404(Agenda, slug=slug)
|
||||
return redirect(reverse('chrono-manager-agenda-view', kwargs={'pk': agenda.pk}))
|
||||
|
||||
|
||||
class BundleCheck(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def put(self, request, *args, **kwargs):
|
||||
return Response({'err': 0, 'data': {}})
|
||||
|
||||
|
||||
bundle_check = BundleCheck.as_view()
|
||||
|
||||
|
||||
class BundleImport(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
install = True
|
||||
|
||||
def put(self, request, *args, **kwargs):
|
||||
tar_io = io.BytesIO(request.read())
|
||||
agendas = []
|
||||
with tarfile.open(fileobj=tar_io) as tar:
|
||||
manifest = json.loads(tar.extractfile('manifest.json').read().decode())
|
||||
self.application = Application.update_or_create_from_manifest(
|
||||
manifest,
|
||||
tar,
|
||||
editable=not self.install,
|
||||
)
|
||||
|
||||
for element in manifest.get('elements'):
|
||||
if element.get('type') != 'agendas':
|
||||
continue
|
||||
agendas.append(
|
||||
json.loads(tar.extractfile(f'agendas/{element["slug"]}').read().decode()).get('data')
|
||||
)
|
||||
# init cache of application elements, from manifest
|
||||
self.application_elements = set()
|
||||
# import agendas
|
||||
self.do_something(agendas)
|
||||
# create application elements
|
||||
self.link_objects(agendas)
|
||||
# remove obsolete application elements
|
||||
self.unlink_obsolete_objects()
|
||||
return Response({'err': 0})
|
||||
|
||||
def do_something(self, agendas):
|
||||
if agendas:
|
||||
import_site({'agendas': agendas})
|
||||
|
||||
def link_objects(self, agendas):
|
||||
for agenda in agendas:
|
||||
try:
|
||||
existing_agenda = Agenda.objects.get(slug=agenda['slug'])
|
||||
except Agenda.DoesNotExist:
|
||||
pass
|
||||
else:
|
||||
element = ApplicationElement.update_or_create_for_object(self.application, existing_agenda)
|
||||
self.application_elements.add(element.content_object)
|
||||
|
||||
def unlink_obsolete_objects(self):
|
||||
known_elements = ApplicationElement.objects.filter(application=self.application)
|
||||
for element in known_elements:
|
||||
if element.content_object not in self.application_elements:
|
||||
element.delete()
|
||||
|
||||
|
||||
bundle_import = BundleImport.as_view()
|
||||
|
||||
|
||||
class BundleDeclare(BundleImport):
|
||||
install = False
|
||||
|
||||
def do_something(self, agendas):
|
||||
# no installation on declare
|
||||
pass
|
||||
|
||||
|
||||
bundle_declare = BundleDeclare.as_view()
|
||||
|
||||
|
||||
class BundleUnlink(GenericAPIView):
|
||||
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
if request.POST.get('application'):
|
||||
try:
|
||||
application = Application.objects.get(slug=request.POST['application'])
|
||||
except Application.DoesNotExist:
|
||||
pass
|
||||
else:
|
||||
application.delete()
|
||||
|
||||
return Response({'err': 0})
|
||||
|
||||
|
||||
bundle_unlink = BundleUnlink.as_view()
|
|
@ -0,0 +1,62 @@
|
|||
# Generated by Django 3.2.18 on 2023-10-13 09:19
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('contenttypes', '0002_remove_content_type_name'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Application',
|
||||
fields=[
|
||||
(
|
||||
'id',
|
||||
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
|
||||
),
|
||||
('name', models.CharField(max_length=100)),
|
||||
('slug', models.SlugField(max_length=100, unique=True)),
|
||||
('icon', models.FileField(blank=True, null=True, upload_to='applications/icons/')),
|
||||
('description', models.TextField(blank=True)),
|
||||
('documentation_url', models.URLField(blank=True)),
|
||||
('version_number', models.CharField(max_length=100)),
|
||||
('version_notes', models.TextField(blank=True)),
|
||||
('editable', models.BooleanField(default=True)),
|
||||
('visible', models.BooleanField(default=True)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
],
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='ApplicationElement',
|
||||
fields=[
|
||||
(
|
||||
'id',
|
||||
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
|
||||
),
|
||||
('object_id', models.PositiveIntegerField()),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
(
|
||||
'application',
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE, to='export_import.application'
|
||||
),
|
||||
),
|
||||
(
|
||||
'content_type',
|
||||
models.ForeignKey(
|
||||
on_delete=django.db.models.deletion.CASCADE, to='contenttypes.contenttype'
|
||||
),
|
||||
),
|
||||
],
|
||||
options={
|
||||
'unique_together': {('application', 'content_type', 'object_id')},
|
||||
},
|
||||
),
|
||||
]
|
|
@ -0,0 +1,131 @@
|
|||
# chrono - content management system
|
||||
# Copyright (C) 2016-2023 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 collections
|
||||
|
||||
from django.contrib.contenttypes.fields import GenericForeignKey
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.db import models
|
||||
|
||||
|
||||
class Application(models.Model):
|
||||
name = models.CharField(max_length=100)
|
||||
slug = models.SlugField(max_length=100, unique=True)
|
||||
icon = models.FileField(
|
||||
upload_to='applications/icons/',
|
||||
blank=True,
|
||||
null=True,
|
||||
)
|
||||
description = models.TextField(blank=True)
|
||||
documentation_url = models.URLField(blank=True)
|
||||
version_number = models.CharField(max_length=100)
|
||||
version_notes = models.TextField(blank=True)
|
||||
editable = models.BooleanField(default=True)
|
||||
visible = models.BooleanField(default=True)
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
def __str__(self):
|
||||
return str(self.name)
|
||||
|
||||
@classmethod
|
||||
def update_or_create_from_manifest(cls, manifest, tar, editable=False):
|
||||
application, dummy = cls.objects.get_or_create(
|
||||
slug=manifest.get('slug'), defaults={'editable': editable}
|
||||
)
|
||||
application.name = manifest.get('application')
|
||||
application.description = manifest.get('description')
|
||||
application.documentation_url = manifest.get('documentation_url')
|
||||
application.version_number = manifest.get('version_number') or 'unknown'
|
||||
application.version_notes = manifest.get('version_notes')
|
||||
if not editable:
|
||||
application.editable = editable
|
||||
application.visible = manifest.get('visible', True)
|
||||
application.save()
|
||||
icon = manifest.get('icon')
|
||||
if icon:
|
||||
application.icon.save(icon, tar.extractfile(icon), save=True)
|
||||
else:
|
||||
application.icon.delete()
|
||||
return application
|
||||
|
||||
@classmethod
|
||||
def select_for_object_class(cls, object_class):
|
||||
content_type = ContentType.objects.get_for_model(object_class)
|
||||
elements = ApplicationElement.objects.filter(content_type=content_type)
|
||||
return cls.objects.filter(pk__in=elements.values('application'), visible=True).order_by('name')
|
||||
|
||||
@classmethod
|
||||
def populate_objects(cls, object_class, objects):
|
||||
content_type = ContentType.objects.get_for_model(object_class)
|
||||
elements = ApplicationElement.objects.filter(content_type=content_type)
|
||||
elements_by_objects = collections.defaultdict(list)
|
||||
for element in elements:
|
||||
elements_by_objects[element.content_object].append(element)
|
||||
applications_by_ids = {
|
||||
a.pk: a for a in cls.objects.filter(pk__in=elements.values('application'), visible=True)
|
||||
}
|
||||
for obj in objects:
|
||||
applications = []
|
||||
elements = elements_by_objects.get(obj) or []
|
||||
for element in elements:
|
||||
application = applications_by_ids.get(element.application_id)
|
||||
if application:
|
||||
applications.append(application)
|
||||
obj._applications = sorted(applications, key=lambda a: a.name)
|
||||
|
||||
@classmethod
|
||||
def load_for_object(cls, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj.__class__)
|
||||
elements = ApplicationElement.objects.filter(content_type=content_type, object_id=obj.pk)
|
||||
applications_by_ids = {
|
||||
a.pk: a for a in cls.objects.filter(pk__in=elements.values('application'), visible=True)
|
||||
}
|
||||
applications = []
|
||||
for element in elements:
|
||||
application = applications_by_ids.get(element.application_id)
|
||||
if application:
|
||||
applications.append(application)
|
||||
obj._applications = sorted(applications, key=lambda a: a.name)
|
||||
|
||||
def get_objects_for_object_class(self, object_class):
|
||||
content_type = ContentType.objects.get_for_model(object_class)
|
||||
elements = ApplicationElement.objects.filter(content_type=content_type, application=self)
|
||||
return object_class.objects.filter(pk__in=elements.values('object_id'))
|
||||
|
||||
|
||||
class ApplicationElement(models.Model):
|
||||
application = models.ForeignKey(Application, on_delete=models.CASCADE)
|
||||
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
||||
object_id = models.PositiveIntegerField()
|
||||
content_object = GenericForeignKey('content_type', 'object_id')
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
class Meta:
|
||||
unique_together = ['application', 'content_type', 'object_id']
|
||||
|
||||
@classmethod
|
||||
def update_or_create_for_object(cls, application, obj):
|
||||
content_type = ContentType.objects.get_for_model(obj.__class__)
|
||||
element, created = cls.objects.get_or_create(
|
||||
application=application,
|
||||
content_type=content_type,
|
||||
object_id=obj.pk,
|
||||
)
|
||||
if not created:
|
||||
element.save()
|
||||
return element
|
|
@ -0,0 +1,43 @@
|
|||
# chrono - content management system
|
||||
# Copyright (C) 2016-2023 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.urls import path
|
||||
|
||||
from . import api_views
|
||||
|
||||
urlpatterns = [
|
||||
path('export-import/', api_views.index, name='api-export-import'),
|
||||
path('export-import/agendas/', api_views.list_agendas, name='api-export-import-agendas-list'),
|
||||
path(
|
||||
'export-import/agendas/<slug:slug>/',
|
||||
api_views.export_agenda,
|
||||
name='api-export-import-agenda-export',
|
||||
),
|
||||
path(
|
||||
'export-import/agendas/<slug:slug>/dependencies/',
|
||||
api_views.agenda_dependencies,
|
||||
name='api-export-import-agenda-dependencies',
|
||||
),
|
||||
path(
|
||||
'export-import/agendas/<slug:slug>/redirect/',
|
||||
api_views.agenda_redirect,
|
||||
name='api-export-import-agenda-redirect',
|
||||
),
|
||||
path('export-import/bundle-check/', api_views.bundle_check),
|
||||
path('export-import/bundle-declare/', api_views.bundle_declare),
|
||||
path('export-import/bundle-import/', api_views.bundle_import),
|
||||
path('export-import/unlink/', api_views.bundle_unlink),
|
||||
]
|
|
@ -58,6 +58,7 @@ INSTALLED_APPS = (
|
|||
'chrono.api',
|
||||
'chrono.manager',
|
||||
'chrono.apps.ants_hub',
|
||||
'chrono.apps.export_import',
|
||||
'rest_framework',
|
||||
'django_filters',
|
||||
)
|
||||
|
|
|
@ -20,6 +20,8 @@ from django.contrib.auth.decorators import login_required
|
|||
from django.contrib.staticfiles.urls import staticfiles_urlpatterns
|
||||
from django.urls import include, path, re_path
|
||||
|
||||
from chrono.apps.export_import import urls as export_import_urls
|
||||
|
||||
from .api.urls import urlpatterns as chrono_api_urls
|
||||
from .manager.urls import urlpatterns as chrono_manager_urls
|
||||
from .urls_utils import decorated_includes
|
||||
|
@ -29,6 +31,7 @@ urlpatterns = [
|
|||
path('', homepage, name='home'),
|
||||
re_path(r'^manage/', decorated_includes(login_required, include(chrono_manager_urls))),
|
||||
path('api/', include(chrono_api_urls)),
|
||||
path('api/', include(export_import_urls)),
|
||||
path('logout/', LogoutView.as_view(), name='auth_logout'),
|
||||
path('login/', LoginView.as_view(), name='auth_login'),
|
||||
]
|
||||
|
|
|
@ -0,0 +1,333 @@
|
|||
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
|
||||
from chrono.apps.export_import.models import Application, ApplicationElement
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_object_types(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
resp = app.get('/api/export-import/')
|
||||
assert resp.json == {
|
||||
'data': [
|
||||
{
|
||||
'id': 'agendas',
|
||||
'text': 'Agendas',
|
||||
'singular': 'Agenda',
|
||||
'urls': {'list': 'http://testserver/api/export-import/agendas/'},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def test_list_agendas(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
Agenda.objects.all().delete()
|
||||
Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
|
||||
Agenda.objects.create(label='Event', slug='event', kind='events')
|
||||
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/',
|
||||
},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def test_export_agenda(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
group1 = Group.objects.create(name='group1')
|
||||
group2 = Group.objects.create(name='group2')
|
||||
Agenda.objects.all().delete()
|
||||
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_agenda_dependencies_groups(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
group1 = Group.objects.create(name='group1')
|
||||
group2 = Group.objects.create(name='group2')
|
||||
Agenda.objects.all().delete()
|
||||
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': {}},
|
||||
{'id': group1.id, 'text': group1.name, 'type': 'roles', 'urls': {}},
|
||||
],
|
||||
'err': 0,
|
||||
}
|
||||
|
||||
|
||||
def test_agenda_dependencies_virtual_agendas(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
Agenda.objects.all().delete()
|
||||
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_redirect(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
agenda = Agenda.objects.create(label='Rdv', slug='rdv', kind='meetings')
|
||||
redirect_url = '/api/export-import/agendas/rdv/redirect/'
|
||||
resp = app.get(redirect_url, status=302)
|
||||
assert resp.location == f'/manage/agendas/{agenda.pk}/'
|
||||
|
||||
|
||||
def create_bundle(app, user, visible=True, version_number='42.0'):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
group, _ = Group.objects.get_or_create(name='plop')
|
||||
Agenda.objects.get_or_create(
|
||||
slug='rdv', defaults={'label': 'Rdv', 'kind': 'meetings', 'edit_role': group}
|
||||
)
|
||||
agenda_export = app.get('/api/export-import/agendas/rdv/').content
|
||||
|
||||
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': [
|
||||
{'type': 'agendas', 'slug': 'rdv', 'name': 'Test agenda', 'auto-dependency': False},
|
||||
{'type': 'form', 'slug': 'xxx', 'name': 'Xxx'},
|
||||
],
|
||||
}
|
||||
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)
|
||||
|
||||
tarinfo = tarfile.TarInfo('agendas/rdv')
|
||||
tarinfo.size = len(agenda_export)
|
||||
tar.addfile(tarinfo, fileobj=io.BytesIO(agenda_export))
|
||||
bundle = tar_io.getvalue()
|
||||
return bundle
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bundle(app, user):
|
||||
return create_bundle(app, user)
|
||||
|
||||
|
||||
def test_bundle_import(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
|
||||
Agenda.objects.all().delete()
|
||||
bundles = []
|
||||
for version_number in ['42.0', '42.1']:
|
||||
bundles.append(create_bundle(app, user, version_number=version_number))
|
||||
|
||||
resp = app.put('/api/export-import/bundle-import/', bundles[0])
|
||||
assert Agenda.objects.all().count() == 1
|
||||
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() == 1
|
||||
|
||||
# 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.put('/api/export-import/bundle-import/', bundles[1])
|
||||
assert Agenda.objects.all().count() == 1
|
||||
assert resp.json['err'] == 0
|
||||
assert Application.objects.count() == 1
|
||||
application = Application.objects.latest('pk')
|
||||
assert application.editable is False
|
||||
assert ApplicationElement.objects.count() == 1
|
||||
assert (
|
||||
ApplicationElement.objects.filter(
|
||||
application=application,
|
||||
content_type=ContentType.objects.get_for_model(Agenda),
|
||||
object_id=last_agenda.pk + 1,
|
||||
).exists()
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
def test_bundle_declare(app, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
|
||||
bundle = create_bundle(app, user, visible=False)
|
||||
resp = app.put('/api/export-import/bundle-declare/', bundle)
|
||||
assert Agenda.objects.all().count() == 1
|
||||
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() == 1
|
||||
|
||||
bundle = create_bundle(app, 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 a page to have an unkown reference in manifest
|
||||
Agenda.objects.all().delete()
|
||||
|
||||
resp = app.put('/api/export-import/bundle-declare/', bundle)
|
||||
assert Application.objects.count() == 1
|
||||
application = Application.objects.latest('pk')
|
||||
assert application.visible is True
|
||||
assert ApplicationElement.objects.count() == 0
|
||||
|
||||
|
||||
def test_bundle_unlink(app, user, bundle):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
|
||||
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, user):
|
||||
app.authorization = ('Basic', ('john.doe', 'password'))
|
||||
assert app.put('/api/export-import/bundle-check/').json == {'err': 0, 'data': {}}
|
Loading…
Reference in New Issue