Produits Entr'ouvert » Passerelle

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021  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/>.

from django.contrib import admin

from passerelle.apps.sector.models import Sector, Sectorization

class SectorAdmin(admin.ModelAdmin):

    prepopulated_fields = {'slug': ('title',)}

    list_display = ('title', 'slug', 'resource')

    list_filter = ('resource',)

class SectorizationAdmin(admin.ModelAdmin):

    list_display = (

        'id',

        'street_id',

        'parity',

        'min_housenumber',

        'max_housenumber',

        'sector',

        'resource',

    )

    list_filter = ('sector__resource',)

admin.site.register(Sector, SectorAdmin)

admin.site.register(Sectorization, SectorizationAdmin)

# Generated by Django 2.2.19 on 2021-08-15 19:40

import django.db.models.deletion

from django.db import migrations, models

import passerelle.apps.sector.models

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0029_auto_20210202_1627'),

    ]

    operations = [

        migrations.CreateModel(

            name='Sector',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=256, verbose_name='Title')),

                ('slug', models.CharField(max_length=128, verbose_name='Identifier')),

            ],

            options={

                'ordering': ['resource', 'slug'],

            },

        ),

        migrations.CreateModel(

            name='SectorResource',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                (

                    'csv_file',

                    models.FileField(

                        help_text='CSV file',

                        upload_to=passerelle.apps.sector.models.upload_to,

                        verbose_name='Sectorization file',

                    ),

                ),

                (

                    'titles_in_first_line',

                    models.BooleanField(

                        default=True,

                        help_text='If not, column titles are: street_id,parity,min_housenumber,max_housenumber,sector_id,sector_name, …',

                        verbose_name='First line defines column titles',

                    ),

                ),

                (

                    'users',

                    models.ManyToManyField(

                        blank=True,

                        related_name='_sectorresource_users_+',

                        related_query_name='+',

                        to='base.ApiUser',

                    ),

                ),

            ],

            options={

                'abstract': False,

            },

        ),

        migrations.CreateModel(

            name='Sectorization',

            fields=[

                (

                    'id',

                    models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),

                ),

                ('street_id', models.CharField(max_length=64, verbose_name='Street Identifier')),

                (

                    'parity',

                    models.SmallIntegerField(

                        choices=[(0, 'all'), (1, 'odd'), (2, 'even')],

                        default=0,

                        verbose_name='Parity of numbers',

                    ),

                ),

                (

                    'min_housenumber',

                    models.PositiveIntegerField(default=0, verbose_name='Minimal house number'),

                ),

                (

                    'max_housenumber',

                    models.PositiveIntegerField(default=999999, verbose_name='Maximal house number'),

                ),

                (

                    'sector',

                    models.ForeignKey(

                        on_delete=django.db.models.deletion.CASCADE, to='sector.Sector', verbose_name='Sector'

                    ),

                ),

            ],

            options={

                'ordering': ['street_id', 'min_housenumber', 'parity'],

            },

        ),

        migrations.AddField(

            model_name='sector',

            name='resource',

            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='sector.SectorResource'),

        ),

        migrations.AlterUniqueTogether(

            name='sector',

            unique_together={('resource', 'slug')},

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021  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 csv

import datetime

import mimetypes

import os

from django.conf import settings

from django.core.exceptions import ValidationError

from django.core.files.base import ContentFile

from django.db import models, transaction

from django.db.models import Q

from django.http import HttpResponse

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

from django.utils.timezone import now

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.jsonresponse import APIError

PARITY_ALL = 0

PARITY_ODD = 1

PARITY_EVEN = 2

PARITY_CHOICES = (

    (PARITY_ALL, _('all')),

    (PARITY_ODD, _('odd')),

    (PARITY_EVEN, _('even')),

)

CSV_TITLES = ['street_id', 'parity', 'min_housenumber', 'max_housenumber', 'sector_id', 'sector_name']

MAX_HOUSENUMBER = 999_999

def upload_to(instance, filename):

    return '%s/%s/%s' % (instance.get_connector_slug(), instance.slug, filename)

class SectorResource(BaseResource):

    csv_file = models.FileField(

        _('Sectorization file'),

        upload_to=upload_to,

        help_text=_('CSV file'),

    )

    titles_in_first_line = models.BooleanField(

        _('First line defines column titles'),

        default=True,

        help_text=_('If not, column titles are: %s, …') % ','.join(CSV_TITLES),

    )

    category = _('Geographic information system')

    hide_description_fields = ['csv_file']

    def __str__(self):

        return '%s [%s]' % (self.title, self.slug)

    def daily(self):

        super().daily()

        self.clean_old_csv_files()

    def clean_old_csv_files(self):

        if not os.path.exists(self.csv_file.path):

            return

        base_dir = os.path.dirname(self.csv_file.path)

        if os.path.dirname(self.csv_file.name) != os.path.join(self.get_connector_slug(), self.slug):

            # path is not compliant with upload_to, do nothing

            return

        for filename in os.listdir(base_dir):

            filepath = os.path.join(base_dir, filename)

            if not os.path.isfile(filepath):

                continue

            if os.path.basename(self.csv_file.name) == filename:

                # current file

                continue

            mtime = os.stat(filepath).st_mtime

            if mtime > (now() + datetime.timedelta(days=-7)).timestamp():

                # too young

                continue

            if getattr(settings, 'SECTOR_REMOVE_ON_CLEAN', False) is True:

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

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

        try:

            self.import_csv(validate_only=True)

        except Exception as e:

            raise ValidationError(_('Invalid CSV file: %s') % e)

        return super().clean(*args, **kwargs)

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

        import_csv = kwargs.pop('import_csv', True)

        result = super().save(*args, **kwargs)

        if import_csv:

            self.import_csv()

        return result

    @transaction.atomic

    def import_csv(self, validate_only=False):

        try:

            self.csv_file.seek(0)

            content = force_bytes(self.csv_file.read())

            content = force_str(content.decode('utf-8-sig', 'ignore').encode('utf-8'))  # handle BOM

            dialect = csv.Sniffer().sniff(content)

            reader = csv.reader(content.splitlines(), dialect)

        except Exception as e:

            raise ValidationError(_('failed to read CSV (%s)') % e)

        if self.titles_in_first_line:

            first_line = next(reader)

            titles = [name.strip().lower() for name in first_line]

            if not set(CSV_TITLES).issubset(titles):

                raise ValidationError(

                    _('missing column(s) in header: %s.') % ', '.join(set(CSV_TITLES) - set(titles))

                )

        else:

            titles = CSV_TITLES

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

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

        # now ready to import data, first delete all sectors

        # sectorization will also be delete (cascade)

        # (will be cancelled by transaction on any import error)

        if not validate_only:

            self.sector_set.all().delete()

        def get_cell(row, index):

            try:

                return row[index]

            except IndexError:

                return ''

        sector_id = sector_name = None

        sectors = {}

        for row in reader:

            if not row:

                continue  # do not consider empty lines

            row = [smart_text(x).strip() for x in row]

            row = {caption: get_cell(row, index) for caption, index in zip(captions, indexes)}

            if row['sector_id']:

                sector_id = row['sector_id']

                sector_name = row['sector_name'] or row['sector_id']

            if not sector_id:

                raise ValidationError(_('missing sector_id, line %s') % reader.line_num)

            if not row['street_id']:

                raise ValidationError(_('missing street_id, line %s') % reader.line_num)

            if not row['parity']:

                parity = PARITY_ALL

            elif row['parity'].lower()[0] in (str(PARITY_EVEN), 'e', 'p'):  # p = pair (even, in french)

                parity = PARITY_EVEN

            elif row['parity'].lower()[0] in (str(PARITY_ODD), 'o', 'i'):  # i = impair (odd, in french)

                parity = PARITY_ODD

            else:

                parity = PARITY_ALL

            try:

                min_housenumber = int(row['min_housenumber'])

            except (ValueError, TypeError):

                min_housenumber = 0

            try:

                max_housenumber = int(row['max_housenumber'])

            except (ValueError, TypeError):

                max_housenumber = MAX_HOUSENUMBER

            if not validate_only:

                if sector_id not in sectors:

                    sectors[sector_id] = self.sector_set.create(

                        resource=self, slug=sector_id, title=sector_name

                    )

                sectors[sector_id].sectorization_set.create(

                    sector=sectors[sector_id],

                    street_id=row['street_id'],

                    parity=parity,

                    min_housenumber=min_housenumber,

                    max_housenumber=max_housenumber,

                )

    @endpoint(

        description=_('Update sectorization with a CSV file'),

        display_category=_('Management'),

        perm='can_access',

        methods=['put'],

    )

    def update(self, request):

        ext = mimetypes.guess_extension(request.content_type)

        if not ext:

            raise APIError(

                "can't guess filename extension for '%s' content type" % request.content_type, http_status=400

            )

        name = self.csv_file.storage.get_available_name('api-uploaded-file' + ext)

        self.csv_file = ContentFile(content=request.body, name=name)

        try:

            self.clean()

        except ValidationError as e:

            raise APIError(e, http_status=400)

        self.save()

        return {

            'updated': self.csv_file.name,

            'data': [

                {

                    'id': sector.slug,

                    'text': sector.title,

                }

                for sector in self.sector_set.all()

            ],

        }

    @endpoint(

        description=_('Get sectorization as a CSV file'),

        display_category=_('Management'),

        perm='can_access',

        parameters={

            'even': {

                'description': _('Even numbers indicator (default: %s)') % PARITY_EVEN,

                'example_value': 'P',

            },

            'odd': {

                'description': _('Odd numbers indicator (default: %s)') % PARITY_ODD,

                'example_value': 'I',

            },

            'mix': {

                'description': _('Even or odd numbers indicator (default: %s)') % PARITY_ALL,

                'example_value': '',

            },

            'limits': {'description': _('Show housenumber min/max (0/%s)') % MAX_HOUSENUMBER, 'type': 'bool'},

            'repeat': {'description': _('Repeat sector id and name on all lines'), 'type': 'bool'},

        },

    )

    def export(self, request, even=None, odd=None, mix=None, limits=False, repeat=False):

        response = HttpResponse(content_type='text/csv')

        date = now().strftime('%Y-%m-%d_%H:%M')

        response['Content-Disposition'] = 'attachment; filename="sector-%s-%s.csv"' % (self.slug, date)

        writer = csv.writer(response, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)

        if self.titles_in_first_line:

            writer.writerow(CSV_TITLES)

        parity = dict(PARITY_CHOICES)

        if even is not None:

            parity[PARITY_EVEN] = even

        if odd is not None:

            parity[PARITY_ODD] = odd

        if mix is not None:

            parity[PARITY_ALL] = mix

        for sector in self.sector_set.all().order_by('slug'):

            first = True

            for sectorization in sector.sectorization_set.all().order_by('street_id'):

                writer.writerow(

                    [

                        sectorization.street_id,

                        parity.get(sectorization.parity, mix),

                        sectorization.min_housenumber if (limits or sectorization.min_housenumber) else '',

                        sectorization.max_housenumber

                        if (limits or sectorization.max_housenumber < MAX_HOUSENUMBER)

                        else '',

                        sector.slug if (repeat or first) else '',

                        sector.title if (repeat or first) else '',

                    ]

                )

                first = False

        return response

    @endpoint(

        name='sectors',

        description=_('List of Sectors'),

        perm='can_access',

        display_category=_('Data sources'),

        parameters={

            'id': {'description': _('Sector identifier (slug)')},

            'q': {'description': _('Filter by Sector Title or Identifier')},

            'street_id': {'description': _('Get sectors for this Street identifier')},

            'house_number': {

                'description': _('Get sectors by this House Number (requires a street_id)'),

                'type': 'integer',

            },

        },

    )

    def sectors(self, request, q=None, id=None, street_id=None, house_number=None):

        if house_number and not street_id:

            raise APIError('house_number requires a street_id', http_status=400)

        # search by street and house number

        if street_id:

            query = Sectorization.objects.filter(sector__resource=self, street_id=street_id)

            if house_number is not None:

                house_number = int(house_number)

                query = query.filter(min_housenumber__lte=house_number, max_housenumber__gte=house_number)

                parity = PARITY_ODD if house_number % 2 else PARITY_EVEN

                query = query.filter(Q(parity=PARITY_ALL) | Q(parity=parity))

            else:

                query = query.filter(parity=PARITY_ALL, min_housenumber=0, max_housenumber=MAX_HOUSENUMBER)

            return {

                'data': [

                    {

                        'id': sectorization.sector.slug,

                        'text': sectorization.sector.title,

                    }

                ]

                for sectorization in query.reverse()

            }

        # list of sectors

        query = self.sector_set.all()

        if id is not None:

            query = query.filter(slug=id)

        elif q is not None:

            query = query.filter(Q(slug__icontains=q) | Q(title__icontains=q))

        return {

            'data': [

                {

                    'id': sector.slug,

                    'text': sector.title,

                }

                for sector in query

            ]

        }

class Sector(models.Model):

    resource = models.ForeignKey(SectorResource, on_delete=models.CASCADE)

    title = models.CharField(max_length=256, verbose_name=_('Title'))

    slug = models.CharField(max_length=128, verbose_name=_('Identifier'))

    class Meta:

        ordering = ['resource', 'slug']

        unique_together = ('resource', 'slug')

    def __str__(self):

        return '%s > %s [%s]' % (self.resource, self.title, self.slug)

class Sectorization(models.Model):

    sector = models.ForeignKey(Sector, on_delete=models.CASCADE, verbose_name=_('Sector'))

    street_id = models.CharField(max_length=64, verbose_name=_('Street Identifier'))

    parity = models.SmallIntegerField(

        choices=PARITY_CHOICES, default=PARITY_ALL, verbose_name=_('Parity of numbers')

    )

    min_housenumber = models.PositiveIntegerField(default=0, verbose_name=_('Minimal house number'))

    max_housenumber = models.PositiveIntegerField(

        default=MAX_HOUSENUMBER, verbose_name=_('Maximal house number')

    )

    class Meta:

        ordering = ['street_id', 'min_housenumber', 'parity']

    def __str__(self):

        return '%s, parity:%s, min:%s, max:%s → %s' % (

            self.street_id,

            dict(PARITY_CHOICES).get(self.parity),

            self.min_housenumber,

            self.max_housenumber,

            self.sector,

        )

    @property

    def resource(self):

        return self.sector.resource

    def clean(self):

        if not self.max_housenumber or self.max_housenumber > MAX_HOUSENUMBER:

            self.max_housenumber = MAX_HOUSENUMBER

        if self.min_housenumber > self.max_housenumber:

            raise ValidationError(_('Minimal house number may not be lesser than maximal house number.'))

    'passerelle.apps.sector',

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2021 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 datetime

import os

from posix import stat_result

from stat import ST_MTIME

import pytest

import utils

import webtest

from django.core.exceptions import ValidationError

from django.core.files import File

from django.core.management import call_command

from django.urls import reverse

from django.utils.encoding import force_str, force_text

from django.utils.six import StringIO

from django.utils.timezone import now

from test_manager import login

from passerelle.apps.sector.models import Sectorization, SectorResource

from passerelle.base.models import AccessRight

CSV = """street_id,parity,min_housenumber,max_housenumber,sector_id,sector_name

75114_1912,P,,, gs-moulin, Groupe Scolaire Moulin

75114_1912,I,0,999999,gs-zola,Groupe Scolaire Zola

75114_1913,N,0,999999,ecole-hugo,École Hugo

75114_1914,,,10,ecole-hugo, École Hugo

75114_1914,,11,, ecole-hugo2, École Hugo 2

75114_1915,,,,ecole-hugo2 , École Hugo 2

"""

CSV_BOM = force_str(force_text(CSV, 'utf-8').encode('utf-8-sig'))

CSV_NO_FIRST_LINE = """75114_1912,P,,, gs-moulin, Groupe Scolaire Moulin

75114_1912,I,0,999999,gs-zola,Groupe Scolaire Zola

75114_1913,N,0,999999,ecole-hugo,École Hugo

75114_1914,,,10,ecole-hugo, École Hugo

75114_1914,,11,, ecole-hugo, École Hugo"""

CSV_REORDERED = """sector_id,sector_name,street_id,parity,min_housenumber,max_housenumber,foo,bar

gs-moulin, Groupe Scolaire Moulin, 75114_1912,P,,,aaa,bbb

gs-zola,Groupe Scolaire Zola,75114_1912,I,0,999999,xxx,yyy

ecole-hugo,École Hugo,75114_1913,N,0,999999,000,1

,,75114_1999,N,0,999999,,

"""

CSV_MISSING_COLUMN = """street_id,min_housenumber,max_housenumber,sector_id,sector_name

75114_1912,,,foo,

,0,999999,gs-zola,Groupe Scolaire Zola"""

CSV_MISSING_SECTOR = """street_id,parity,min_housenumber,max_housenumber,sector_id,sector_name

75114_1912,P,,, ,

75114_1912,I,0,999999,gs-zola,Groupe Scolaire Zola"""

CSV_MISSING_STREET = """street_id,parity,min_housenumber,max_housenumber,sector_id,sector_name

75114_1912,P,,,foo,

,I,0,999999,gs-zola,Groupe Scolaire Zola"""

pytestmark = pytest.mark.django_db

@pytest.fixture

def sector(db):

    return utils.setup_access_rights(

        SectorResource.objects.create(

            slug='test',

            title='title',

            csv_file=File(StringIO(CSV), 'sectorization.csv'),

        )

    )

def test_sector_creation(sector):

    assert '%s' % sector == 'title [test]'

    assert sector.sector_set.count() == 4

    hugo = sector.sector_set.get(slug='ecole-hugo')

    assert Sectorization.objects.filter(sector=hugo).count() == 2

    hugo2 = sector.sector_set.get(slug='gs-zola')

    assert Sectorization.objects.filter(sector=hugo2).count() == 1

    sector.clean()

    sector.save()

    assert sector.sector_set.count() == 4  # no change

    # forced reset

    sector.sector_set.all().delete()

    assert sector.sector_set.count() == 0

    sector.save()

    assert sector.sector_set.count() == 4

def test_sector_creation_bom(sector):

    # forced reset

    sector.sector_set.all().delete()

    assert sector.sector_set.count() == 0

    sector.csv_file = File(StringIO(CSV_BOM), 'sectorization.csv')

    sector.clean()

    sector.save()

    assert sector.sector_set.count() == 4

def test_sector_creation_nofirstline(sector):

    sector.csv_file = File(StringIO(CSV_NO_FIRST_LINE), 'sectorization.csv')

    with pytest.raises(ValidationError, match='Invalid CSV file:.*missing column'):

        sector.clean()

    assert sector.sector_set.count() == 4  # nothing change from initial creation

    sector.titles_in_first_line = False

    sector.save()

    assert sector.sector_set.count() == 3

def test_sector_reordered(sector):

    assert sector.sector_set.count() == 4

    sector.csv_file = File(StringIO(CSV_REORDERED), 'sectorization.csv')

    sector.save()

    assert sector.sector_set.count() == 3

def test_sector_empty_file(sector):

    sector.csv_file = File(StringIO(''), 'sectorization.csv')

    with pytest.raises(ValidationError, match='Invalid CSV file:.*failed to read CSV'):

        sector.clean()

    with pytest.raises(ValidationError, match='failed to read CSV'):

        sector.save()

    assert sector.sector_set.count() == 4  # nothing change

def test_sector_missing_sector(sector):

    sector.csv_file = File(StringIO(CSV_MISSING_SECTOR), 'sectorization.csv')

    with pytest.raises(ValidationError, match='Invalid CSV file:.*missing sector_id, line 2'):

        sector.clean()

    with pytest.raises(ValidationError, match='missing sector_id, line 2'):

        sector.save()

    assert sector.sector_set.count() == 4  # nothing change

def test_sector_missing_street(sector):

    sector.csv_file = File(StringIO(CSV_MISSING_STREET), 'sectorization.csv')

    with pytest.raises(ValidationError, match='Invalid CSV file:.*missing street_id, line 3'):

        sector.clean()

    with pytest.raises(ValidationError, match='missing street_id, line 3'):

        sector.save()

    assert sector.sector_set.count() == 4  # nothing change

def test_sector_missing_column(sector):

    sector.csv_file = File(StringIO(CSV_MISSING_COLUMN), 'sectorization.csv')

    with pytest.raises(ValidationError, match='Invalid CSV file:.*missing column.*: parity\.'):

        sector.clean()

    with pytest.raises(ValidationError, match='missing column.*: parity\.'):

        sector.save()

    assert sector.sector_set.count() == 4  # nothing change

def test_sector_endpoint_sectors(app, sector):

    url = reverse(

        'generic-endpoint',

        kwargs={

            'connector': 'sector',

            'slug': sector.slug,

            'endpoint': 'sectors',

        },

    )

    result = app.get(url).json

    assert result['err'] == 0

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

    assert {'id': 'ecole-hugo', 'text': 'École Hugo'} in result['data']

    result = app.get(url, params={'id': 'ecole-hugo'}).json

    assert result['err'] == 0

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

    assert {'id': 'ecole-hugo', 'text': 'École Hugo'} in result['data']

    result = app.get(url, params={'q': 'hugo'}).json

    assert result['err'] == 0

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

    assert {'id': 'ecole-hugo', 'text': 'École Hugo'} in result['data']

    assert {'id': 'ecole-hugo2', 'text': 'École Hugo 2'} in result['data']

    result = app.get(url, params={'q': 'foobar'}).json

    assert result['err'] == 0

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

    # search a sector by street and house number

    result = app.get(url, params={'street_id': '75114_1915'}).json

    assert result['err'] == 0

    assert result['data'] == [{'id': 'ecole-hugo2', 'text': 'École Hugo 2'}]

    result = app.get(url, params={'street_id': '75114_1915', 'house_number': '123'}).json

    assert result['err'] == 0

    assert result['data'] == [{'id': 'ecole-hugo2', 'text': 'École Hugo 2'}]

    result = app.get(url, params={'street_id': '75114_1912', 'house_number': '12'}).json  # even

    assert result['err'] == 0

    assert result['data'] == [{'id': 'gs-moulin', 'text': 'Groupe Scolaire Moulin'}]

    result = app.get(url, params={'street_id': '75114_1912', 'house_number': '13'}).json  # odd

    assert result['err'] == 0

    assert result['data'] == [{'id': 'gs-zola', 'text': 'Groupe Scolaire Zola'}]

    result = app.get(url, params={'street_id': '75114_1914', 'house_number': '5'}).json  # 5 <= 10

    assert result['err'] == 0

    assert result['data'] == [{'id': 'ecole-hugo', 'text': 'École Hugo'}]

    result = app.get(url, params={'street_id': '75114_1914', 'house_number': '20'}).json  # 20 >= 11

    assert result['err'] == 0

    assert result['data'] == [{'id': 'ecole-hugo2', 'text': 'École Hugo 2'}]

    # bad searches

    result = app.get(url, params={'street_id': '75114_1915', 'house_number': 'abc'}, status=400).json

    assert result['err'] == 1

    assert result['err_desc'] == 'invalid value for parameter "house_number"'  # not an integer

    result = app.get(url, params={'house_number': '123'}, status=400).json

    assert result['err'] == 1

    assert result['err_desc'] == 'house_number requires a street_id'

    # access right is needed

    AccessRight.objects.all().delete()

    result = app.get(url, status=403).json

    assert result['err'] == 1

    assert 'PermissionDenied' in result['err_class']

    assert result['data'] is None

def test_sector_endpoint_export(app, sector):

    url = reverse(

        'generic-endpoint',

        kwargs={

            'connector': 'sector',

            'slug': sector.slug,

            'endpoint': 'export',

        },

    )

    resp = app.get(url)

    assert resp.headers['content-type'] == 'text/csv'

    assert resp.text.startswith(

        '"street_id","parity","min_housenumber","max_housenumber","sector_id","sector_name"'

    )

    assert len(resp.text.splitlines()) == 7

    sector.titles_in_first_line = False

    sector.save(import_csv=False)

    resp = app.get(url)

    assert resp.headers['content-type'] == 'text/csv'

    assert 'street_id' not in resp.text

    assert len(resp.text.splitlines()) == 6

    sector.titles_in_first_line = True

    sector.save()

    resp = app.get(url)

    assert resp.headers['content-type'] == 'text/csv'

    # "street_id","parity","min_housenumber","max_housenumber","sector_id","sector_name"

    # "75114_1913","all","","","ecole-hugo","École Hugo"

    # "75114_1914","all","","10","",""

    # "75114_1915","all","","","ecole-hugo2","École Hugo 2"

    # "75114_1914","all","11","","",""

    # "75114_1912","odd","","","gs-zola","Groupe Scolaire Zola"

    # "75114_1912","even","","","gs-moulin","Groupe Scolaire Moulin"

    assert len(resp.text.splitlines()) == 7

    assert resp.text.count('"all"') == 4

    assert resp.text.count('"odd"') == 1

    assert resp.text.count('"even"') == 1

    assert resp.text.count('"ecole-hugo"') == 1

    assert resp.text.count('"0"') == 0

    assert resp.text.count('"999999"') == 0

    # import -> export again

    initial_export = resp.text

    sector.csv_file = File(StringIO(initial_export), 'data.csv')

    sector.save()

    resp = app.get(url)

    assert resp.text == initial_export

    # modify export format

    resp = app.get(

        url, params={'odd': 'IMPAIRS', 'even': 'PAIRS', 'mix': 'TOUS', 'repeat': 'true', 'limits': 'true'}

    )

    assert len(resp.text.splitlines()) == 7

    assert resp.text.count('"TOUS"') == 4

    assert resp.text.count('"IMPAIRS"') == 1

    assert resp.text.count('"PAIRS"') == 1

    assert resp.text.count('"ecole-hugo"') == 2  # repeat

    assert resp.text.count('"0"') == 5  # limits

    assert resp.text.count('"999999"') == 5

    # access right is needed

    AccessRight.objects.all().delete()

    result = app.get(url, status=403).json

    assert result['err'] == 1

    assert 'PermissionDenied' in result['err_class']

    assert result['data'] is None

def test_sector_endpoint_update(app, sector):

    url = reverse(

        'generic-endpoint',

        kwargs={

            'connector': 'sector',

            'slug': sector.slug,

            'endpoint': 'update',

        },

    )

    assert sector.sector_set.count() == 4

    result = app.put(url, params=CSV_REORDERED, headers={'Content-Type': 'text/csv'}).json

    assert sector.sector_set.count() == 3

    assert result['err'] == 0

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

    assert result['updated'] == 'sector/test/api-uploaded-file.csv'

    result = app.put(url, params=CSV_MISSING_COLUMN, headers={'Content-Type': 'text/csv'}, status=400).json

    assert result['err'] == 1

    assert "missing column" in result['err_desc']

    result = app.put(url, params=CSV_REORDERED, headers={}, status=400).json

    assert result['err'] == 1

    assert "can't guess filename extension" in result['err_desc']

    # access right is needed

    AccessRight.objects.all().delete()

    result = app.put(url, params=CSV_REORDERED, headers={'Content-Type': 'text/csv'}, status=403).json

    assert result['err'] == 1

    assert 'PermissionDenied' in result['err_class']

@pytest.mark.parametrize('remove_files', [False, True])

def test_daily_clean(settings, remove_files, sector):

    settings.SECTOR_REMOVE_ON_CLEAN = remove_files

    sectordata_dir = os.path.dirname(sector.csv_file.path)

    other_dir = os.path.join(settings.MEDIA_ROOT, 'foo', sector.slug)

    os.makedirs(other_dir)

    # create additional file in sector dir

    with open(os.path.join(sectordata_dir, 'csv-file.csv'), 'w'):

        pass

    os.makedirs(os.path.join(sectordata_dir, 'not-a-file'))

    # create additional file in other dir

    with open(os.path.join(other_dir, 'csv-file.csv'), 'w'):

        pass

    call_command('cron', 'daily')

    # not changed

    assert os.listdir(other_dir) == ['csv-file.csv']

    # too soon to be removed

    dir_list = os.listdir(sectordata_dir)

    dir_list.sort()

    assert dir_list == ['csv-file.csv', 'not-a-file', 'sectorization.csv']

    orig_os_stat = os.stat

    def _fake_stat(arg, delta):

        faked = list(orig_os_stat(arg))

        faked[ST_MTIME] = (now() + delta).timestamp()

        return stat_result(faked)

    try:

        # 1 week ago but one minute too soon

        os.stat = lambda arg: _fake_stat(arg, datetime.timedelta(days=-7, minutes=1))

        call_command('cron', 'daily')

        # not changed

        assert os.listdir(other_dir) == ['csv-file.csv']

        # still too soon to be removed

        dir_list = os.listdir(sectordata_dir)

        dir_list.sort()

        assert dir_list == ['csv-file.csv', 'not-a-file', 'sectorization.csv']

        # 1 week ago

        os.stat = lambda arg: _fake_stat(arg, datetime.timedelta(days=-7))

        call_command('cron', 'daily')

        # not changed

        assert os.listdir(other_dir) == ['csv-file.csv']

        # removed or moved

        dir_list = os.listdir(sectordata_dir)

        dir_list.sort()

        if remove_files:

            assert dir_list == ['not-a-file', 'sectorization.csv']

        else:

            assert dir_list == ['not-a-file', 'sectorization.csv', 'unused-files']

            assert os.listdir(os.path.join(sectordata_dir, 'unused-files')) == ['csv-file.csv']

        # wrong storage directory, do nothing

        with open(os.path.join(other_dir, 'bar.csv'), 'w'):

            pass

        sector.csv_file.name = 'foo/%s/csv-file.csv' % sector.slug

        sector.save()

        assert sorted(os.listdir(other_dir)) == ['bar.csv', 'csv-file.csv']

        call_command('cron', 'daily')

        assert sorted(os.listdir(other_dir)) == ['bar.csv', 'csv-file.csv']

        # unknown file

        sector.csv_file.name = 'sector/%s/bar.csv' % sector.slug

        sector.save()

        call_command('cron', 'daily')

    finally:

        os.stat = orig_os_stat

def test_sector_manage_create(app, admin_user):

    app = login(app)

    response = app.get(reverse('create-connector', kwargs={'connector': 'sector'}))

    response.form.set('title', 'test title')

    response.form.set('slug', 'test-slug')

    response.form.set('description', 'test description')

    response.form.set('csv_file', webtest.Upload('test.csv', CSV.encode('utf-8'), 'application/octet-stream'))

    response = response.form.submit()

    assert response.location

    response = response.follow()

    assert 'test title' in response

    assert 'test description' in response

    assert SectorResource.objects.count() == 1

    resource = SectorResource.objects.get()

    assert resource.title == 'test title'

    assert resource.slug == 'test-slug'

    assert resource.description == 'test description'

    assert resource.csv_file.read() == CSV.encode('utf-8')

    assert resource.sector_set.count() == 4