0001-solis-afi-mss-provide-url-and-credentials-for-GED-60.patch

Nicolas Roche, 28 janvier 2022 11:20

Voir les différences: en ligne côte à côte

Subject: [PATCH] solis-afi-mss: provide url and credentials for GED (#60557)

 .../migrations/0002_auto_20220128_1001.py     | 51 +++++++++++++++++++
 passerelle/contrib/solis_afi_mss/models.py    | 49 +++++++++++-------
 tests/test_solis_afi_mss.py                   | 10 +++-
 3 files changed, 92 insertions(+), 18 deletions(-)
 create mode 100644 passerelle/contrib/solis_afi_mss/migrations/0002_auto_20220128_1001.py

     # Generated by Django 2.2.19 on 2022-01-28 09:01
     from django.db import migrations, models
     class Migration(migrations.Migration):
         dependencies = [
             ('solis_afi_mss', '0001_initial'),
+        ]
         operations = [
             migrations.AddField(
                 model_name='solisafimss',
                 name='ged_basic_auth_password',
                 field=models.CharField(
                     blank=True, max_length=128, verbose_name='Basic GED authentication password'
                 ),
             ),
             migrations.AddField(
                 model_name='solisafimss',
                 name='ged_basic_auth_username',
                 field=models.CharField(
                     blank=True, max_length=128, verbose_name='Basic GED authentication username'
                 ),
             ),
             migrations.AddField(
                 model_name='solisafimss',
                 name='ged_url',
                 field=models.CharField(
                     blank=True,
                     help_text='example: https://solis.mon-application.fr/solisapiged/',
                     max_length=256,
                     verbose_name='GED Service URL',
                 ),
             ),
             migrations.AlterField(
                 model_name='solisafimss',
                 name='base_url',
                 field=models.CharField(
                     help_text='example: https://msppsolis.mon-application.fr/solisAPI/',
                     max_length=256,
                     verbose_name='Service URL',
                 ),
             ),
             migrations.AlterField(
                 model_name='solisafimss',
                 name='verify_cert',
                 field=models.BooleanField(blank=True, default=True, verbose_name='TLS verify certificates'),
             ),
+        ]

     # 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 base64
     from urllib.parse import urljoin
     from django.db import models
     from django.utils.translation import ugettext_lazy as _
     from passerelle.base.models import BaseResource, HTTPResource
     from passerelle.utils.api import endpoint
     from passerelle.utils.jsonresponse import APIError
-...
+    }
     class SolisAfiMss(BaseResource, HTTPResource):
         base_url = models.CharField(
             max_length=256,
             blank=False,
             verbose_name=_('Service URL'),
             help_text=_('example: https://solis.mon-application.fr/api-mss-afi/'),
             help_text=_('example: https://msppsolis.mon-application.fr/solisAPI/'),
+        )
         ged_url = models.CharField(
             max_length=256,
             blank=True,
             verbose_name=_('GED Service URL'),
             help_text=_('example: https://solis.mon-application.fr/solisapiged/'),
+        )
         ged_basic_auth_username = models.CharField(
             max_length=128, verbose_name=_('Basic GED authentication username'), blank=True
+        )
         ged_basic_auth_password = models.CharField(
             max_length=128, verbose_name=_('Basic GED authentication password'), blank=True
+        )
         category = _('Business Process Connectors')
         class Meta:
             verbose_name = _('Solis (mss-afi)')
         def request(self, uri, params=None, json=None, files=None):
             url = urljoin(self.base_url, uri)
         def request(self, url, params=None, json=None, files=None, **kwargs):
             headers = {'Accept': 'application/json'}
             if json:
                 response = self.requests.post(url, json=json, headers=headers)
                 response = self.requests.post(url, json=json, headers=headers, **kwargs)
             elif files:
                 response = self.requests.post(url, files=files, params=params, headers=headers)
                 response = self.requests.post(url, files=files, params=params, headers=headers, **kwargs)
             else:
                 response = self.requests.get(url, params=params, headers=headers)
                 response = self.requests.get(url, params=params, headers=headers, **kwargs)
             if response.status_code // 100 != 2:
                 try:
                     json_content = response.json()
                 except ValueError:
                     json_content = None
                 raise APIError(
                     'error status:%s %r, content:%r'
-...
             json_response.pop('err_desc', None)
             return json_response
         def check_status(self):
             """
             Raise an exception if something goes wrong.
             """
             return self.request('main/isAlive/')
             return self.request(self.base_url + 'main/isAlive/')
         def search_from_email(self, email):
             response = self.request('afi/agent/rechercherParEmail/', params={'adresseMail': email})
             response = self.request(
                 self.base_url + 'afi/agent/rechercherParEmail/', params={'adresseMail': email}
+            )
             index = response.get('indexAgent')
             adults = []
             for key in ('agent', 'conjointAgent'):
                 if response.get(key):
                     adults.append(response.get(key))
             children = response.get('enfantsAgent')
             for record in adults + children:
-...
             description=_("Update contact details for an agent"),
             parameters={
                 'email': {'description': _("Agent's email address")},
             },
             post={'request_body': {'schema': {'application/json': CONTACT_SCHEMA}}},
+        )
         def update_contact(self, request, email, post_data):
             post_data['indexAgent'] = str(self.search_from_email(email)['index'])
             response = self.request('afi/agent/updateCoordonnees/', json=post_data)
             response = self.request(self.base_url + 'afi/agent/updateCoordonnees/', json=post_data)
             return {'data': response}
         @endpoint(
             display_category=_('Budget'),
             display_order=1,
             perm='can_access',
             methods=['get'],
             description=_('Retrieve the list of charges for an agent'),
-...
             },
+        )
         def taxes(self, request, email, year=None):
             index = self.search_from_email(email)['index']
             params = {'indexAgent': str(index)}
             if year:
                 params['anneeImposition'] = year
                 response = self.request('afi/budget/getImposition/', params=params)
                 response = self.request(self.base_url + 'afi/budget/getImposition/', params=params)
                 data = [response] if response else []
             else:
                 response = self.request('afi/budget/getImpositionsParAgent/', params=params)
                 response = self.request(self.base_url + 'afi/budget/getImpositionsParAgent/', params=params)
                 data = response.get('listeImposition')
             for record in data:
                 record['id'] = record.get('anneeImposition')
                 record['text'] = '%(anneeImposition)s: %(montantImposition)s' % record
             return {'data': data}
         @endpoint(
-...
             description=_("Register an agent's tax for one year"),
             parameters={
                 'email': {'description': _("Agent's email address")},
             },
             post={'request_body': {'schema': {'application/json': TAX_SCHEMA}}},
+        )
         def declare_tax(self, request, email, post_data):
             post_data['indexAgent'] = str(self.search_from_email(email)['index'])
             response = self.request('afi/budget/declarerImpot/', json=post_data)
             response = self.request(self.base_url + 'afi/budget/declarerImpot/', json=post_data)
             return {'data': response}
         @endpoint(
             display_category=_('Budget'),
             display_order=3,
             name='simulate-quotient',
             perm='can_access',
             methods=['get'],
-...
             },
+        )
         def simulate_quotient(self, request, code, nb_parts, amount):
             params = {
                 'codeCalcul': code,
                 'nbrPartImposition': nb_parts,
                 'mntImposition': amount,
+            }
             response = self.request('afi/budget/calculer/', params=params)
             response = self.request(self.base_url + 'afi/budget/calculer/', params=params)
             return {'data': response}
         @endpoint(
             display_category=_('Allowance'),
             display_order=1,
             perm='can_access',
             methods=['get'],
             description=_('Retrieve the list of allowance from an agent'),
             parameters={
                 'email': {'description': _("Agent's email address")},
                 'text_template': {'description': _('Text template')},
             },
+        )
         def helps(self, request, email):
             params = {'indexAgent': str(self.search_from_email(email)['index'])}
             response = self.request('afi/aide/getAidesParAgent/', params=params)
             response = self.request(self.base_url + 'afi/aide/getAidesParAgent/', params=params)
             for record in response['aidesFinancieres']:
                 record['id'] = record.get('indexAideFinanciere')
                 record['text'] = '%(dateDemandeAide)s (%(suiviAide)s)' % record
             return {'data': response['aidesFinancieres']}
         @endpoint(
             display_category=_('Allowance'),
-...
             related_persons = []
             for person in results['adults'] + results['children']:
                 for index in [x.strip() for x in post_data['individusConcernes'].split(':') if x.strip()]:
                     if str(person['indexIndividu']) == index:
                         related_persons.append({"indexIndividu": index})
             post_data['indexAgent'] = str(results['index'])
             post_data['individusConcernes'] = related_persons
             response = self.request('afi/aide/deposer/', json=post_data)
             response = self.request(self.base_url + 'afi/aide/deposer/', json=post_data)
             return {'data': response}
         @endpoint(
             display_category=_('Allowance'),
             display_order=3,
             name='add-document',
             perm='can_access',
             methods=['post'],
-...
                 params['typeIdMetierClassement'] = 'INDIVIDU'
                 params['idMetierClassement'] = str(self.search_from_email(email)['index'])
             else:
                 raise APIError("'email' or 'indexAideFinanciere' is a required property", http_status=400)
             document = post_data['document']
             content = base64.b64decode(document['content'])
             files = {'document': (document['filename'], content, document['content_type'])}
             response = self.request('ged/web/document/deposer', params=params, files=files)
             auth = (self.ged_basic_auth_username, self.ged_basic_auth_password)
             response = self.request(
                 self.ged_url + 'ged/web/document/deposer', params=params, files=files, auth=auth
+            )
             return {'data': response}

     from passerelle.contrib.solis_afi_mss.models import SolisAfiMss
     from passerelle.utils.jsonresponse import APIError
     @pytest.fixture
     def connector(db):
         return utils.setup_access_rights(
             SolisAfiMss.objects.create(slug='test', base_url='https://solis-afi-mss.example.net')
             SolisAfiMss.objects.create(
                 slug='test',
                 base_url='https://solis-afi-mss.example.net/',
                 ged_url='https://solis-ged-mss.example.net/',
                 ged_basic_auth_username='jdoe',
                 ged_basic_auth_password='xxx',
+            )
+        )
     TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'solis_afi_mss')
     def json_get_data(filename):
         with open(os.path.join(TEST_BASE_DIR, "%s.json" % filename)) as fd:
-...
             'document': {
                 'filename': '201x201.jpg',
                 'content_type': 'image/jpeg',
                 'content': force_text(base64.b64encode(get_media_file('201x201.jpg'))),
             },
+        }
         resp = app.post_json(endpoint + '?email=foo@dummy.org', params=payload)
         assert mocked_post.mock_calls[0][1][0] == 'https://solis-ged-mss.example.net/ged/web/document/deposer'
         assert mocked_post.mock_calls[0][2]['auth'] == ('jdoe', 'xxx')
         assert mocked_post.mock_calls[0][2]['files']['document'][1] == get_media_file('201x201.jpg')
         assert mocked_post.mock_calls[0][2]['params'] == {
             'codeGedDocument': 9,
             'typeIdMetierClassement': 'INDIVIDU',
             'idMetierClassement': '389227',
+        }
         assert not resp.json['err']
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0001-solis-afi-mss-provide-url-and-credentials-for-GED-60.patch