passerelle/functests/toulouse_maelis/conftest.py

465 lines
15 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': 'VM',
'rl1': {
'civility': 'MME',
'firstname': 'Marge',
'lastname': 'Simpson',
'maidenName': 'Bouvier',
'quality': 'MERE',
'birth': {'dateBirth': '1950-10-01'},
'adresse': {'street1': 'Evergreen Terrace', 'town': 'Springfield', 'zipcode': '62701'},
},
'rl2': {
'civility': 'M.',
'firstname': 'Homer',
'lastname': 'Simpson',
'quality': 'PERE',
'birth': {'dateBirth': '1956-05-12'},
'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': '13',
'employerName': 'Burns',
'phone': '0133333333',
'profession': 'Inspecteur de sécurité',
},
'CAFInfo': {
'number': '123',
'organ': 'A10007752822',
},
'indicatorList': [
{
'code': 'AVL',
'isActive': True,
},
{
'code': 'ETABSPEC',
'note': 'SNPP',
'isActive': True,
},
],
},
'childList': [
{
'sexe': 'M',
'firstname': 'Bart',
'lastname': 'Simpson',
'birth': {'dateBirth': '1978-04-01'},
'bPhoto': True,
'bLeaveAlone': True,
'dietcode': 'STD',
'paiInfoBean': {
'code': 'PAIMED',
'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': 'DTC',
'vaccinationDate': '2011-01-11',
},
{
'code': 'ROR',
'vaccinationDate': '2022-02-22',
},
],
},
'indicatorList': [
{
'code': 'LUNETTE',
'isActive': True,
},
{
'code': 'AUTRE',
'note': 'rebellious',
'isActive': True,
},
],
'authorizedPersonList': [
{
'personInfo': {
'civility': 'M.',
'firstname': 'Abraham Jebediah',
'lastname': 'Simpson',
'dateBirth': '1927-05-24',
'sexe': 'M',
'contact': {
'phone': '0312345678',
'mobile': '',
'mail': 'abe.simpson@example.org',
},
},
'personQuality': {
'code': 'GPP',
},
},
{
'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': 'GMP',
},
},
],
},
{
'sexe': 'F',
'firstname': 'Lisa',
'lastname': 'Simpson',
'birth': {'dateBirth': '1980-05-09'},
'dietcode': 'RSV',
'paiInfoBean': {
'code': 'PAIALI',
},
},
{
'sexe': 'F',
'firstname': 'Maggie',
'lastname': 'Simpson',
'birth': {'dateBirth': '1989-12-17'},
'dietcode': 'BB',
'paiInfoBean': {
'code': 'PAIALI',
},
},
],
'emergencyPersonList': [
{
'civility': 'MME',
'firstname': 'Patty',
'lastname': 'Bouvier',
'dateBirth': '1948-08-30',
'sexe': 'F',
'quality': 'T',
'contact': {
'phone': '0112345678',
'mobile': '0612345678',
'mail': 'patty.bouvier@example.org',
},
},
{
'civility': 'MME',
'firstname': 'Selma',
'lastname': 'Bouvier',
'dateBirth': '1946-04-29',
'sexe': 'F',
'quality': 'OS',
'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_id_on_child(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'
def remove_id_on_rlg(rlg):
if rlg:
rlg['num'] = 'N/A'
rlg['lastname'] = 'N/A'
def remove_id_on_family(family):
family['number'] = 'N/A'
remove_id_on_rlg(family['RL1'])
remove_id_on_rlg(family['RL2'])
for child in family['childList']:
remove_id_on_child(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(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(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(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)
)
output = subprocess.run(cmd, shell=True, check=False, stdout=None, stderr=None)
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 = uuid4().hex[0:30]
# 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': '2022-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'],
'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']:
# 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']):
child['num'] = data['childList'][i]['num'] # required for update
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,
}