Produits Entr'ouvert » Combo

# 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_files():

    media_prefix = default_storage.path('')

    for basedir, dirnames, filenames in os.walk(media_prefix):

        for filename in filenames:

            os.remove('%s/%s' % (basedir, filename))

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 untar_assets_files(tar, overwrite=False):

    media_prefix = default_storage.path('')

    data = {}

    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'))

        elif tarinfo.name != '_site.json':

            tar.extract(tarinfo, path=media_prefix)

    return data

def tar_assets_files(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))

def import_assets(fd, overwrite=False):

    tar = tarfile.open(mode='r', fileobj=fd)

    data = untar_assets_files(tar, overwrite=overwrite)

    Asset.load_serialized_objects(data.get('assets') or [])

    tar.close()

def export_assets(fd):

    tar = tarfile.open(mode='w', fileobj=fd)

    tar_assets_files(tar)

    tar.close()

from combo.apps.assets.utils import import_assets, export_assets

            import_assets(form.cleaned_data['assets_file'], overwrite)

    export_assets(fd)

import os

import tarfile

from django.core.files.storage import default_storage

from django.core.files import File

from django.utils.six import BytesIO

from combo.apps.assets.utils import (add_tar_content, clean_assets_files,

         export_assets, import_assets, untar_assets_files, tar_assets_files)

@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

    clean_assets_files()

def test_clean_assets_files(some_assets):

    assert count_asset_files() == 2

    clean_assets_files()

    assert count_asset_files() == 0

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_tar_untar_assets(some_assets):

    assert Asset.objects.count() == 2

    assert count_asset_files() == 2

    fd = BytesIO()

    tar = tarfile.open(mode='w', fileobj=fd)

    tar_assets_files(tar)

    tar_bytes = fd.getvalue()

    tar.close()

    path = default_storage.path('')

    os.remove('%s/assets/test.png' % path)

    open('%s/assets/test2.png' % path, 'w').write('foo')

    assert count_asset_files() == 1

    Asset.objects.all().delete()

    assert Asset.objects.count() == 0

    fd = BytesIO(tar_bytes)

    tar = tarfile.open(mode='r', fileobj=fd)

    data = untar_assets_files(tar)

    assert [x['fields']['key'] for x in data['assets']] == ['banner', 'favicon']

    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'

    clean_assets_files()

def test_import_export_assets(some_assets, tmpdir):

    filename = os.path.join(str(tmpdir), 'file.tar')

    assert Asset.objects.count() == 2

    assert count_asset_files() == 2

    fd = open(filename, 'wb')

    export_assets(fd)

    path = default_storage.path('')

    os.remove('%s/assets/test.png' % path)

    open('%s/assets/test2.png' % path, 'w').write('foo')

    assert count_asset_files() == 1

    Asset.objects.all().delete()

    assert Asset.objects.count() == 0

    fd = open(filename, 'rb')

    import_assets(fd, overwrite=True)

    assert count_asset_files() == 2

    assert open('%s/assets/test.png' % path, 'r').read() == 'test'

    assert open('%s/assets/test2.png' % path, 'r').read() == 'test2'

    clean_assets_files()

    assert count_asset_files() == 0

    clean_assets_files()