Produits Entr'ouvert » Passerelle

import os

import re

import csv

import unicodedata

from collections import OrderedDict

from pyexcel_ods import get_data as get_data_ods

from pyexcel_xls import get_data as get_data_xls

from django.utils.timezone import datetime, make_aware

from django.db import models, transaction

from django.core.exceptions import ValidationError

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

        code_cache[expr] = compile(expr, '<inline>', 'eval')

    return code_cache[expr]

def normalize(value):

    return unicodedata.normalize('NFKD', value).encode('ascii', 'ignore')

    slug = models.SlugField(_('Name (slug)'))

    label = models.CharField(_('Label'), max_length=100)

    description = models.TextField(_('Description'), blank=True)

    class Meta:

        ordering = ['slug']

class CsvDataSource(BaseResource):

        help_text=_('Supported file formats: csv, ods, xls, xlsx'))

    columns_keynames = models.CharField(

        max_length=256,

        return result

    def cache_data(self):

        titles = [t.strip() for t in self.columns_keynames.split(',')]

        indexes = [titles.index(t) for t in titles if t]

        caption = [titles[i] for i in indexes]

        options = {}

        for k, v in self._dialect_options.items():

                v = v.encode('ascii')

            options[k.encode('ascii')] = v

        return [smart_text(t.strip()) for t in self.columns_keynames.split(',')]

    @endpoint(perm='can_access', methods=['get'],

    def select(self, request, query_name, **kwargs):

        try:

            query = Query.objects.get(resource=self.id, slug=query_name)

        if order:

            generator = stream_expressions(order, data, kind='order')

            new_data = [(tuple(new_row), row) for new_row, row in generator]

            data = [row for key, row in new_data]

        distinct = query.get_list('distinct')

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

import json

import os

import pytest

from StringIO import StringIO

from django.core.files import File

from django.core.urlresolvers import reverse

from django.contrib.contenttypes.models import ContentType

from django.core.management import call_command

from passerelle.base.models import ApiUser, AccessRight

from test_manager import login, admin_user

import webtest

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

pytestmark = pytest.mark.django_db

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

def get_file_content(filename):

@pytest.fixture

def setup():

        assert TableRow.objects.filter(resource=csv).count() == len(csv.get_rows())

        obj_type = ContentType.objects.get_for_model(csv)

        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_default_column_keynames(setup, filetype):

    csvdata = CsvDataSource.objects.create(csv_file=File(StringIO(get_file_content(filetype)), filetype),

                                           sheet_name='Feuille2',

    assert 'id' in csvdata.columns_keynames

    assert 'text' in csvdata.columns_keynames

def test_sheet_name_error(setup, app, filetype, admin_user):

    csvdata, url = setup('field,,another_field,', filename=filetype, data=get_file_content(filetype))

    app = login(app)

    else:

        assert resp.status_code == 302

def test_unfiltered_data(client, setup, filetype):

    csvdata, url = setup('field,,another_field,', filename=filetype, data=get_file_content(filetype))

    resp = client.get(url)

        assert 'field' in item

        assert 'another_field' in item

def test_empty_file(client, setup):

    resp = client.get(url)

    result = parse_response(resp)

    assert len(result) == 0

def test_view_manage_page(setup, app, filetype, admin_user):

    csvdata, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

    app = login(app)

def test_good_filter_data(client, setup, filetype):

    filter_criteria = 'Zakia'

        assert 'text' in item

        assert filter_criteria in item['text']

def test_bad_filter_data(client, setup, filetype):

    filter_criteria = 'bad'

    csvdata, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

    result = parse_response(resp)

    assert len(result) == 0

def test_useless_filter_data(client, setup, filetype):

    csvdata, url = setup('id,,nom,prenom,sexe', filename=filetype, data=get_file_content(filetype))

    filters = {'text': 'Ali'}

    result = parse_response(resp)

    assert len(result) == 20

def test_columns_keynames_with_spaces(client, setup, filetype):

    csvdata, url = setup('id , , nom,text , ', filename=filetype, data=get_file_content(filetype))

    filters = {'text': 'Yaniss'}

    result = parse_response(resp)

    assert len(result) == 1

def test_skipped_header_data(client, setup, filetype):

    csvdata, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype), skip_header=True)

    filters = {'q': 'Eliot'}

    result = parse_response(resp)

    assert len(result) == 0

def test_data(client, setup, filetype):

    csvdata, url = setup('fam,id,, text,sexe ', filename=filetype, data=get_file_content(filetype))

    filters = {'text': 'Sacha'}

    assert result[0] == {'id': '59', 'text': 'Sacha',

                         'fam': '2431', 'sexe': 'H'}

def test_unicode_filter_data(client, setup, filetype):

    filter_criteria = u'Benoît'

    csvdata, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

        assert 'text' in item

        assert filter_criteria in item['text']

def test_unicode_case_insensitive_filter_data(client, setup, filetype):

    csvdata, url = setup(',id,,text,', filename=filetype, data=get_file_content(filetype))

    filter_criteria = u'anaëlle'

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert len(result)

        assert 'text' in item

        assert filter_criteria.lower() in item['text'].lower()

def test_data_bom(client, setup):

    csvdata, url = setup('fam,id,, text,sexe ', data=data_bom)

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert result[0] == {'id': '69981', 'text': 'Eliot', 'fam': '121', 'sexe': 'H'}

def test_multi_filter(client, setup, filetype):

    csvdata, url = setup('fam,id,, text,sexe ', filename=filetype, data=get_file_content(filetype))

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert result[0] == {'id': '6', 'text': 'Shanone',

                         'fam': '525', 'sexe': 'F'}

    assert len(result) == 10

def test_query(client, setup, filetype):

    csvdata, url = setup('fam,id,, text,sexe ', filename=filetype, data=get_file_content(filetype))

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert result[0]['text'] == 'Eliot'

    assert len(result) == 1

def test_query_insensitive_and_unicode(client, setup, filetype):

    csvdata, url = setup('fam,id,, text,sexe ', filename=filetype, data=get_file_content(filetype))

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert result[0]['text'] == 'Eliot'

    assert result[0]['text'] == 'Eliot'

    assert len(result) == 1

def test_query_insensitive_and_filter(client, setup, filetype):

    csvdata, url = setup('fam,id,,text,sexe', filename=filetype, data=get_file_content(filetype))

    resp = client.get(url, filters)

    result = parse_response(resp)

    assert result[0]['text'] == 'Eliot'

    assert result[0]['id'] == '22'

    assert result[0]['lname'] == 'MARTIN'

def test_on_the_fly_dialect_detection(client, setup):

    # fake a connector that was not initialized during .save(), because it's

    # been migrated and we didn't do dialect detection at save() time.

    assert result['err'] == 0

    assert len(result['data']) == 20

def test_missing_columns(client, setup):

    csvdata, url = setup(data=data + 'A;B;C\n')

    resp = client.get(url)

    assert len(result['data']) == 21

    assert result['data'][-1] == {'lname': 'C', 'sex': None, 'id': 'B', 'fname': None, 'fam': 'A'}

def test_unknown_sheet_name(client, setup):

    csvdata, url = setup('field,,another_field,', filename='data2.ods', data=get_file_content('data2.ods'))

    csvdata.sheet_name = 'unknown'

    result = json.loads(resp.content)

    assert len(result['data']) == 20

def test_cache_new_shorter_file(client, setup):

    csvdata, url = setup(data=data + 'A;B;C\n')

    resp = client.get(url)

    result = json.loads(resp.content)

    assert len(result['data']) == 20

def test_query_array(app, setup, filetype):

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

    url = reverse('generic-endpoint', kwargs={

    for row in response.json['data']:

        assert len(row) == 2

        assert isinstance(row[0], int)

def test_query_q_filter(app, setup, filetype):

    for row in response.json['data']:

        assert len(row) == 2

        assert isinstance(row['id'], int)

def test_query_tuples(app, setup, filetype):

        assert row[0][0] == 'id'

        assert isinstance(row[0][1], int)

        assert row[1][0] == 'prenom'

def test_query_onerow(app, setup, filetype):

    assert isinstance(response.json['data'], list)

    assert len(response.json['data']) == 2

def test_query_keyed_distinct(app, setup, filetype):

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

    url = reverse('generic-endpoint', kwargs={

    assert isinstance(response.json['data'], dict)

    assert response.json['data']['MARTIN']['prenom'] == 'Sandra'

def test_query_order(app, setup, filetype):

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

    url = reverse('generic-endpoint', kwargs={

    assert response.json['data']['expr'] == 'zob'

    assert 'row' in response.json['data']

def test_edit_connector_queries(admin_user, app, setup, filetype):

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

    url = reverse('view-connector', kwargs={'connector': 'csvdatasource', 'slug': csvdata.slug})

    resp = app.get(url)

    new_query_url = reverse('csv-new-query', kwargs={'connector_slug': csvdata.slug})

    resp = app.get(new_query_url, status=302)

    assert resp.location.endswith('/login/?next=%s' % new_query_url)

    assert 'Syntax error' in resp.body

    resp.form['projections'] = 'id:id\nprenom:prenom'

    resp = resp.form.submit().follow()

    resp.form['filters'] = 'int(id) == 511'

    resp.form['description'] = 'in the sky without diamonds'

    resp = resp.form.submit().follow()

    assert 'Lucy alone' in resp.body

    assert 'in the sky without diamonds' in resp.body

    assert len(resp.json['data']) == 1

    assert resp.json['data'][0]['prenom'] == 'Lucie'

def test_download_file(app, setup, filetype, admin_user):

    csvdata, url = setup('field,,another_field,', filename=filetype, data=get_file_content(filetype))

    assert '/login' in app.get('/manage/csvdatasource/test/download/').location

    assert response.json['err'] == 0

    assert len(response.json['data']) == 2

def test_delete_connector_query(admin_user, app, setup, filetype):

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

    url = reverse('view-connector', kwargs={'connector': 'csvdatasource', 'slug': csvdata.slug})