passerelle/functests/toulouse_maelis/conftest.py

675 lines
21 KiB
Python

import copy
import datetime
import json
import os
import subprocess
import time
from urllib import parse as urlparse
from uuid import uuid4
import pytest
import requests
from django.core.serializers.json import DjangoJSONEncoder
from zeep.helpers import serialize_object
FAMILY_PAYLOAD = {
'category': 'BI',
'situation': 'MARI',
'nbChild': '3',
'nbTotalChild': '4',
'nbAES': '1',
'rl1': {
'civility': 'MME',
'firstname': 'Marge',
'lastname': 'Simpson',
'maidenName': 'Bouvier',
'quality': 'MERE',
'birth': {
'dateBirth': '1950-10-01',
'countryCode': '404',
},
'adresse': {
'idStreet': '2317',
'num': '4',
'street1': 'requeried having idStreet provided',
'town': 'Springfield',
'zipcode': '62701',
},
},
'rl2': {
'civility': 'MR',
'firstname': 'Homer',
'lastname': 'Simpson',
'quality': 'PERE',
'birth': {
'dateBirth': '1956-05-12',
'place': 'Brive-la-Gaillarde',
'communeCode': '19031',
'cdDepartment': '19',
'countryCode': '',
},
'adresse': {
'num': '742',
'numComp': None,
'street1': 'Evergreen Terrace',
'street2': None,
'town': 'Springfield',
'zipcode': '90701',
},
'contact': {
'isContactMail': True,
'isContactSms': True,
'isInvoicePdf': True,
'mail': 'homer.simpson@example.org.com',
'mobile': '0622222222',
'phone': '0122222222',
},
'profession': {
'addressPro': {
'num': None,
'street': None,
'town': 'Springfield',
'zipcode': '90701',
},
'codeCSP': '46',
'employerName': 'Burns',
'phone': '0133333333',
'profession': 'Inspecteur de sécurité',
},
'CAFInfo': {
'number': '123',
'organ': 'GENE',
},
'indicatorList': [
{
'code': 'AVL',
'isActive': True,
},
{
'code': 'ETABSPEC',
'note': 'SNPP',
'isActive': True,
},
],
},
'childList': [
{
'sexe': 'M',
'firstname': 'Bart',
'lastname': 'Simpson',
'birth': {
'dateBirth': '2014-04-01',
'place': 'Brive-la-Gaillarde',
'communeCode': '19031',
'cdDepartment': '19',
'countryCode': '',
},
'bPhoto': True,
'bLeaveAlone': True,
'dietcode': 'MENU_AV',
'paiInfoBean': {
'code': 'PAI_01',
'dateDeb': '2022-09-01',
'dateFin': '2023-07-01',
'description': 'mischievous, rebellious, misunderstood, disruptive',
},
'medicalRecord': {
'familyDoctor': {
'name': 'MONROE',
'phone': '0612341234',
'address': {
'street1': 'Alameda',
'zipcode': '90701',
'town': 'Springfield',
},
},
'allergy1': 'butterscotch, imitation butterscotch, glow-in-the-dark monster make-up',
'allergy2': 'shrimp and cauliflower',
'comment1': "the shrimp allergy isn't fully identified",
'comment2': None,
'observ1': 'Ay Caramba!',
'observ2': 'Eat my shorts!',
'isAuthHospital': True,
'hospital': 'Springfield General Hospital',
'vaccinList': [
{
'code': '45',
'vaccinationDate': '2011-01-11',
},
{
'code': '24',
'vaccinationDate': '2022-02-22',
},
],
},
'indicatorList': [
{
'code': 'LUNETTE',
'isActive': True,
},
{
'code': 'AUTRE',
'note': 'rebellious',
'isActive': True,
},
],
'authorizedPersonList': [
{
'personInfo': {
'civility': 'MR',
'firstname': 'Abraham Jebediah',
'lastname': 'Simpson',
'dateBirth': '1927-05-24',
'sexe': 'M',
'contact': {
'phone': '0312345678',
'mobile': '',
'mail': 'abe.simpson@example.org',
},
},
'personQuality': {
'code': '13',
},
},
{
'personInfo': {
'civility': 'MME',
'firstname': 'Mona Penelope',
'lastname': 'Simpson',
'dateBirth': '1929-03-15',
'sexe': 'F',
'contact': {
'phone': '0412345678',
'mobile': '0612345678',
'mail': 'mona.simpson@example.org',
},
},
'personQuality': {
'code': '13',
},
},
],
},
{
'sexe': 'F',
'firstname': 'Lisa',
'lastname': 'Simpson',
'birth': {'dateBirth': '2016-05-09'},
'dietcode': 'MENU_SV',
'paiInfoBean': {
'code': 'PAI_02',
},
},
{
'sexe': 'F',
'firstname': 'Maggie',
'lastname': 'Simpson',
'birth': {'dateBirth': '2018-12-17'},
'dietcode': 'MENU_PAI',
'paiInfoBean': {
'code': 'PAI_02',
},
},
{
'sexe': 'M',
'firstname': 'Hugo',
'lastname': 'Simpson',
'birth': {'dateBirth': '2018-04-01'},
'dietcode': 'MENU_AV',
'paiInfoBean': {
'code': 'PAI_01',
},
},
],
'emergencyPersonList': [
{
'civility': 'MME',
'firstname': 'Patty',
'lastname': 'Bouvier',
'dateBirth': '1948-08-30',
'sexe': 'F',
'quality': '13',
'contact': {
'phone': '0112345678',
'mobile': '0612345678',
'mail': 'patty.bouvier@example.org',
},
},
{
'civility': 'MME',
'firstname': 'Selma',
'lastname': 'Bouvier',
'dateBirth': '1946-04-29',
'sexe': 'F',
'quality': '13',
'contact': {
'phone': '0112345678',
'mobile': '0612345678',
'mail': 'selma.bouvier@example.org',
},
},
],
}
def pytest_addoption(parser):
parser.addoption(
'--url',
help='Url of a passerelle Toulouse Maelis connector instance',
default='https://parsifal-passerelle.dev.publik.love/toulouse-maelis/test',
)
parser.addoption('--nameid', help='Publik Name ID', default='functest')
parser.addoption('--dui', help='DUI number', default='')
parser.addoption(
'--lastname', help='override lastname to create a new "update" family', default='Simpson'
)
def unlink(conn, name_id):
url = conn + '/unlink?NameID=%s' % name_id
resp = requests.post(url)
resp.raise_for_status()
return resp
def link(conn, data):
url = conn + '/link?NameID=%s' % data['name_id']
payload = {
'family_id': data['family_id'],
'firstname': data['family_payload']['rl1']['firstname'],
'lastname': data['family_payload']['rl1']['lastname'],
'dateBirth': data['family_payload']['rl1']['birth']['dateBirth'],
}
resp = requests.post(url, json=payload)
resp.raise_for_status()
res = resp.json()
assert res == {'data': 'ok', 'err': 0}
def read_family(conn, name_id):
url = conn + '/read-family?NameID=%s' % name_id
resp = requests.get(url)
resp.raise_for_status()
res = resp.json()
assert res['err'] == 0
return res['data']
def diff(result, expected_file):
class JSONEncoder(DjangoJSONEncoder):
def default(self, o):
if isinstance(o, time.struct_time):
o = datetime.datetime(*tuple(o)[:6])
return super().default(o)
expected_file_path = 'data/' + expected_file
assert os.path.isfile(expected_file_path), 'please, touch %s' % expected_file_path
result_file_path = '/tmp/%s.res' % expected_file
with open(result_file_path, 'w') as json_file:
json.dump(serialize_object(result), json_file, cls=DjangoJSONEncoder, indent=2)
json_file.write('\n')
cmd = 'diff %s %s' % (result_file_path, expected_file_path)
output = subprocess.run(cmd, shell=True, check=False, stdout=None, stderr=None)
assert output.returncode == 0, 'please, %s' % cmd
return True
def remove_extra_indicators(conn, indicators, referential_name):
url = conn + '/read-%s-list' % referential_name
resp = requests.get(url)
resp.raise_for_status()
legacy_id = [x['id'] for x in resp.json()['data']]
for i, item in enumerate(indicators):
if item['code'] not in legacy_id:
del indicators[i]
def remove_id_on_child(conn, child):
child['num'] = 'N/A'
child['lastname'] = 'N/A'
child['mother'] = 'N/A' # dont care yet
child['father'] = 'N/A' # dont care yet
for person in child['authorizedPersonList']:
person['personInfo']['num'] = 'N/A'
remove_extra_indicators(conn, child['indicatorList'], 'child-indicator')
child['indicatorList'].sort(key=lambda x: x['code'])
del child['indicators'] # order may change
child['subscribeSchoolList'] = [] # not managed by test yet
child['subscribeActivityList'] = [] # not managed by test yet
del child['subscribe_natures'] # order may change
def remove_id_on_rlg(conn, rlg):
if rlg:
rlg['num'] = 'N/A'
rlg['lastname'] = 'N/A'
remove_extra_indicators(conn, rlg['indicatorList'], 'rl-indicator')
rlg['indicatorList'].sort(key=lambda x: x['code'])
rlg['quotientList'].sort(key=lambda x: (x['yearRev'], x['dateStart']))
del rlg['indicators'] # order may change
del rlg['quotients'] # order may change
rlg['subscribeActivityList'] = [] # not managed by test yet
del rlg['subscribe_natures'] # order may change
def remove_id_on_family(conn, family):
family['number'] = 'N/A'
family['family_id'] = 'N/A'
remove_id_on_rlg(conn, family['RL1'])
remove_id_on_rlg(conn, family['RL2'])
for child in family['childList']:
remove_id_on_child(conn, child)
for person in family['emergencyPersonList']:
person['numPerson'] = 'N/A'
def diff_child(conn, name_id, index, expected_file, key=None):
data = read_family(conn, name_id)
child = copy.deepcopy(data['childList'][index])
remove_id_on_child(conn, child)
if not key:
assert diff(child, expected_file)
else:
assert diff(child[key], expected_file)
return data
def diff_rlg(conn, name_id, index, expected_file, key=None):
data = read_family(conn, name_id)
rlg = copy.deepcopy(data['RL%s' % index])
remove_id_on_rlg(conn, rlg)
if not key:
assert diff(rlg, expected_file)
else:
assert diff(rlg[key], expected_file)
return data
def diff_family(conn, name_id, expected_file, key=None):
data = read_family(conn, name_id)
family = copy.deepcopy(data)
remove_id_on_family(conn, family)
if not key:
assert diff(family, expected_file)
else:
assert diff(family[key], expected_file)
return data
@pytest.fixture(scope='session')
def conn(request):
return request.config.getoption('--url')
@pytest.fixture(scope='session')
def referentials(conn):
url = urlparse.urlparse(conn)
slug = url.path.split('/')[2]
cmd = (
'passerelle-manage tenant_command cron -v2 --connector toulouse-maelis daily --connector-slug %s --domain %s'
% (slug, url.netloc)
)
print('\nupdate referentials (it may last some time)')
output = subprocess.run(cmd, shell=True, check=False, capture_output=True)
assert output.returncode == 0, 'fails to run cron: %s' % cmd
@pytest.fixture(scope='session')
def create_data(request, conn):
name_id = request.config.getoption('--nameid')
unlink(conn, name_id)
lastname = 'EO_' + uuid4().hex[0:27]
# create family
create_family_payload = copy.deepcopy(FAMILY_PAYLOAD)
del create_family_payload['childList'][1] # without Lisa
del create_family_payload['rl2'] # without Homer
create_family_payload['rl1']['lastname'] = lastname
for child in create_family_payload['childList']:
child['lastname'] = lastname
# test insurance here because it cannot be reset
create_family_payload['childList'][0]['insurance'] = {
'company': 'Total Disaster Insurance',
'contractNumber': '123',
'memberNumber': '456',
'contractStart': '2022-01-01',
'contractEnd': '2222-12-31',
}
url = conn + '/create-family?NameID=%s' % name_id
resp = requests.post(url, json=create_family_payload)
resp.raise_for_status()
create_result = resp.json()
assert create_result['err'] == 0
print('\ncreate DUI: %s' % str(create_result['data']['number']))
data = diff_family(conn, name_id, 'test_create_family.json')
return {
'name_id': name_id, # linked
'family_id': str(create_result['data']['number']),
'family_payload': create_family_payload,
'lastname': lastname,
'rl1_num': data['RL1']['num'],
'bart_num': data['childList'][0]['num'],
'maggie_num': data['childList'][1]['num'],
'hugo_num': data['childList'][2]['num'],
'data': data,
}
@pytest.fixture(scope='session')
def update_data(request, conn):
name_id = request.config.getoption('--nameid')
lastname = request.config.getoption('--lastname')
unlink(conn, name_id)
# allow to create then update a new family if current is broken,
# using --lastname option
FAMILY_PAYLOAD['rl1']['lastname'] = lastname
FAMILY_PAYLOAD['rl2']['lastname'] = lastname
for child in FAMILY_PAYLOAD['childList']:
child['lastname'] = lastname
# try to re-create test family
url = conn + '/create-family?NameID=%s' % name_id
resp = requests.post(url, json=FAMILY_PAYLOAD)
resp.raise_for_status()
create_result = resp.json()
if not create_result['err']:
# create DUI if it is the first time the test is run
family_id = str(create_result['data']['number'])
print('\ncreate DUI: %s' % family_id)
elif 'E54a' in create_result['err_desc']:
# else find DUI number in the error message
family_id = str(create_result['err_desc'][-7:-1])
print('\nre-use DUI: %s' % family_id)
# and link NameID to it
url = conn + '/link?NameID=%s' % name_id
link_payload = {
'family_id': family_id,
'firstname': FAMILY_PAYLOAD['rl1']['firstname'],
'lastname': FAMILY_PAYLOAD['rl1']['lastname'],
'dateBirth': FAMILY_PAYLOAD['rl1']['birth']['dateBirth'],
}
resp = requests.post(url, json=link_payload)
resp.raise_for_status()
assert not resp.json()['err']
else:
# FAMILY_PAYLOAD looks wrong
assert False, create_result
# update DUI
data = read_family(conn, name_id)
update_family_payload = copy.deepcopy(FAMILY_PAYLOAD)
for i, child in enumerate(update_family_payload['childList']):
try:
child['num'] = data['childList'][i]['num'] # required for update
except IndexError:
pass
url = conn + '/update-family?NameID=%s' % name_id
resp = requests.post(url, json=update_family_payload)
resp.raise_for_status()
assert resp.json()['err'] == 0
data = diff_family(conn, name_id, 'test_update_family.json')
return {
'name_id': name_id, # linked
'family_id': family_id,
'family_payload': update_family_payload,
'lastname': lastname,
'rl2_num': data['RL2']['num'],
'bart_num': data['childList'][0]['num'],
'lisa_num': data['childList'][1]['num'],
'maggie_num': data['childList'][2]['num'],
'data': data,
}
@pytest.fixture(scope='session')
def reference_year():
some_date = datetime.date.today()
if some_date.month <= 8:
# between january and august, reference year is the year just before
return some_date.year - 1
return some_date.year
def get_subscription_info(nature, activity_text, unit_text, place_text, con, name_id, person_id, year):
def select_item(resp, text):
item = None
for item in resp.json()['data']:
if item['text'] == text:
break
else:
raise Exception("do not find '%s'" % text)
return item
# select activity
url = con + '/get-person-activity-list'
params = {
'NameID': name_id,
'person_id': person_id,
'start_date': '%s-09-01' % year,
'end_date': '%s-08-31' % (year + 1),
'nature': nature,
}
resp = requests.get(url, params=params)
resp.raise_for_status()
assert resp.json()['err'] == 0
assert len(resp.json()['data']) > 0
activity = select_item(resp, activity_text)
# select unit
url = con + '/get-person-unit-list'
params = {
'NameID': name_id,
'person_id': person_id,
'start_date': '%s-09-01' % year,
'end_date': '%s-08-31' % (year + 1),
'activity_id': activity['id'],
}
resp = requests.get(url, params=params)
resp.raise_for_status()
assert resp.json()['err'] == 0
assert len(resp.json()['data']) > 0
unit = select_item(resp, unit_text)
# select place
url = con + '/get-person-place-list'
params = {
'NameID': name_id,
'person_id': person_id,
'start_date': '%s-09-01' % year,
'end_date': '%s-08-31' % (year + 1),
'activity_id': activity['id'],
'unit_id': unit['id'],
}
resp = requests.get(url, params=params)
resp.raise_for_status()
assert resp.json()['err'] == 0
assert len(resp.json()['data']) > 0
place = select_item(resp, place_text)
assert place['capacityInfo']['controlOK'] is True
# check subscription info
url = con + '/get-person-subscription-info'
params = {
'NameID': name_id,
'person_id': person_id,
'activity_id': activity['id'],
'unit_id': unit['id'],
'place_id': place['id'],
'ref_date': datetime.date.today().strftime('%Y-%m-%d'),
}
resp = requests.get(url, params=params)
resp.raise_for_status()
assert resp.json()['err'] == 0
info = resp.json()['data']
assert info['controlResult']['controlOK'] is True
return {
'activity': activity,
'unit': unit,
'place': place,
'info': info,
}
@pytest.fixture(scope='session')
def loisirs_subscribe_info(conn, create_data, reference_year):
unlink(conn, create_data['name_id'])
link(conn, create_data)
return get_subscription_info(
'LOISIRS',
# Sigec made this loisirs activity available for functests
'TEST ECOLE DES SPORTS 22/23 SEMESTRE 2 - MULTIACTIVITES',
'MERCREDI - 15h30/17h - 8/15Ans',
'ARGOULETS',
conn,
create_data['name_id'],
create_data['bart_num'],
reference_year,
)
@pytest.fixture(scope='session')
def extrasco_subscribe_info(conn, create_data, reference_year):
unlink(conn, create_data['name_id'])
link(conn, create_data)
return get_subscription_info(
'EXTRASCO',
# Sigec made this extra-sco activity available for functests
'ADL ELEMENTAIRE Maourine Avril 2023',
'ADL ELEMENTAIRE Maourine Avril 2023',
'MAOURINE (la) ELEMENTAIRE',
conn,
create_data['name_id'],
create_data['bart_num'],
reference_year,
)
@pytest.fixture(scope='session')
def perisco_subscribe_info(conn, create_data, reference_year):
'''This fixture is a configuration trick from Sigec
as peri-sco should not be available for subscription
and as a consequence, should not be displayed from catalogs'''
unlink(conn, create_data['name_id'])
link(conn, create_data)
return get_subscription_info(
None,
# Sigec made this peri-sco activity available for functests
'TEMPS DU MIDI 22/23',
'TEMPS DU MIDI 22/23',
'DOLTO FRANCOISE MATERNELLE',
conn,
create_data['name_id'],
create_data['bart_num'],
reference_year,
)