Produits Entr'ouvert » Passerelle

# 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 csv

import datetime

import os

import re

import tempfile

from collections import OrderedDict

import pytz

import six

from django.conf import settings

from django.contrib.postgres.fields import JSONField

from django.core.exceptions import ValidationError

from django.db import models, transaction

from django.urls import reverse

from django.utils.encoding import force_str, force_text, smart_text

from django.utils.timezone import make_aware, now

from django.utils.translation import ugettext_lazy as _

from pyexcel_ods import get_data as get_data_ods

from pyexcel_xls import get_data as get_data_xls

    _dialect_options = JSONField(editable=False, null=True)

    sheet_name = models.CharField(_('Sheet name'), blank=True, max_length=150)

    category = _('Data Sources')

    documentation_url = (

        'https://doc-publik.entrouvert.com/admin-fonctionnel/parametrage-avance/source-de-donnees-csv/'

    )

    class Meta:

        verbose_name = _('Spreadsheet File')

    def clean(self, *args, **kwargs):

        file_type = self.csv_file.name.split('.')[-1]

        if file_type in ('ods', 'xls', 'xlsx') and not self.sheet_name:

            raise ValidationError(_('You must specify a sheet name'))

        if file_type not in ('ods', 'xls', 'xlsx'):

                # remove

                os.unlink(filepath)

            else:

                # move file in unused-files dir

                unused_dir = os.path.join(base_dir, 'unused-files')

                os.makedirs(unused_dir, exist_ok=True)

                os.rename(filepath, os.path.join(unused_dir, filename))

class TableRow(models.Model):

    resource = models.ForeignKey('CsvDataSource', on_delete=models.CASCADE)

    line_number = models.IntegerField(null=False)

    data = JSONField(blank=True)

    class Meta:

        ordering = ('line_number',)

  {% trans "File:" %}

  {% if object|can_edit:request.user %}<a href="{% url 'csv-download' connector_slug=object.slug %}">{{object.csv_file}}</a>

  — {% trans "added on" %} {{object.csv_file_datetime}}

  {% else %}{{object.csv_file}}{% endif %}

</p>

{% endblock %}

{% block endpoints %}

<ul class="endpoints">

{% url 'generic-endpoint' connector='csvdatasource' slug=object.slug endpoint='data' as csvdatasource_data_url %}

<li class="get-method">

  <div class="description"><span class="description--label">{% trans "Returning all file lines:" %}</span>

  <a class="example-url" href="{{ csvdatasource_data_url }}">{{ csvdatasource_data_url }}</a></div>

</li>

<li class="get-method">

  <div class="description"><span class="description--label">{% trans "Returning lines containing 'abc' in 'text' column (case insensitive):" %}</span>

</li>

{% for query in object.queries.all %}

<li class="get-method"><div class="description"><span class="description--label">{{ query.label }}:</span>

  <a class="example-url" href="{% url 'generic-endpoint' connector='csvdatasource' slug=object.slug endpoint='query' rest=query.slug %}/">{% url 'generic-endpoint' connector='csvdatasource' slug=object.slug endpoint='query' rest=query.slug %}/</a></div>

   {% if query.description %}— {{ query.description }}{% endif %}

</li>

{% endfor %}

</ul>

{% endblock %}

from stat import ST_MTIME

import mock

import pytest

import six

import webtest

from django.contrib.contenttypes.models import ContentType

from django.core.files import File

from django.core.management import call_command

from django.test import Client, override_settings

from django.urls import reverse

from django.utils.encoding import force_bytes, force_str, force_text

from django.utils.six import StringIO

from django.utils.six.moves.urllib.parse import urlencode

from django.utils.timezone import now

from test_manager import login

from passerelle.base.models import AccessRight, ApiUser

from passerelle.compat import json_loads

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

    csv.refresh_from_db()

    assert list(csv.get_rows()) == []

    call_command('change-csv', 'test', os.path.join(TEST_BASE_DIR, 'data.ods'), sheet_name='Feuille2')

    csv.refresh_from_db()

    assert list(csv.get_rows()) != []

@pytest.mark.parametrize(

    'payload,expected',

    [

        ({}, 20),

        ({'limit': 10}, 10),

        ({'limit': 10, 'offset': 0}, 10),

        ({'limit': 10, 'offset': 15}, 5),

        ({'limit': 10, 'offset': 42}, 0),

    private = ApiUser.objects.create(username='private', fullname='private', keytype='', key='xxx')

    public = ApiUser.objects.create(username='public', fullname='private', keytype='', key='')

    assert AccessRight.objects.count() == 0

    # adding private api user works

    app = login(app)

    resp = app.get(csv.get_absolute_url())

    assert 'Access is limited' in resp.html.find('div', {'id': 'security'}).div.text.strip()

    resp.form['apiuser'] = private.pk

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

    assert AccessRight.objects.count() == 1

    # adding public user displays a warning

    resp.form['apiuser'] = public.pk

    resp = resp.form.submit()

    assert AccessRight.objects.count() == 1

    assert 'user has no security' in resp.text

    resp = resp.form.submit()

    assert AccessRight.objects.count() == 1

    assert 'user has no security' in resp.text