Produits Entr'ouvert » Passerelle

    def export_json(self):

        raise NotImplementedError

    def export_json(self):

        return {

            'slug': self.slug,

            'label': self.label,

            'description': self.description,

            'filters': self.filters,

            'projections': self.projections,

            'order': self.order,

            'distinct': self.distinct,

            'structure': self.structure,

        }

    @classmethod

    def import_json(cls, d):

        return cls(**d)

    def export_json(self):

        d = super(CsvDataSource, self).export_json()

        d['queries'] = [query.export_json() for query in Query.objects.filter(resource=self)]

        return d

    @classmethod

    def import_json_real(cls, d, **kwargs):

        queries = d.pop('queries', [])

        instance = super(CsvDataSource, cls).import_json_real(d, **kwargs)

        new = []

        for query in queries:

            q = Query.import_json(query)

            q.resource = instance

            new.append(q)

        Query.objects.bulk_create(new)

        return instance

import json

import sys

from django.core.management.base import BaseCommand

from passerelle.utils import export_site

class Command(BaseCommand):

    args = ''

    help = 'Export the site'

    def handle(self, *args, **options):

        json.dump(export_site(), sys.stdout, indent=4)

import json

from optparse import make_option

from django.core.management.base import BaseCommand

from passerelle.utils import import_site

class Command(BaseCommand):

    args = '<filename>'

    help = 'Import an exported site'

    option_list = BaseCommand.option_list + (

        make_option('--import-users', action='store_true', default=False,

                    help='Import users and access rights'),

    )

    def handle(self, filename, **options):

        import_site(json.load(open(filename)), import_users=options['import_users'])

import os

import base64

from django.apps import apps

from django.db import models, transaction

from django.core.files.base import ContentFile

from jsonfield import JSONField

    def export_json(self):

        return {

            '@type': 'passerelle-user',

            'username': self.username,

            'fullname': self.fullname,

            'description': self.description,

            'keytype': self.keytype,

            'key': self.key,

            'ipsource': self.ipsource,

        }

    @classmethod

    def import_json(self, d):

        if d.get('@type') != 'passerelle-user':

            raise ValueError('not a passerelle user export')

        d = d.copy()

        d.pop('@type')

        return self.objects.get_or_create(username=d['username'],

                                          defaults=d)

    def export_json(self):

        d = {

            '@type': 'passerelle-resource',

            'resource_type': '%s.%s' % (self.__class__._meta.app_label,

                                        self.__class__._meta.model_name),

            'title': self.title,

            'slug': self.slug,

            'description': self.description,

            'log_level': self.log_level,

            'access_rights': []

        }

        resource_type = ContentType.objects.get_for_model(self)

        for ar in AccessRight.objects.filter(resource_type=resource_type,

                                             resource_pk=self.pk).select_related():

            d['access_rights'].append({

                'codename': ar.codename,

                'apiuser': ar.apiuser.username,

            })

        for field, model in self.__class__._meta.get_concrete_fields_with_model():

            if field.name == 'id':

                continue

            if isinstance(field, (models.TextField, models.CharField, models.SlugField,

                                  models.URLField, models.BooleanField, models.IntegerField,

                                  models.CommaSeparatedIntegerField, models.EmailField,

                                  models.IntegerField, models.PositiveIntegerField, JSONField)):

                d[field.name] = getattr(self, field.attname)

            elif isinstance(field, models.FileField):

                value = getattr(self, field.attname)

                d[field.name] = {

                    'name': os.path.basename(value.name),

                    'content': base64.b64encode(value.read()),

                }

            else:

                raise Exception('export_json: field %s of ressource class %s is unsupported' % (

                    field, self.__class__))

        return d

    @staticmethod

    def import_json(d, import_users=False):

        if d.get('@type') != 'passerelle-resource':

            raise ValueError('not a passerelle resource export')

        d = d.copy()

        d.pop('@type')

        app_label, model_name = d['resource_type'].split('.')

        model = apps.get_model(app_label, model_name)

        try:

            return model.objects.get(slug=d['slug'])

        except model.DoesNotExist:

            pass

        with transaction.atomic():

            # prevent semi-creation of ressources

            instance = model.import_json_real(d)

            resource_type = ContentType.objects.get_for_model(instance)

            # We can only connect AccessRight objects to the new Resource after its creation

            if import_users:

                for ar in d['access_rights']:

                    apiuser = ApiUser.objects.get(username=ar['apiuser'])

                    AccessRight.objects.get_or_create(

                        codename=ar['codename'],

                        resource_type=resource_type,

                        resource_pk=instance.pk,

                        apiuser=apiuser)

        return instance

    @classmethod

    def import_json_real(cls, d, **kwargs):

        init_kwargs = {

            'title': d['title'],

            'slug': d['slug'],

            'description': d['description'],

            'log_level': d['log_level'],

        }

        init_kwargs.update(kwargs)

        instance = cls(**init_kwargs)

        for field, model in cls._meta.get_concrete_fields_with_model():

            if field.name == 'id':

                continue

            if isinstance(field, (models.TextField, models.CharField, models.SlugField,

                                  models.URLField, models.BooleanField, models.IntegerField,

                                  models.CommaSeparatedIntegerField, models.EmailField,

                                  models.IntegerField, models.PositiveIntegerField, JSONField)):

                setattr(instance, field.attname, d[field.name])

            elif isinstance(field, models.FileField):

                getattr(instance, field.attname).save(

                    d[field.name]['name'],

                    ContentFile(base64.b64decode(d[field.name]['content'])),

                    save=False)

            else:

                raise Exception('import_json_real: field %s of ressource class '

                                '%s is unsupported' % (field, cls))

        instance.save()

        return instance

    def export_json(self):

        raise NotImplementedError

from django.db import transaction

from passerelle.base.models import ApiUser, AccessRight, BaseResource

def export_site():

    '''Dump passerelle configuration (users, resources and ACLs) to JSON dumpable dictionnary'''

    d = {}

    d['apiusers'] = [apiuser.export_json() for apiuser in ApiUser.objects.all()]

    d['resources'] = resources = []

    for subclass in BaseResource.__subclasses__():

        if subclass._meta.abstract:

            continue

        for resource in subclass.objects.all():

            try:

                resources.append(resource.export_json())

            except NotImplementedError:

                break

    return d

def import_site(d, import_users=False):

    '''Load passerelle configuration (users, resources and ACLs) from a dictionnary loaded from

       JSON

    '''

    d = d.copy()

    with transaction.atomic():

        if import_users:

            for apiuser in d['apiusers']:

                ApiUser.import_json(apiuser)

        for resource in d['resources']:

            BaseResource.import_json(resource, import_users=import_users)

# -*- coding: utf-8 -*-

import sys

import json

import os

import pytest

from StringIO import StringIO

import tempfile

from django.core.wsgi import get_wsgi_application

from webtest import TestApp

from django.contrib.auth.models import User

from django.core.files import File

from django.core.urlresolvers import reverse

from django.core.management import call_command

from django.contrib.contenttypes.models import ContentType

from django.test import Client

from passerelle.base.models import ApiUser, AccessRight

from test_manager import login, admin_user

from passerelle.utils import import_site, export_site

data = """121;69981;DELANOUE;Eliot;H

525;6;DANIEL WILLIAMS;Shanone;F

253;67742;MARTIN;Sandra;F

511;38142;MARCHETTI;Lucie;F

235;22;MARTIN;Sandra;F

620;52156;ARNAUD;Mathis;H

902;36;BRIGAND;Coline;F

2179;48548;THEBAULT;Salima;F

3420;46;WILSON-LUZAYADIO;Anaëlle;F

1421;55486;WONE;Fadouma;F

2841;51;FIDJI;Zakia;F

2431;59;BELICARD;Sacha;H

4273;60;GOUBERT;Adrien;H

4049;64;MOVSESSIAN;Dimitri;H

4605;67;ABDOU BACAR;Kyle;H

4135;22231;SAVERIAS;Marius;H

4809;75;COROLLER;Maelys;F

5427;117;KANTE;Aliou;H

116642;118;ZAHMOUM;Yaniss;H

216352;38;Dupont;Benoît;H

"""

data_bom = data.decode('utf-8').encode('utf-8-sig')

from csvdatasource.models import CsvDataSource, Query

pytestmark = pytest.mark.django_db

TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'csvdatasource')

def get_file_content(filename):

    return file(os.path.join(TEST_BASE_DIR, filename)).read()

@pytest.fixture

def setup():

    def maker(columns_keynames='fam,id,lname,fname,sex', filename='data.csv', sheet_name='Feuille2',

              data=''):

        api = ApiUser.objects.create(username='all', keytype='', key='')

        csv = CsvDataSource.objects.create(csv_file=File(StringIO(data), filename),

                                           sheet_name=sheet_name, columns_keynames=columns_keynames,

                                           slug='test', title='a title',

                                           description='a description')

        obj_type = ContentType.objects.get_for_model(csv)

        AccessRight.objects.create(codename='can_access', apiuser=api, resource_type=obj_type,

                                   resource_pk=csv.pk)

        url = reverse('csvdatasource-data', kwargs={'slug': csv.slug})

        return csv, url

    return maker

def parse_response(response):

    return json.loads(response.content)['data']

@pytest.fixture

def client():

    return Client()

@pytest.fixture(params=['data.csv', 'data.ods', 'data.xls', 'data.xlsx'])

def filetype(request):

    return request.param

def test_export_csvdatasource(app, setup, filetype):

    def clear():

        Query.objects.all().delete()

        CsvDataSource.objects.all().delete()

        ApiUser.objects.all().delete()

    csvdata, url = setup('id,whatever,nom,prenom,sexe', filename=filetype,

                         data=get_file_content(filetype))

    query = Query(slug='query-1_', resource=csvdata, structure='array')

    query.projections = '\n'.join(['id:int(id)', 'prenom:prenom'])

    query.save()

    first = export_site()

    clear()

    import_site(first, import_users=True)

    second = export_site()

    # we don't care about file names

    first['resources'][0]['csv_file']['name'] = 'whocares'

    second['resources'][0]['csv_file']['name'] = 'whocares'

    assert first == second

    old_stdout = sys.stdout

    output = sys.stdout = StringIO()

    call_command('export_site')

    sys.stdout = old_stdout

    output = output.getvalue()

    third = json.loads(output)

    third['resources'][0]['csv_file']['name'] = 'whocares'

    assert first == third

    clear()

    with tempfile.NamedTemporaryFile() as f:

        f.write(output)

        f.flush()

        call_command('import_site', f.name, import_users=True)

    fourth = export_site()

    fourth['resources'][0]['csv_file']['name'] = 'whocares'

    assert first == fourth