# 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 . import pytest from django.utils.encoding import force_bytes, force_str from django.utils.timezone import now from passerelle.apps.gesbac.models import Gesbac from passerelle.utils import SFTP from tests.utils import make_resource @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-%s.csv' % ( resource.output_files_prefix, timestamp.strftime('%y%m%d-%H%M%S'), form.get_gesbac_id(), ) assert response.json['data']['filename'] == expected_filename assert response.json['data']['gesbac_id'] == '4204200' 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 expected_filename = '%s%s-%s.csv' % ( resource.output_files_prefix, timestamp.strftime('%y%m%d-%H%M%S'), form.get_gesbac_id(), ) assert response.json['data']['filename'] == expected_filename assert response.json['data']['gesbac_id'] == '4204201' with sftpserver.serve_content({'output': {response.json['data']['filename']: 'content'}}): resource.jobs() assert resource.form_set.filter(status='sent').count() == 2 def test_demand_creation_limit(app, resource, freezer): timestamp = now() payload = { 'form_id': '42-44', '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, } for i in range(20): response = app.post_json('/gesbac/test/create-demand/', params=payload) assert response.json['err'] == 0 response = app.post_json('/gesbac/test/create-demand/', params=payload) assert response.json['err'] == 1 def test_get_demand_response(app, resource, freezer, sftpserver): response = app.get('/gesbac/test/get-response/', params={'gesbac_id': '42043'}, 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 # files are encoded in latin-1 comment = force_str('propriétaire') content = force_bytes(force_str('CARTE;%s;3;2;1234;;;;;;;;%s' % (gesbac_id, comment)), 'latin-1') with sftpserver.serve_content({'input': {response_filename: content}}): 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', '', '', '', '', '', '', '', comment] response = app.get('/gesbac/test/get-response/', params={'gesbac_id': gesbac_id}) assert response.json['err'] == 0 assert response.json['data'] == [ 'CARTE', gesbac_id, '3', '2', '1234', '', '', '', '', '', '', '', comment, ]