Produits Entr'ouvert » Passerelle

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

# Generated by Django 1.11.20 on 2019-09-19 15:23

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

import jsonfield.fields

import passerelle.utils.sftp

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0014_auto_20190820_0914'),

    ]

    operations = [

        migrations.CreateModel(

            name='Form',

            fields=[

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

                ('form_id', models.CharField(max_length=64)),

                ('creation_datetime', models.DateTimeField(auto_now_add=True)),

                ('filename', models.CharField(max_length=128, null=True)),

                ('status', models.CharField(choices=[(b'new', b'New'), (b'sent', b'Sent'), (b'closed', b'Closed')], default=b'new', max_length=8)),

                ('demand_data', jsonfield.fields.JSONField(default=dict)),

                ('card_data', jsonfield.fields.JSONField(default=dict)),

            ],

            options={

                'get_latest_by': 'creation_datetime',

            },

        ),

        migrations.CreateModel(

            name='Gesbac',

            fields=[

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

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

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

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

                ('outcoming_sftp', passerelle.utils.sftp.SFTPField(default=None, verbose_name='Outcoming SFTP')),

                ('incoming_sftp', passerelle.utils.sftp.SFTPField(default=None, verbose_name='Incoming SFTP')),

                ('output_files_prefix', models.CharField(max_length=32, verbose_name='Output files prefix')),

                ('input_files_prefix', models.CharField(max_length=32, verbose_name='Input files prefix')),

                ('users', models.ManyToManyField(blank=True, related_name='_gesbac_users_+', related_query_name='+', to='base.ApiUser')),

            ],

            options={

                'verbose_name': 'Gesbac',

            },

        ),

        migrations.AddField(

            model_name='form',

            name='resource',

            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='gesbac.Gesbac'),

        ),

    ]

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

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

from collections import OrderedDict

from django.db import models

from django.utils import six

from django.utils.translation import ugettext_lazy as _

from django.http import Http404

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint, APIError

from passerelle.utils import SFTPField

from jsonfield import JSONField

CSV_DELIMITER = ';'

APPLICANT_SCHEMA = OrderedDict((

    ("form_id", {

        "type": "string",

        "required": True

    }),

    ("demand_date", {

        "type": "string",

        "pattern": "^[0-9]{8}$",

        "required": True

    }),

    ("demand_time", {

        "type": "string",

        "pattern": "^[0-9]{6}$",

        "required": True

    }),

    ("producer_code", {

        "type": "integer",

        "required": True

    }),

    ("invariant_number", {

        "type": "string",

        "required": False,

        "maxLength": 10,

        "default": ""

    }),

    ("city_insee_code", {

        "type": "string",

        "required": True

    }),

    ("street_rivoli_code", {

        "type": "string",

        "required": True

    }),

    ("street_name", {

        "type": "string",

        "required": False,

        "default": ""

    }),

    ("address_complement", {

        "type": "string",

        "maxLength": 32,

        "required": False,

        "default": ""

    }),

    ("street_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("bis_ter", {

        "type": "string",

        "maxLength": 3,

        "required": False,

        "default": ""

    }),

    ("building", {

        "type": "string",

        "maxLength": 5,

        "required": False,

        "default": ""

    }),

    ("hall", {

        "type": "string",

        "maxLength": 5,

        "required": False,

        "default": ""

    }),

    ("appartment_number", {

        "type": "string",

        "maxLength": 5,

        "required": False,

        "default": ""

    }),

    ("producer_social_reason", {

        "type": "string",

        "maxLength": 38,

        "required": False,

        "default": ""

    }),

    ("producer_title_code", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("producer_last_name", {

        "type": "string",

        "maxLength": 38,

        "required": False,

        "default": ""

    }),

    ("producer_first_name", {

        "type": "string",

        "maxLength": 32,

        "required": False,

        "default": ""

    }),

    ("producer_phone", {

        "type": "string",

        "maxLength": 20,

        "required": False,

        "default": ""

    }),

    ("producer_email", {

        "type": "string",

        "maxLength": 50,

        "required": False,

        "default": ""

    }),

    ("owner_last_name", {

        "type": "string",

        "maxLength": 38,

        "required": False,

        "default": ""

    }),

    ("owner_first_name", {

        "type": "string",

        "maxLength": 32,

        "required": False,

        "default": ""

    }),

    ("owner_phone", {

        "type": "string",

        "maxLength": 20,

        "required": False,

        "default": ""

    }),

    ("owner_email", {

        "type": "string",

        "maxLength": 50,

        "required": False,

        "default": ""

    }),

    ("activity_code", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("family_members_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("houses_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t1_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t2_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t3_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t4_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t5_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("t6_flats_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("shops_number", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("garden_size", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("expected_date", {

        "type": "string",

        "pattern": "^[0-9]{8}$",

        "required": False,

        "default": ""

    }),

    ("expected_time", {

        "type": "string",

        "pattern": "^[0-9]{4}$",

        "required": False,

        "default": ""

    }),

    ("modification_code", {

        "type": "integer",

        "required": False,

        "default": 0

    }),

    ("demand_reason_label", {

        "type": "string",

        "required": False,

        "default": ""

    }),

    ("comment", {

        "type": "string",

        "maxLength": 500,

        "required": False,

        "default": ""

    }))

)

CARD_SCHEMA = OrderedDict((

    ("card_subject", {

        "type": "integer",

        "required": True

    }),

    ("card_type", {

        "type": "integer",

        "required": True

    }),

    ("card_demand_reason", {

        "type": "integer",

        "required": True

    }),

    ("cards_quantity", {

        "type": "integer",

        "required": True,

    }),

    ("card_number", {

        "type": "string",

        "maxLength": 20,

        "required": False

    }),

    ("card_bar_code", {

        "type": "string",

        "maxLength": 20,

        "required": False,

        "default": "",

    }),

    ("card_code", {

        "type": "string",

        "maxLength": 20,

        "required": False,

        "default": "",

    }),

    ("card_validity_start_date", {

        "type": "string",

        "required": False,

        "pattern": "^[0-9]{8}$",

        "default": "",

    }),

    ("card_validity_end_date", {

        "type": "string",

        "required": False,

        "pattern": "^[0-9]{8}$",

        "default": "",

    }),

    ("card_comment", {

        "type": "string",

        "required": False,

        "maxLength": 100,

        "default": "",

    }))

)

DEMAND_SCHEMA = APPLICANT_SCHEMA.copy()

DEMAND_SCHEMA.update(CARD_SCHEMA)

SCHEMA = {

    "$schema": "http://json-schema.org/draft-03/schema#",

    "title": "Gesbac",

    "description": "",

    "type": "object",

    "properties": DEMAND_SCHEMA,

}

class Gesbac(BaseResource):

    outcoming_sftp = SFTPField(verbose_name=_('Outcoming SFTP'))

    incoming_sftp = SFTPField(verbose_name=_('Incoming SFTP'))

    output_files_prefix = models.CharField(_('Output files prefix'),

                                           blank=False, max_length=32)

    input_files_prefix = models.CharField(_('Input files prefix'),

                                          blank=False, max_length=32)

    category = _('Business Process Connectors')

    class Meta:

        verbose_name = u'Gesbac'

    def check_status(self):

        with self.outcoming_sftp.client() as out_sftp:

            out_sftp.listdir()

        with self.incoming_sftp.client() as in_sftp:

            in_sftp.listdir()

    def get_responses(self):

        data = []

        with self.incoming_sftp.client() as client:

            for csv_file in client.listdir():

                if not csv_file.startswith(self.input_files_prefix):

                    continue

                with client.open(csv_file) as fd:

                    for row in csv.reader(fd, delimiter=CSV_DELIMITER):

                        data.append(row)

        for card_data in data:

            for form in self.form_set.filter(status='sent'):

                if card_data[1] == form.get_gesbac_id():

                    form.card_data = card_data

                    form.status = 'closed'

                    form.save()

    def hourly(self):

        super(Gesbac, self).hourly()

        self.get_responses()

    def send_demand(self, form_id):

        form = Form.objects.get(id=form_id)

        form.send()

    @endpoint(name='create-demand',

              perm='can_access',

              description=_('Create demand'),

              post={

                  'description': _('Creates a demand file'),

                  'request_body': {

                      'schema': {

                          'application/json': SCHEMA

                      }

                  }

              }

    )

    def create_demand(self, request, post_data):

        form_id = post_data['form_id']

        form = Form.objects.create(resource=self, form_id=form_id)

        post_data['form_id'] = form.get_gesbac_id()

        data = []

        applicant_data = ['E']

        # get applicant attributes

        for name, value in APPLICANT_SCHEMA.items():

            if value['type'] == 'string':

                item = post_data.get(name, '')

            elif value['type'] == 'integer':

                item = post_data.get(name, 0)

            applicant_data.append(item)

        data.append(applicant_data)

        # get card attributes

        card_data = ['CARTE', post_data['form_id']]

        for name, value in CARD_SCHEMA.items():

            if value['type'] == 'string':

                item = post_data.get(name, '')

            elif value['type'] == 'integer':

                item = post_data.get(name, 0)

            card_data.append(item)

        data.append(card_data)

        form.demand_data = data

        form.save()

        self.add_job('send_demand', form_id=form.id)

        return {'data': {'filename': form.get_filename()}}

    @endpoint(name='get-response', perm='can_access',

              description=_('Get response'),

              parameters={

                  'form_id': {

                      'description': _('Form identifier'),

                      'example_value': '42-01'

                  }

              }

    )

    def get_response(self, request, form_id):

        try:

            response = self.form_set.filter(status='closed',

                                            form_id=form_id).latest()

            return {'data': response.card_data}

        except Form.DoesNotExist:

            raise Http404('No response found')

FORM_STATUSES = (

    ('new', 'New'),

    ('sent', 'Sent'),

    ('closed', 'Closed')

)

class Form(models.Model):

    resource = models.ForeignKey(Gesbac)

    form_id = models.CharField(max_length=64)

    creation_datetime = models.DateTimeField(auto_now_add=True)

    filename = models.CharField(max_length=128, null=True)

    status = models.CharField(max_length=8, default='new',

                              choices=FORM_STATUSES

                              )

    demand_data = JSONField()

    card_data = JSONField()

    class Meta:

        get_latest_by = 'creation_datetime'

    def get_gesbac_id(self):

        return self.form_id.replace('-', '%03d' % self.id)

    def get_filename(self):

        if not self.filename:

            timestamp = self.creation_datetime.strftime('%y%m%d-%H%M%S')

            self.filename = '%s%s-%s.csv' % (self.resource.output_files_prefix,

                                             timestamp, self.get_gesbac_id())

            self.save()

        return self.filename

    def send(self):

        with self.resource.outcoming_sftp.client() as client:

            with client.open(self.get_filename(), mode='w') as fd:

                writer = csv.writer(fd, delimiter=CSV_DELIMITER)

                for row in self.demand_data:

                    # encode strings to ASCII

                    writer.writerow([item.encode('latin-1') if isinstance(item, six.string_types) else item

                                     for item in row])

        self.status = 'sent'

        self.save()

    'passerelle.apps.gesbac',

from .sftp import SFTPField, SFTP

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

# tests/test_gesbac.py

# Copyright (C) 2019  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 pytest

from django.utils.timezone import now

from .utils import make_resource

from passerelle.utils import SFTP

from passerelle.apps.gesbac.models import Gesbac, Form

@pytest.fixture

def resource(db, sftpserver):

    return make_resource(

        Gesbac,

        slug='test',

        title='Gesbac',

        description='gesbac',

        outcoming_sftp=SFTP(

            'sftp://foo:bar@{server.host}:{server.port}/output/'.format(server=sftpserver)),

        incoming_sftp=SFTP(

            'sftp://foo:bar@{server.host}:{server.port}/input/'.format(server=sftpserver)),

        output_files_prefix='output-',

        input_files_prefix='input-')

def test_check_status(app, resource, sftpserver):

    with sftpserver.serve_content({'input': {'test': 'content'},

                                   'output': {'file': 'content'}}):

        resource.check_status()

def test_create_demand(app, resource, freezer, sftpserver):

    assert resource.form_set.count() == 0

    timestamp = now()

    payload = {

        'form_id': '42-42',

        'demand_date': timestamp.strftime('%Y%m%d'),

        'demand_time': timestamp.strftime('%H%M%S'),

        'producer_code': 1,

        'city_insee_code': '75114',

        'street_name': 'Château',

        'street_rivoli_code': 'xxxx',

        'producer_social_reason': 'SCOP',

        'producer_last_name': 'Bar',

        'producer_first_name': 'Foo',

        'producer_email': 'foo@example.com',

        'owner_last_name': 'Bar',

        'owner_first_name': 'Foo',

        'owner_email': 'foo@example.com',

        'family_members_number': 5,

        'houses_number': 1,

        'card_type': 1,

        'card_subject': 1,

        'card_demand_reason': 1,

        'card_demand_purpose': 1,

        'cards_quantity': 1

    }

    response = app.post_json('/gesbac/test/create-demand/', params=payload)

    assert resource.form_set.filter(status='new').count() == 1

    form = resource.form_set.get(status='new')

    assert len(form.demand_data) == 2

    expected_filename = '%s%s-4200142.csv' % (resource.output_files_prefix,

                                              timestamp.strftime('%y%m%d-%H%M%S'))

    assert response.json['data']['filename'] == expected_filename

    assert resource.form_set.filter(status='new').count() == 1

    with sftpserver.serve_content({'output': {expected_filename: 'content'}}):

        resource.jobs()

    assert resource.form_set.filter(status='sent').count() == 1

    payload['owner_email'] = 'bar@example.com'

    response = app.post_json('/gesbac/test/create-demand/', params=payload)

    assert resource.form_set.filter(status='new').count() == 1

    form = resource.form_set.get(status='new')

    assert len(form.demand_data) == 2

    assert response.json['data']['filename'] == '%s%s-4200242.csv' % (resource.output_files_prefix,

                                                                      timestamp.strftime('%y%m%d-%H%M%S'))

    with sftpserver.serve_content({'output': {response.json['data']['filename']: 'content'}}):

        resource.jobs()

    assert resource.form_set.filter(status='sent').count() == 2

def test_get_demand_response(app, resource, freezer, sftpserver):

    response = app.get('/gesbac/test/get-response/',

                       params={'form_id': '42-43'}, status=404)

    timestamp = now()

    payload = {

        'form_id': '42-43',

        'demand_date': timestamp.strftime('%Y%m%d'),

        'demand_time': timestamp.strftime('%H%M%S'),

        'producer_code': 1,

        'city_insee_code': '75114',

        'street_name': 'Château',

        'street_rivoli_code': 'xxxx',

        'producer_social_reason': 'SCOP',

        'producer_last_name': 'Bar',

        'producer_first_name': 'Foo',

        'producer_email': 'foo@example.com',

        'owner_last_name': 'Bar',

        'owner_first_name': 'Foo',

        'owner_email': 'foo@example.com',

        'family_members_number': 5,

        'houses_number': 1,

        'card_type': 1,

        'card_subject': 1,

        'card_demand_reason': 1,

        'card_demand_purpose': 1,

        'cards_quantity': 1

    }

    response = app.post_json('/gesbac/test/create-demand/', params=payload)

    data = response.json['data']

    with sftpserver.serve_content({'output': {data['filename']: 'content'},

                                   'input': {}}):

        resource.jobs()

    assert resource.form_set.filter(status='closed').count() == 0

    assert resource.form_set.filter(status='sent').count() == 1

    data = resource.form_set.get(status='sent')

    gesbac_id = data.get_gesbac_id()

    response_filename = '%s91001-090300-%s.csv' % (

        resource.input_files_prefix, gesbac_id)

    assert resource.form_set.filter(

        form_id='42-43', filename=response_filename, status='closed').count() == 0

    with sftpserver.serve_content({'input': {response_filename: 'CARTE;%s;3;2;1234;;;;;;;;' % gesbac_id}}):

        resource.hourly()

    assert resource.form_set.filter(

        form_id='42-43', status='closed').count() == 1

    response = resource.form_set.get(status='closed')

    assert response.card_data == ['CARTE', gesbac_id,

                                  '3', '2', '1234', '', '', '', '', '', '', '', '']

    response = app.get('/gesbac/test/get-response/',

                       params={'form_id': '42-43'})

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

    assert response.json['data'] == ['CARTE', gesbac_id,

                                     '3', '2', '1234', '', '', '', '', '', '', '', '']