0001-data-merge-assets-and-site-export-file-39425.patch

Nicolas Roche, 13 février 2020 14:19

Voir les différences: en ligne côte à côte

Subject: [PATCH] data: merge assets and site export file (#39425)

 combo/apps/assets/utils.py                    |  64 +++
 combo/apps/assets/views.py                    |  34 +-
 combo/data/management/commands/export_site.py |  13 +-
 combo/data/management/commands/import_site.py |  31 +-
 combo/data/utils.py                           |  46 +-
 combo/manager/views.py                        |  37 +-
 tests/test_assets.py                          |  58 +++
 tests/test_import_export.py                   | 468 ++++++++++--------
 tests/test_manager.py                         |  25 +-
 tests/test_pages.py                           |  79 +--
 10 files changed, 534 insertions(+), 321 deletions(-)
 create mode 100644 combo/apps/assets/utils.py

     # combo - content management system
     # 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/>.
     import json
     import os
     import tarfile
     from django.core.files.storage import default_storage
     from django.utils.six import BytesIO
     from .models import Asset
     def clean_assets():
         media_prefix = default_storage.path('')
         for basedir, dirnames, filenames in os.walk(media_prefix):
             for filename in filenames:
                 os.remove('%s/%s' % (basedir, filename))
         Asset.objects.all().delete()
     def add_tar_content(tar, filename, content):
         file = tarfile.TarInfo(filename)
         fd = BytesIO()
         fd.write(content.encode('utf-8'))
         file.size = fd.tell()
         fd.seek(0)
         tar.addfile(file, fileobj=fd)
         fd.close()
     def import_assets(tar, overwrite=False):
         media_prefix = default_storage.path('')
         for tarinfo in tar.getmembers():
             filepath = default_storage.path(tarinfo.name)
             if not overwrite and os.path.exists(filepath):
                 continue
             if tarinfo.name == '_assets.json':
                 json_assets = tar.extractfile(tarinfo).read()
                 data = json.loads(json_assets.decode('utf-8'))
                 Asset.load_serialized_objects(data.get('assets') or [])
             elif tarinfo.name != '_site.json':
                 tar.extract(tarinfo, path=media_prefix)
     def export_assets(tar):
         media_prefix = default_storage.path('')
         for basedir, dirnames, filenames in os.walk(media_prefix):
             for filename in filenames:
                 tar.add(
                         os.path.join(basedir, filename),
                         os.path.join(basedir, filename)[len(media_prefix):])
         export = {'assets': Asset.export_all_for_json()}
         add_tar_content(tar, '_assets.json', json.dumps(export, indent=2))

     from django.utils.translation import ugettext_lazy as _
     from django.views.generic import TemplateView, ListView, FormView
     import ckeditor
     from sorl.thumbnail.shortcuts import get_thumbnail
     from combo.data.models import CellBase
     from combo.data.utils import import_site
     from combo.apps.assets.utils import import_assets, export_assets
     from .forms import AssetUploadForm, AssetsImportForm
     from .models import Asset
     class CkEditorAsset(object):
         def __init__(self, filepath):
             self.filepath = filepath
-...
     class AssetsImport(FormView):
         form_class = AssetsImportForm
         template_name = 'combo/manager_assets_import.html'
         success_url = reverse_lazy('combo-manager-assets')
         def form_valid(self, form):
             overwrite = form.cleaned_data.get('overwrite')
             try:
                 assets = tarfile.open(fileobj=form.cleaned_data['assets_file'])
                 tar = tarfile.open(fileobj=form.cleaned_data['assets_file'])
             except tarfile.TarError:
                 messages.error(self.request, _('The assets file is not valid.'))
                 return super(AssetsImport, self).form_valid(form)
             media_prefix = default_storage.path('')
             for tarinfo in assets.getmembers():
                 filepath = default_storage.path(tarinfo.name)
                 if not overwrite and os.path.exists(filepath):
                     continue
                 if tarinfo.name == '_assets.json':
                     json_assets = assets.extractfile(tarinfo).read()
                     import_site(json.loads(json_assets.decode('utf-8')))
                 else:
                     assets.extract(tarinfo, path=media_prefix)
             import_assets(tar, overwrite)
             tar.close()
             messages.success(self.request, _('The assets file has been imported.'))
             return super(AssetsImport, self).form_valid(form)
     assets_import = AssetsImport.as_view()
     def assets_export(request, *args, **kwargs):
         fd = BytesIO()
         assets_file = tarfile.open('assets.tar', 'w', fileobj=fd)
         media_prefix = default_storage.path('')
         for basedir, dirnames, filenames in os.walk(media_prefix):
             for filename in filenames:
                 assets_file.add(
                         os.path.join(basedir, filename),
                         os.path.join(basedir, filename)[len(media_prefix):])
         if Asset.objects.exists():
             json_file = tarfile.TarInfo('_assets.json')
             json_fd = BytesIO()
             export = {'assets': Asset.export_all_for_json(),}
             json_fd.write(json.dumps(export).encode('utf-8'))
             json_file.size = json_fd.tell()
             json_fd.seek(0)
             assets_file.addfile(json_file, fileobj=json_fd)
         assets_file.close()
         tar = tarfile.open('assets.tar', 'w', fileobj=fd)
         export_assets(tar)
         tar.close()
         return HttpResponse(fd.getvalue(), content_type='application/x-tar')
     def serve_asset(request, key):
         asset = get_object_or_404(Asset, key=key)
         if not os.path.exists(asset.asset.path):
             raise Http404()

     # 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 json
     import sys
     import tarfile
     from django.core.management.base import BaseCommand
     from django.core.management.base import BaseCommand, CommandError
     from combo.data.utils import export_site
     class Command(BaseCommand):
         help = 'Export the site'
         def add_arguments(self, parser):
             parser.add_argument(
                     '--output', metavar='FILE', default=None,
                     help='name of a file to write output to')
         def handle(self, *args, **options):
             if options['output'] and options['output'] != '-':
                 output = open(options['output'], 'w')
                 try:
                     tar = tarfile.open(options['output'], 'w')
                 except IOError as e:
                     raise CommandError(e)
             else:
                 output = sys.stdout
             json.dump(export_site(), output, indent=2)
                 tar = tarfile.open(mode='w|', fileobj=sys.stdout.buffer)
             export_site(tar)
             tar.close()

     # 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 json
     import sys
     import tarfile
     from django.core.management.base import BaseCommand, CommandError
     from django.utils.encoding import force_text
     from django.utils.six import BytesIO
     from combo.data.utils import import_site, MissingGroups
     from combo.data.utils import import_site, ImportSiteError
     class Command(BaseCommand):
         help = 'Import an exported site'
         def add_arguments(self, parser):
             parser.add_argument('filename', metavar='FILENAME', type=str,
                     help='name of file to import')
             parser.add_argument(
                     '--clean', action='store_true', default=False,
                     help='Clean site before importing')
             parser.add_argument(
                     '--if-empty', action='store_true', default=False,
                     help='Import only if site is empty')
         def handle(self, filename, *args, **options):
             if filename == '-':
                 fd = sys.stdin
                 fd = BytesIO(sys.stdin.buffer.read())
             else:
                 fd = open(filename)
                 try:
                     fd = open(filename, 'rb')
                 except IOError as e:
                     raise CommandError(e)
             tar = None
             json_data = None
             if fd != sys.stdin:
                 try:
                     tar = tarfile.open(fileobj=fd)
                 except tarfile.TarError as e:
                     pass
             if not tar:
                 try:
                     json_data = json.loads(force_text(fd.read()))
                 except ValueError as e:
                     raise CommandError('%s: %s' % (e.__class__.__name__, str(e)))
             try:
                 import_site(json.load(fd),
                 import_site(data=json_data,
                             tar=tar,
                             if_empty=options['if_empty'],
                             clean=options['clean'])
             except MissingGroups as e:
             except ImportSiteError as e:
                 raise CommandError(e)
             if tar:
                 tar.close()

     # 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 json
     from django.contrib.auth.models import Group
     from django.db import transaction
     from django.utils import six
     from django.utils.encoding import python_2_unicode_compatible
     from django.utils.translation import ugettext_lazy as _
     from combo.apps.assets.models import Asset
     from combo.apps.assets.utils import add_tar_content, clean_assets, import_assets, export_assets
     from combo.apps.maps.models import MapLayer
     from combo.apps.pwa.models import PwaSettings, PwaNavigationEntry
     from .models import Page
     class ImportSiteError(Exception):
         pass
     @python_2_unicode_compatible
     class MissingGroups(Exception):
     class MissingGroups(ImportSiteError):
         def __init__(self, names):
             self.names = names
         def __str__(self):
             return _('Missing groups: %s') % ', '.join(self.names)
     def export_site():
         '''Dump site objects to JSON-dumpable dictionnary'''
         return {'pages': Page.export_all_for_json(),
                 'map-layers': MapLayer.export_all_for_json(),
                 'assets': Asset.export_all_for_json(),
                 'pwa': {
                     'settings': PwaSettings.export_for_json(),
                     'navigation': PwaNavigationEntry.export_all_for_json(),
+                }
+               }
     def export_site(tar):
         data = {
             'pages': Page.export_all_for_json(),
             'map-layers': MapLayer.export_all_for_json(),
             'pwa': {
                 'settings': PwaSettings.export_for_json(),
                 'navigation': PwaNavigationEntry.export_all_for_json(),
+            }
+        }
         add_tar_content(tar, '_site.json', json.dumps(data, indent=2))
         export_assets(tar)
     def import_site(data, if_empty=False, clean=False):
     def import_site(data=None, tar=None, if_empty=False, clean=False):
         if isinstance(data, list):
             # old export form with a list of pages, convert it to new dictionary
             # format.
             data = {'pages': data}
         if if_empty and (Page.objects.count() or MapLayer.objects.count()):
             return
         if tar:
             try:
                 tarinfo = tar.getmember('_site.json')
             except KeyError:
                 raise ImportSiteError(_('TAR file should provide _site.json file'))
             json_data = tar.extractfile(tarinfo).read()
             data = json.loads(json_data.decode('utf-8'))
         # check groups used in access control are all available.
         groups = set()
         for page in data.get('pages') or []:
             for group in page['fields']['groups']:
                 groups.add(group if isinstance(group, six.string_types) else group[0])
             for cell in page['cells']:
                 for group in cell['fields']['groups']:
                     groups.add(group if isinstance(group, six.string_types) else group[0])
-...
         existing_groups = set([x.name for x in Group.objects.filter(name__in=groups)])
         missing_groups = groups - existing_groups
         if missing_groups:
             raise MissingGroups(names=sorted([x for x in missing_groups]))
         with transaction.atomic():
             if clean:
                 MapLayer.objects.all().delete()
                 clean_assets()
                 Asset.objects.all().delete()
                 Page.objects.all().delete()
                 PwaSettings.objects.all().delete()
                 PwaNavigationEntry.objects.all().delete()
             MapLayer.load_serialized_objects(data.get('map-layers') or [])
             Asset.load_serialized_objects(data.get('assets') or [])
             if not tar:
                 Asset.load_serialized_objects(data.get('assets') or [])
             else:
                 import_assets(tar, overwrite=not if_empty)
             Page.load_serialized_pages(data.get('pages') or [])
             if data.get('pwa'):
                 PwaSettings.load_serialized_settings(data['pwa'].get('settings'))
                 PwaNavigationEntry.load_serialized_objects(data['pwa'].get('navigation'))

     # 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 hashlib
     import json
     import os
     import tarfile
     from django.conf import settings
     from django.contrib import messages
     from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
     from django.core.exceptions import ObjectDoesNotExist
     from django.core.urlresolvers import reverse, reverse_lazy
     from django.http import HttpResponse, HttpResponseRedirect, Http404
     from django.shortcuts import redirect
     from django.shortcuts import get_object_or_404
     from django.utils.six import BytesIO
     from django.utils.translation import ugettext_lazy as _
     from django.utils.encoding import force_text, force_bytes
     from django.utils.formats import date_format
     from django.utils.timezone import localtime
     from django.views.decorators.csrf import requires_csrf_token
     from django.views.generic import (RedirectView, DetailView,
             CreateView, UpdateView, ListView, DeleteView, FormView)
     from combo.data.models import Page, CellBase, ParentContentCell, PageSnapshot, LinkListCell
     from combo.data.library import get_cell_class
     from combo.data.utils import export_site, import_site, MissingGroups
     from combo.data.utils import export_site, import_site, ImportSiteError
     from combo import plugins
     from .forms import (PageEditTitleForm, PageVisibilityForm, SiteImportForm,
             PageEditRedirectionForm, PageSelectTemplateForm, PageEditSlugForm,
             PageEditPictureForm, PageEditIncludeInNavigationForm,
             PageEditDescriptionForm, CellVisibilityForm)
-...
     homepage = HomepageView.as_view()
     class SiteExportView(ListView):
         model = Page
         def render_to_response(self, context, **response_kwargs):
             response = HttpResponse(content_type='application/json')
             json.dump(export_site(), response, indent=2)
             return response
             fd = BytesIO()
             tar = tarfile.open(mode='w', fileobj=fd)
             export_site(tar)
             tar.close()
             return HttpResponse(fd.getvalue(), content_type='application/x-tar')
     site_export = SiteExportView.as_view()
     class SiteImportView(FormView):
         form_class = SiteImportForm
         template_name = 'combo/site_import.html'
         success_url = reverse_lazy('combo-manager-homepage')
         def form_valid(self, form):
             content = self.request.FILES['site_json'].read()
             fd = BytesIO(content)
             tar = None
             json_data = None
             try:
                 json_site = json.loads(force_text(self.request.FILES['site_json'].read()))
             except ValueError:
                 form.add_error('site_json', _('File is not in the expected JSON format.'))
                 return self.form_invalid(form)
                 tar = tarfile.open(fileobj=fd)
             except tarfile.TarError as e:
                 try:
                     json_data = json.loads(force_text(content))
                 except ValueError:
                     form.add_error('site_json', _('File is not in the expected JSON format.'))
                     return self.form_invalid(form)
             try:
                 import_site(json_site)
             except MissingGroups as e:
                 import_site(data=json_data, tar=tar)
             except ImportSiteError as e:
                 form.add_error('site_json', force_text(e))
                 return self.form_invalid(form)
             if tar:
                 tar.close()
             return super(SiteImportView, self).form_valid(form)
     site_import = SiteImportView.as_view()
     class PageAddView(CreateView):
         model = Page

     # -*- coding: utf-8 -*-
     import base64
     import os
     import tarfile
     from django.core.urlresolvers import reverse
     from django.core.files.storage import default_storage
     from django.core.files import File
     from django.utils.six import BytesIO
     import pytest
     from combo.apps.assets.models import Asset
     from combo.apps.assets.utils import add_tar_content, clean_assets, export_assets, import_assets
     pytestmark = pytest.mark.django_db
     @pytest.fixture
     def some_assets():
         Asset(key='banner', asset=File(BytesIO(b'test'), 'test.png')).save()
         Asset(key='favicon', asset=File(BytesIO(b'test2'), 'test2.png')).save()
     def count_asset_files():
         nb_assets = 0
         media_prefix = default_storage.path('')
         for basedir, dirnames, filenames in os.walk(media_prefix):
             nb_assets += len(filenames)
         return nb_assets
     def test_asset_set_api(app, john_doe):
         app.authorization = ('Basic', (john_doe.username, john_doe.username))
         resp = app.post_json(reverse('api-assets-set', kwargs={'key': 'plop'}), params={
             'asset': {
                 'content': base64.encodebytes(b'plop').decode('ascii'),
                 'content_type': 'text/plain',
                 'filename': 'plop.txt',
+            }
-...
             resp = app.post_json(reverse('api-assets-set', kwargs={'key': 'plop'}), params={
                 'asset': {
                     'content': invalid_value,
                     'content_type': 'text/plain',
                     'filename': 'plop.txt',
+                }
                 }, status=400)
             assert resp.json.get('err') == 1
     def test_add_tar_content(tmpdir):
         filename = os.path.join(str(tmpdir), 'file.tar')
         tar = tarfile.open(filename, 'w')
         add_tar_content(tar, 'foo.txt', 'bar')
         tar.close()
         tar = tarfile.open(filename, 'r')
         tarinfo = tar.getmember('foo.txt')
         assert tar.extractfile(tarinfo).read().decode('utf-8') == 'bar'
     def test_import_export_assets(app, some_assets):
         assert Asset.objects.count() == 2
         assert count_asset_files() == 2
         fd = BytesIO()
         assets_file = tarfile.open(mode='w', fileobj=fd)
         export_assets(assets_file)
         tar_bytes = fd.getvalue()
         path = default_storage.path('')
         Asset.objects.all().delete()
         assert Asset.objects.count() == 0
         os.remove('%s/assets/test.png' % path)
         open('%s/assets/test2.png' % path, 'w').write('foo')
         fd = BytesIO(tar_bytes)
         assert count_asset_files() == 1
         assets_file = tarfile.open(fileobj=fd)
         import_assets(assets_file)
         assert Asset.objects.count() == 2
         assert count_asset_files() == 2
         assert open('%s/assets/test.png' % path, 'r').read() == 'test'
         assert open('%s/assets/test2.png' % path, 'r').read() == 'foo'
         fd = BytesIO(tar_bytes)
         assets_file = tarfile.open(fileobj=fd)
         import_assets(assets_file, True)
         assert open('%s/assets/test2.png' % path, 'r').read() == 'test2'
         clean_assets()
         assert count_asset_files() == 0
         assert Asset.objects.count() == 0

     import base64
     import datetime
     import io
     import json
     import os
     import shutil
     import sys
     import tarfile
     import tempfile
     import pytest
     from django.contrib.auth.models import Group
     from django.core.files import File
     from django.core.management import call_command
     from django.core.management.base import CommandError
     from django.utils.encoding import force_bytes, force_text
     from django.utils.six import BytesIO, StringIO
     from combo.apps.assets.models import Asset
     from combo.apps.assets.utils import export_assets
     from combo.apps.gallery.models import Image, GalleryCell
     from combo.apps.maps.models import MapLayer, Map
     from combo.apps.pwa.models import PwaSettings, PwaNavigationEntry
     from combo.data.models import Page, TextCell
     from combo.data.utils import export_site, import_site, MissingGroups
     from combo.data.utils import import_site, ImportSiteError
     pytestmark = pytest.mark.django_db
     @pytest.fixture
     def some_data():
         page = Page(title='One', slug='one')
         page.save()
-...
         MapLayer(label='Foo', slug='foo', geojson_url='http://example.net/foo/').save()
         MapLayer(label='Bar', slug='bar', geojson_url='http://example.net/bar/').save()
     @pytest.fixture
     def some_assets():
         Asset(key='banner', asset=File(BytesIO(b'test'), 'test.png')).save()
         Asset(key='favicon', asset=File(BytesIO(b'test2'), 'test2.png')).save()
     def get_output_of_command(command, *args, **kwargs):
         old_stdout = sys.stdout
         output = sys.stdout = StringIO()
     def get_output_of_command(capsysbinary, command, *args, **kwargs):
         captured = capsysbinary.readouterr()
         call_command(command, *args, **kwargs)
         sys.stdout = old_stdout
         return output.getvalue()
     def test_import_export(app, some_data):
         output = get_output_of_command('export_site')
         assert len(json.loads(output)['pages']) == 3
         import_site(data={}, clean=True)
         assert Page.objects.all().count() == 0
         assert TextCell.objects.all().count() == 0
         empty_output = get_output_of_command('export_site')
         assert len(json.loads(empty_output)['pages']) == 0
         Page(title='test', slug='test').save()
         old_stdin = sys.stdin
         sys.stdin = StringIO(json.dumps({}))
         assert Page.objects.count() == 1
         try:
         captured = capsysbinary.readouterr()
         return captured.out
     def get_embedded_json(tar_as_bytes, filename):
         fd = BytesIO(tar_as_bytes)
         tar = tarfile.open(fileobj=fd)
         tarinfo = tar.getmember(filename)
         data = tar.extractfile(tarinfo).read()
         tar.close()
         return json.loads(data.decode('utf-8'))
     def test_import_export(app, some_data, capsysbinary, monkeypatch):
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(output, '_site.json')['pages']) == 3
             import_site(data={}, clean=True)
             assert Page.objects.all().count() == 0
             assert TextCell.objects.all().count() == 0
         empty_output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(empty_output, '_site.json')['pages']) == 0
             Page(title='test', slug='test').save()
             sys.stdin = StringIO(json.dumps({}))
             assert Page.objects.count() == 1
             monkeypatch.setattr('sys.stdin', io.TextIOWrapper(BytesIO(empty_output)))
             call_command('import_site', '-', clean=True)
         finally:
             sys.stdin = old_stdin
         assert Page.objects.count() == 0
             assert Page.objects.count() == 0
         with tempfile.NamedTemporaryFile() as f:
             f.write(force_bytes(output))
             f.flush()
             call_command('import_site', f.name)
         assert Page.objects.count() == 3
         assert TextCell.objects.all().count() == 1
         import_site(data={}, if_empty=True)
         assert Page.objects.count() == 3
         assert TextCell.objects.all().count() == 1
         import_site(data=[], clean=True)
         tempdir = tempfile.mkdtemp('chrono-test')
         empty_output = get_output_of_command('export_site', output=os.path.join(tempdir, 't.json'))
         assert os.path.exists(os.path.join(tempdir, 't.json'))
         shutil.rmtree(tempdir)
             with tempfile.NamedTemporaryFile() as f:
                 f.write(force_bytes(output))
                 f.flush()
                 call_command('import_site', f.name)
             assert Page.objects.count() == 3
             assert TextCell.objects.all().count() == 1
             import_site(data={}, if_empty=True)
             assert Page.objects.count() == 3
             assert TextCell.objects.all().count() == 1
             import_site(data=[], clean=True)
             tempdir = tempfile.mkdtemp('chrono-test')
         empty_output = get_output_of_command(
                 capsysbinary, 'export_site', output=os.path.join(tempdir, 't.json'))
         with capsysbinary.disabled():
             assert os.path.exists(os.path.join(tempdir, 't.json'))
             shutil.rmtree(tempdir)
     def test_tar_import_format_error(app, tmpdir):
         filename = os.path.join(str(tmpdir), 'file.tar')
         tar = tarfile.open(filename, 'w')
         export_assets(tar)
         tar.close()
         with pytest.raises(ImportSiteError, match='should provide _site.json file'):
             import_site(tar=tarfile.open(filename))
         with pytest.raises(CommandError, match='should provide _site.json file'):
             call_command('import_site', filename)
         with pytest.raises(CommandError, match=r'No such file or directory'):
             call_command('export_site', '--output', '%s/noway/foo.tar' % tmpdir)
         with pytest.raises(CommandError, match=r'No such file or directory'):
             call_command('import_site', '%s/noway/foo.tar' % tmpdir)
         open(filename, 'w').write('foo')
         with pytest.raises(CommandError, match=r'JSON'):
             call_command('import_site', filename)
     def test_backward_compatibility_import(app, some_data):
         old_export = Page.export_all_for_json()
         Page.objects.all().delete()
         import_site(data=old_export)
         assert Page.objects.count() == 3
     def test_import_export_map_layers(app, some_map_layers):
         output = get_output_of_command('export_site')
         assert len(json.loads(output)['map-layers']) == 2
         import_site(data={}, clean=True)
         assert MapLayer.objects.all().count() == 0
         empty_output = get_output_of_command('export_site')
         assert len(json.loads(empty_output)['map-layers']) == 0
         MapLayer(label='Baz', slug='baz', geojson_url='http://example.net/baz/').save()
         old_stdin = sys.stdin
         sys.stdin = StringIO(json.dumps({}))
         assert MapLayer.objects.count() == 1
         try:
             call_command('import_site', '-', clean=True)
         finally:
             sys.stdin = old_stdin
         assert MapLayer.objects.count() == 0
         with tempfile.NamedTemporaryFile() as f:
             f.write(force_bytes(output))
             f.flush()
             call_command('import_site', f.name)
         assert MapLayer.objects.count() == 2
         import_site(data={}, if_empty=True)
         assert MapLayer.objects.count() == 2
     def test_import_export_map_cells(app, some_data, some_map_layers):
         page = Page.objects.get(slug='one')
         cell = Map(page=page, order=0, placeholder='content')
         cell.save()
         cell.layers.add(MapLayer.objects.get(slug='foo'))
         cell.save()
         site_export = get_output_of_command('export_site')
         import_site(data={}, clean=True)
         assert Map.objects.count() == 0
         import_site(data=json.loads(site_export), clean=True)
         assert Map.objects.count() == 1
         assert Map.objects.all()[0].layers.all()[0].slug == 'foo'
     def test_group_restrictions_import_export(app, some_data):
         group = Group(name='A Group')
         group.save()
         page = Page.objects.get(slug='one')
         page.groups.set([group])
         page.save()
         cell = TextCell.objects.get(order=0)
         cell.groups.set([group])
         cell.save()
         output = get_output_of_command('export_site')
         assert len(json.loads(output)['pages']) == 3
         import_site(data={}, clean=True)
         assert Page.objects.all().count() == 0
         assert TextCell.objects.all().count() == 0
         Group.objects.all().delete()
     def test_import_export_map_layers(app, some_map_layers, capsysbinary, monkeypatch):
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(output, '_site.json')['map-layers']) == 2
             import_site(data={}, clean=True)
             assert MapLayer.objects.all().count() == 0
         empty_output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(empty_output, '_site.json')['map-layers']) == 0
         with pytest.raises(MissingGroups) as excinfo:
             import_site(json.loads(output), clean=True)
             MapLayer(label='Baz', slug='baz', geojson_url='http://example.net/baz/').save()
             assert MapLayer.objects.count() == 1
         assert excinfo.value.names == ['A Group']
             monkeypatch.setattr('sys.stdin', io.TextIOWrapper(BytesIO(json.dumps({}).encode("utf-8"))))
             call_command('import_site', '-', clean=True)
             assert MapLayer.objects.count() == 0
         with pytest.raises(CommandError, match='Missing groups: A Group'):
             with tempfile.NamedTemporaryFile() as f:
                 f.write(force_bytes(output))
                 f.flush()
                 call_command('import_site', f.name, clean=True)
         assert Page.objects.count() == 0
         group = Group(name='A Group')
         group.save()
         import_site(json.loads(output), clean=True)
         assert Page.objects.all().count() == 3
         assert TextCell.objects.all().count() == 1
         page = Page.objects.get(slug='one')
         assert [x.name for x in page.groups.all()] == ['A Group']
         cell = TextCell.objects.get(order=0)
         assert [x.name for x in cell.groups.all()] == ['A Group']
     def test_import_export_assets(app, some_assets):
         output = get_output_of_command('export_site')
         assert len(json.loads(output)['assets']) == 2
         import_site(data={}, clean=True)
         assert Asset.objects.all().count() == 0
         empty_output = get_output_of_command('export_site')
         assert len(json.loads(empty_output)['assets']) == 0
         Asset(key='footer', asset=File(StringIO('test3'), 'test3.png')).save()
         old_stdin = sys.stdin
         sys.stdin = StringIO(json.dumps({}))
         assert Asset.objects.count() == 1
         try:
                 call_command('import_site', f.name)
             assert MapLayer.objects.count() == 2
             import_site(data={}, if_empty=True)
             assert MapLayer.objects.count() == 2
     def test_import_export_map_cells(app, some_data, some_map_layers, capsysbinary):
         with capsysbinary.disabled():
             page = Page.objects.get(slug='one')
             cell = Map(page=page, order=0, placeholder='content')
             cell.save()
             cell.layers.add(MapLayer.objects.get(slug='foo'))
             cell.save()
         site_export = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             import_site(data={}, clean=True)
             assert Map.objects.count() == 0
             import_site(tar=tarfile.open(fileobj=BytesIO(site_export)), clean=True)
             assert Map.objects.count() == 1
             assert Map.objects.all()[0].layers.all()[0].slug == 'foo'
     def test_group_restrictions_import_export(app, some_data, capsysbinary):
         with capsysbinary.disabled():
             group = Group(name='A Group')
             group.save()
             page = Page.objects.get(slug='one')
             page.groups.set([group])
             page.save()
             cell = TextCell.objects.get(order=0)
             cell.groups.set([group])
             cell.save()
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(output, '_site.json')['pages']) == 3
             import_site(data={}, clean=True)
             assert Page.objects.all().count() == 0
             assert TextCell.objects.all().count() == 0
             Group.objects.all().delete()
             with pytest.raises(ImportSiteError) as excinfo:
                 import_site(tar=tarfile.open(fileobj=BytesIO(output)), clean=True)
             assert excinfo.value.names == ['A Group']
             with pytest.raises(CommandError, match='Missing groups: A Group'):
                 with tempfile.NamedTemporaryFile() as f:
                     f.write(force_bytes(output))
                     f.flush()
                     call_command('import_site', f.name, clean=True)
             assert Page.objects.count() == 0
             group = Group(name='A Group')
             group.save()
             import_site(tar=tarfile.open(fileobj=BytesIO(output)), clean=True)
             assert Page.objects.all().count() == 3
             assert TextCell.objects.all().count() == 1
             page = Page.objects.get(slug='one')
             assert [x.name for x in page.groups.all()] == ['A Group']
             cell = TextCell.objects.get(order=0)
             assert [x.name for x in cell.groups.all()] == ['A Group']
     def test_import_export_assets(app, some_assets, capsysbinary, monkeypatch):
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(output, '_assets.json')['assets']) == 2
             import_site(data={}, clean=True)
             assert Asset.objects.all().count() == 0
         empty_output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             assert len(get_embedded_json(empty_output, '_assets.json')['assets']) == 0
             Asset(key='footer', asset=File(StringIO('test3'), 'test3.png')).save()
             monkeypatch.setattr('sys.stdin', io.TextIOWrapper(BytesIO(json.dumps({}).encode("utf-8"))))
             assert Asset.objects.count() == 1
             call_command('import_site', '-', clean=True)
         finally:
             sys.stdin = old_stdin
         assert Asset.objects.count() == 0
         with tempfile.NamedTemporaryFile() as f:
             f.write(force_bytes(output))
             f.flush()
             call_command('import_site', f.name)
         assert Asset.objects.count() == 2
         import_site(data={}, if_empty=True)
         assert Asset.objects.count() == 2
     def test_import_export_pwa_settings(app):
         output = get_output_of_command('export_site')
         pwa_settings = PwaSettings.singleton()
         pwa_settings.offline_text = 'Hello world'
         pwa_settings.offline_retry_button = False
         pwa_settings.save()
         output = get_output_of_command('export_site')
         import_site(data={}, clean=True)
         assert PwaSettings.objects.all().count() == 0
         import_site(data=json.loads(output))
         assert PwaSettings.singleton().offline_retry_button is False
         assert PwaSettings.singleton().offline_text == 'Hello world'
     def test_import_export_pwa_navigation(app, some_data):
         page = Page.objects.get(slug='one')
         entry1 = PwaNavigationEntry(label='a', url='/', order=0)
         entry2 = PwaNavigationEntry(link_page=page, order=1, icon=File(BytesIO(b'te\30st'), 'test.png'))
         entry1.save()
         entry2.save()
         output = get_output_of_command('export_site')
         import_site(data={}, clean=True)
         assert PwaNavigationEntry.objects.all().count() == 0
         import_site(data=json.loads(output))
         assert PwaNavigationEntry.objects.all().count() == 2
         # check identical file was not touched
         assert os.path.basename(PwaNavigationEntry.objects.get(order=1).icon.file.name) == 'test.png'
         assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'te\30st'
         # check a second import doesn't create additional entries
         import_site(data=json.loads(output))
         assert PwaNavigationEntry.objects.all().count() == 2
         # check with a change in icon file content
         data = json.loads(output)
         data['pwa']['navigation'][1]['icon:base64'] = force_text(base64.encodebytes(b'TEST'))
         import_site(data=data)
         assert PwaNavigationEntry.objects.all().count() == 2
         assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'TEST'
         # check with a change in icon file name
         data = json.loads(output)
         data['pwa']['navigation'][1]['fields']['icon'] = 'pwa/test2.png'
         data['pwa']['navigation'][1]['icon:base64'] = force_text(base64.encodebytes(b'TEST2'))
         import_site(data=data)
         assert PwaNavigationEntry.objects.all().count() == 2
         assert os.path.basename(PwaNavigationEntry.objects.get(order=1).icon.file.name) == 'test2.png'
         assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'TEST2'
     def test_import_export_gallery_images(app, some_data):
         page = Page.objects.get(slug='one')
         gallery = GalleryCell(page=page, order=2, placeholder='images')
         gallery.save()
         image1 = Image(gallery=gallery, image='path/foo.jpg', title='foo', order=1)
         image2 = Image(gallery=gallery, image='path/bar.jpg', title='bar', order=2)
         image1.save()
         image2.save()
         output = get_output_of_command('export_site')
         import_site(data={}, clean=True)
         assert Image.objects.all().count() == 0
         import_site(data=json.loads(output))
         assert Image.objects.all().count() == 2
         assert image1.title == 'foo'
         assert image1.image == 'path/foo.jpg'
             assert Asset.objects.count() == 0
             with tempfile.NamedTemporaryFile() as f:
                 f.write(force_bytes(output))
                 f.flush()
                 call_command('import_site', f.name)
             assert Asset.objects.count() == 2
             import_site(data={}, if_empty=True)
             assert Asset.objects.count() == 2
     def test_import_export_pwa_settings(app, capsysbinary):
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             pwa_settings = PwaSettings.singleton()
             pwa_settings.offline_text = 'Hello world'
             pwa_settings.offline_retry_button = False
             pwa_settings.save()
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             import_site(data={}, clean=True)
             assert PwaSettings.objects.all().count() == 0
             import_site(tar=tarfile.open(fileobj=BytesIO(output)))
             assert PwaSettings.singleton().offline_retry_button is False
             assert PwaSettings.singleton().offline_text == 'Hello world'
     def test_import_export_pwa_navigation(app, some_data, capsysbinary):
         with capsysbinary.disabled():
             page = Page.objects.get(slug='one')
             entry1 = PwaNavigationEntry(label='a', url='/', order=0)
             entry2 = PwaNavigationEntry(link_page=page, order=1, icon=File(BytesIO(b'te\30st'), 'test.png'))
             entry1.save()
             entry2.save()
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             import_site(data={}, clean=True)
             assert PwaNavigationEntry.objects.all().count() == 0
             import_site(tar=tarfile.open(fileobj=BytesIO(output)))
             assert PwaNavigationEntry.objects.all().count() == 2
             # check identical file was not touched
             assert os.path.basename(PwaNavigationEntry.objects.get(order=1).icon.file.name) == 'test.png'
             assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'te\30st'
             # check a second import doesn't create additional entries
             import_site(tar=tarfile.open(fileobj=BytesIO(output)))
             assert PwaNavigationEntry.objects.all().count() == 2
             # check with a change in icon file content
             data = get_embedded_json(output, '_site.json')
             data['pwa']['navigation'][1]['icon:base64'] = force_text(base64.encodebytes(b'TEST'))
             import_site(data=data)
             assert PwaNavigationEntry.objects.all().count() == 2
             assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'TEST'
             # check with a change in icon file name
             data = get_embedded_json(output, '_site.json')
             data['pwa']['navigation'][1]['fields']['icon'] = 'pwa/test2.png'
             data['pwa']['navigation'][1]['icon:base64'] = force_text(base64.encodebytes(b'TEST2'))
             import_site(data=data)
             assert PwaNavigationEntry.objects.all().count() == 2
             assert os.path.basename(PwaNavigationEntry.objects.get(order=1).icon.file.name) == 'test2.png'
             assert PwaNavigationEntry.objects.get(order=1).icon.read() == b'TEST2'
     def test_import_export_gallery_images(app, some_data, capsysbinary):
         with capsysbinary.disabled():
             page = Page.objects.get(slug='one')
             gallery = GalleryCell(page=page, order=2, placeholder='images')
             gallery.save()
             image1 = Image(gallery=gallery, image='path/foo.jpg', title='foo', order=1)
             image2 = Image(gallery=gallery, image='path/bar.jpg', title='bar', order=2)
             image1.save()
             image2.save()
         output = get_output_of_command(capsysbinary, 'export_site')
         with capsysbinary.disabled():
             import_site(data={}, clean=True)
             assert Image.objects.all().count() == 0
             import_site(tar=tarfile.open(fileobj=BytesIO(output)))
             assert Image.objects.all().count() == 2
             assert image1.title == 'foo'
             assert image1.image == 'path/foo.jpg'

     import base64
     import json
     import os
     import re
     import shutil
     import mock
     from django.core.files.storage import default_storage
     from django.core.urlresolvers import reverse
     from django.conf import settings
     from django.contrib.auth.models import User, Group
     from django.contrib.auth.models import Group
     from django.template import TemplateSyntaxError
     from django.test import override_settings
     from django.utils.http import urlencode
     from django.utils.six import BytesIO
     from django.utils.six.moves.urllib import parse as urlparse
     import pytest
     from webtest import TestApp
-...
         resp = app.get('/manage/pages/order', params={
             'moved-page-id': page2.id,
             'moved-page-new-parent': page4.id,
             'new-order': ','.join([str(x) for x in [page1.id, page3.id, page4.id, page2.id]])})
         assert Page.objects.get(id=page1.id).parent_id is None
         assert Page.objects.get(id=page3.id).parent_id == page1.id
     def test_export_page(app, admin_user):
     def test_page_export_import(app, admin_user):
         Page.objects.all().delete()
         page = Page(title='One', slug='one', template_name='standard')
         page.save()
         app = login(app)
         resp = app.get('/manage/pages/%s/' % page.id)
         resp = resp.click('Export')
         assert resp.headers['content-type'] == 'application/json'
         assert resp.json['pages'][0].get('fields').get('slug') == 'one'
         page_export = resp.body
         Page.objects.all().delete()
         assert Page.objects.count() == 0
         app = login(app)
         resp = app.get('/manage/')
         resp = resp.click('Import Site')
         resp.form['site_json'] = Upload('page-export.json', page_export, 'application/json')
         resp = resp.form.submit()
         assert Page.objects.count() == 1
     def test_export_page_order():
         Page.objects.all().delete()
         page1 = Page(title='One', slug='one', template_name='standard')
         page2 = Page(title='Two', slug='two', parent=page1, template_name='standard')
         page3 = Page(title='Three', slug='three', parent=page2, template_name='standard')
         page4 = Page(title='Four', slug='four', parent=page1, template_name='standard')
         random_list = [page3, page4, page1, page2]
-...
         cell.save()
         cell = LinkCell(page=page2, placeholder='content', link_page=page1, order=0)
         cell.save()
         app = login(app)
         resp = app.get('/manage/')
         resp = resp.click('Export Site')
         assert resp.headers['content-type'] == 'application/json'
         assert resp.headers['content-type'] == 'application/x-tar'
         site_export = resp.body
         Page.objects.all().delete()
         assert LinkCell.objects.count() == 0
         app = login(app)
         resp = app.get('/manage/')
         resp = resp.click('Import Site')
         resp.form['site_json'] = Upload('site-export.json', site_export, 'application/json')
         resp.form['site_json'] = Upload('site-export.tar', site_export, 'application/x-tar')
         resp = resp.form.submit()
         assert Page.objects.count() == 4
         assert LinkCell.objects.count() == 2
         assert LinkCell.objects.get(page__slug='one').link_page.slug == 'two'
         assert LinkCell.objects.get(page__slug='two').link_page.slug == 'one'
         # check with invalid file
         resp = app.get('/manage/')
         resp = resp.click('Import Site')
         resp.form['site_json'] = Upload('site-export.json', b'invalid content', 'application/json')
         resp.form['site_json'] = Upload('site-export.tar', b'invalid content', 'application/x-tar')
         resp = resp.form.submit()
         assert 'File is not in the expected JSON format.' in resp.text
     def test_site_export_import_missing_group(app, admin_user):
         Page.objects.all().delete()
         group = Group.objects.create(name='foobar')
         page1 = Page(title='One', slug='one', template_name='standard')
         page1.save()
         page1.groups.set([group])
         app = login(app)
         resp = app.get('/manage/')
         resp = resp.click('Export Site')
         assert resp.headers['content-type'] == 'application/json'
         assert resp.headers['content-type'] == 'application/x-tar'
         site_export = resp.body
         Page.objects.all().delete()
         group.delete()
         app = login(app)
         resp = app.get('/manage/')
         resp = resp.click('Import Site')
         resp.form['site_json'] = Upload('site-export.json', site_export, 'application/json')
         resp.form['site_json'] = Upload('site-export.json', site_export, 'application/x-tar')
         resp = resp.form.submit()
         assert 'Missing groups: foobar' in resp.text
     def test_duplicate_page(app, admin_user):
         page = Page.objects.create(title='One', slug='one', template_name='standard', exclude_from_navigation=False)
         TextCell.objects.create(page=page, placeholder='content', text='Foobar', order=0)

     import io
     import os
     import pytest
     import sys
     from django.conf import settings
     from django.contrib.auth.models import User, Group
     from django.test import override_settings
     from django.test.client import RequestFactory
     from django.utils.six import StringIO
     from django.utils.six import BytesIO
     from combo.data.models import Page, PageSnapshot, CellBase, TextCell, LinkCell, LinkListCell
     from combo.data.management.commands.import_site import Command as ImportSiteCommand
     from combo.data.management.commands.export_site import Command as ExportSiteCommand
     from combo.manager.forms import PageVisibilityForm
     pytestmark = pytest.mark.django_db
     def test_page_order():
-...
         page3.save()
         assert page.get_next_page(None) is None
         assert page.get_next_page(check_visibility=False).id == page2.pk
         assert page2.get_next_page(check_visibility=False).id == page3.pk
         assert page3.get_previous_page(check_visibility=False).id == page2.pk
     def test_import_export_management_commands():
         page = Page(title=u'foo', slug='foo', order=0)
         page.save()
     def test_import_export_management_commands(capsysbinary, monkeypatch):
         with capsysbinary.disabled():
             page = Page(title=u'foo', slug='foo', order=0)
             page.save()
         cell = TextCell(page=page, text='foo', order=0, placeholder='content')
         cell.save()
             cell = TextCell(page=page, text='foo', order=0, placeholder='content')
             cell.save()
         page2 = Page(title=u'bar', slug='bar', order=1, parent=page)
         page2.save()
             page2 = Page(title=u'bar', slug='bar', order=1, parent=page)
             page2.save()
         cell = TextCell(page=page2, text='bar', order=0, placeholder='content')
         cell.save()
             cell = TextCell(page=page2, text='bar', order=0, placeholder='content')
             cell.save()
         export_filename = os.path.join(settings.MEDIA_ROOT, 'site-export.json')
         if os.path.exists(export_filename):
             os.unlink(export_filename)
             export_filename = os.path.join(settings.MEDIA_ROOT, 'site-export.json')
             if os.path.exists(export_filename):
                 os.unlink(export_filename)
         cmd = ExportSiteCommand()
         cmd.handle(output=export_filename)
         assert os.path.exists(export_filename)
             cmd = ExportSiteCommand()
             cmd.handle(output=export_filename)
             assert os.path.exists(export_filename)
         stdout = sys.stdout
         try:
             sys.stdout = StringIO()
             cmd.handle(output='-')
             assert sys.stdout.getvalue() == open(export_filename).read()
         finally:
             sys.stdout = stdout
         cmd.handle(output='-')
         captured = capsysbinary.readouterr()
         with capsysbinary.disabled():
             assert captured.out.decode("utf-8")[:1023] == open(export_filename).read()[:1023]
         Page.objects.all().delete()
             Page.objects.all().delete()
         cmd = ImportSiteCommand()
         cmd.handle(filename=export_filename, if_empty=False, clean=False)
             cmd = ImportSiteCommand()
             cmd.handle(filename=export_filename, if_empty=False, clean=False)
         new_page_1 = Page.objects.all().order_by('order')[0]
         new_page_2 = Page.objects.all().order_by('order')[1]
         assert new_page_1.title == 'foo'
         assert new_page_2.title == 'bar'
         assert len(CellBase.get_cells(page_id=new_page_1.id)) == 1
         assert isinstance(CellBase.get_cells(page_id=new_page_1.id)[0], TextCell)
         assert CellBase.get_cells(page_id=new_page_1.id)[0].text == 'foo'
             new_page_1 = Page.objects.all().order_by('order')[0]
             new_page_2 = Page.objects.all().order_by('order')[1]
             assert new_page_1.title == 'foo'
             assert new_page_2.title == 'bar'
             assert len(CellBase.get_cells(page_id=new_page_1.id)) == 1
             assert isinstance(CellBase.get_cells(page_id=new_page_1.id)[0], TextCell)
             assert CellBase.get_cells(page_id=new_page_1.id)[0].text == 'foo'
             Page.objects.all().delete()
             monkeypatch.setattr('sys.stdin', io.TextIOWrapper(BytesIO(captured.out)))
             cmd.handle(filename='-', if_empty=False, clean=False)
             new_page_1 = Page.objects.all().order_by('order')[0]
             new_page_2 = Page.objects.all().order_by('order')[1]
             assert new_page_1.title == 'foo'
             assert new_page_2.title == 'bar'
             assert len(CellBase.get_cells(page_id=new_page_1.id)) == 1
             assert isinstance(CellBase.get_cells(page_id=new_page_1.id)[0], TextCell)
             assert CellBase.get_cells(page_id=new_page_1.id)[0].text == 'foo'
     def test_get_placeholders():
         page = Page(title=u'foo', slug='foo', template_name='standard-sidebar', order=0)
         request = RequestFactory().get('/')
         placeholders = page.get_placeholders(request=request)
         assert [x.key for x in placeholders] == ['content', 'sidebar', 'footer']
         assert placeholders[0].acquired is False
         assert placeholders[-1].acquired is True
+    -

Projet

Général

Profil

Produits Entr'ouvert » Combo

0001-data-merge-assets-and-site-export-file-39425.patch