0001-add-fault-tolerance-on-convert_value_from_anything-u.patch

Thomas Noël, 27 mars 2018 14:19

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

Subject: [PATCH] add fault tolerance on convert_value_from_anything usage
 (#22793)

... and make None a legitimate value.

 tests/test_hobo_notify.py   | 33 +++++++++++++++++++++++++
 tests/test_workflows.py     | 59 ++++++++++++++++++++++++++++++++++++---------
 wcs/ctl/hobo_notify.py      |  8 ++++--
 wcs/fields.py               |  7 +++---
 wcs/qommon/saml2.py         |  8 ++++--
 wcs/wf/backoffice_fields.py |  6 ++---
 wcs/wf/profile.py           | 10 ++++++--
 7 files changed, 108 insertions(+), 23 deletions(-)

         assert user.is_admin is True
         assert set(user.roles) == set([old_role.id])
         for birthdate in ('baddate', '', None):
             notification = {
                 u'@type': u'provision',
                 u'issuer': 'http://idp.example.net/idp/saml/metadata',
                 u'audience': [u'test'],
                 u'objects': {
                     u'@type': 'user',
                     u'data': [
+                        {
                             u'uuid': u'a' * 32,
                             u'first_name': u'John',
                             u'last_name': u'Doe',
                             u'email': u'john.doe@example.net',
                             u'zipcode': u'13600',
                             u'birthdate': birthdate,
                             u'is_superuser': True,
                             u'roles': [
+                                {
                                     u'uuid': u'xyz',
                                     u'name': u'Service état civil',
                                     u'description': u'etc.',
                                 },
                             ],
+                        }
+                    ]
+                }
+            }
             CmdHoboNotify.process_notification(notification)
             assert User.count() == 1
             if birthdate is not None:  # wrong value : no nothing
                 assert User.select()[0].form_data['_birthdate'].tm_year == 2000
             else:  # empty value : empty field
                 assert User.select()[0].form_data['_birthdate'] is None
     def notify_of_exception(exc_info, context):
         raise Exception(exc_info)

                 assert http_patch_request.call_count == 1
                 assert http_patch_request.call_args[0][1] == '{"bar": "2018-03-20"}'
         user.form_data['4'] = datetime.datetime.now().timetuple()
         user.store()
         year = User.get(user.id).form_data.get('4').tm_year
         for date_value in ('baddate', '', {}, [], None):
             item.fields = [{'field_id': 'bar', 'value': date_value}]
             item.perform(formdata)
             if date_value is not None:  # bad value : do nothing
                 assert User.get(user.id).form_data.get('4').tm_year == year
             else:  # empty value : empty field
                 assert User.get(user.id).form_data.get('4') == None
             with mock.patch('wcs.wf.profile.http_patch_request') as http_patch_request:
                 http_patch_request.return_value = (None, 200, '', None)
                 get_response().process_after_jobs()
                 assert http_patch_request.call_count == 1
                 if date_value is not None:  # bad value : do nothing
                     assert http_patch_request.call_args[0][1] == '{}'
                 else:  # empty value : null field
                     assert http_patch_request.call_args[0][1] == '{"bar": null}'
     def test_set_backoffice_field(http_requests, two_pubs):
         Workflow.wipe()
         FormDef.wipe()
-...
         assert formdata.data['bo1'].base_filename == 'hello.txt'
         assert formdata.data['bo1'].get_content() == 'HELLO WORLD'
         # check wrong value
         del formdata.data['bo1']
         formdata.store()
         assert not 'bo1' in formdata.data
         hello_world = formdata.data['bo1']
         # check wrong value, or None (no file)
         for value in ('="HELLO"', '', 'BAD', '={}', '=[]', None):
             formdata.data['bo1'] = hello_world
             formdata.store()
         item = SetBackofficeFieldsWorkflowStatusItem()
         item.parent = st1
         item.fields = [{'field_id': 'bo1', 'value': '="HELLO"'}]
         item.perform(formdata)
             item = SetBackofficeFieldsWorkflowStatusItem()
             item.parent = st1
             item.fields = [{'field_id': 'bo1', 'value': value}]
             item.perform(formdata)
         formdata = formdef.data_class().get(formdata.id)
         assert formdata.data.get('bo1') is None
             formdata = formdef.data_class().get(formdata.id)
             if value is not None:  # wrong value : do nothing
                 assert formdata.data['bo1'].base_filename == 'hello.txt'
                 assert formdata.data['bo1'].get_content() == 'HELLO WORLD'
             else:  # empty value : remove field
                 assert formdata.data['bo1'] is None
         # check wrong field
         item = SetBackofficeFieldsWorkflowStatusItem()
-...
         formdata = formdef.data_class().get(formdata.id)
         assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)
         # invalid values => do nothing
         for value in ('plop', '', {}, [], '{{ blah }}'):
             item = SetBackofficeFieldsWorkflowStatusItem()
             item.parent = st1
             item.fields = [{'field_id': 'bo1', 'value': value}]
             item.perform(formdata)
             formdata = formdef.data_class().get(formdata.id)
             assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)
         # None : empty date
         item = SetBackofficeFieldsWorkflowStatusItem()
         item.parent = st1
         item.fields = [{'field_id': 'bo1', 'value': "plop"}]
         item.fields = [{'field_id': 'bo1', 'value': None}]
         item.perform(formdata)
         formdata = formdef.data_class().get(formdata.id)
         assert formdata.data['bo1'] is None
     def test_set_backoffice_field_immediate_use(http_requests, two_pubs):
         Workflow.wipe()
         FormDef.wipe()

                             if not field.id.startswith('_'):
                                 continue
                             field_value = o.get(field.id[1:])
                             if field_value and field.convert_value_from_anything:
                                 field_value = field.convert_value_from_anything(field_value)
                             if field.convert_value_from_anything:
                                 try:
                                     field_value = field.convert_value_from_anything(field_value)
                                 except ValueError:
                                     publisher.notify_of_exception(sys.exc_info(), context='[PROVISIONNING]')
                                     continue
                             user.form_data[field.id] = field_value
                         user.name_identifiers = [uuid]
                         role_uuids = [role['uuid'] for role in o['roles']]

         @classmethod
         def convert_value_from_anything(cls, value):
             if value is None:
                 return None
             if hasattr(value, 'base_filename'):
                 upload = PicklableUpload(value.base_filename,
                                          value.content_type or 'application/octet-stream')
-...
         @classmethod
         def convert_value_from_anything(cls, value):
             try:
                 date_value = evalutils.make_date(value).timetuple()
             except ValueError:
             if value is None:
                 return None
             date_value = evalutils.make_date(value).timetuple()  # could raise ValueError
             return date_value
         def convert_value_from_str(self, value):

                     continue
                 field_value = d[key]
                 field = dict_fields.get(field_id)
                 if field and field_value and field.convert_value_from_anything:
                     field_value = field.convert_value_from_anything(field_value)
                 if field and field.convert_value_from_anything:
                     try:
                         field_value = field.convert_value_from_anything(field_value)
                     except ValueError:
                         publisher.notify_of_exception(sys.exc_info(), context='[SAML]')
                         continue
                 if user.form_data.get(field_id) != field_value:
                     user.form_data[field_id] = field_value
                     logger.info('setting field %s of user %s to value %r', field_id, user.id, field_value)

                 try:
                     new_value = self.compute(field['value'], raises=True)
                 except:
                     get_publisher().notify_of_exception(sys.exc_info())
                     get_publisher().notify_of_exception(sys.exc_info(), context='[BO_FIELDS]')
                     continue
                 if new_value and formdef_field.convert_value_from_anything:
                 if formdef_field.convert_value_from_anything:
                     try:
                         new_value = formdef_field.convert_value_from_anything(new_value)
                     except ValueError:
                         get_publisher().notify_of_exception(sys.exc_info())
                         get_publisher().notify_of_exception(sys.exc_info(), context='[BO_FIELDS]')
                         continue
                 formdata.data['%s' % field['field_id']] = new_value

             for field in user_formdef.fields:
                 if field.varname in new_data:
                     field_value = new_data.get(field.varname)
                     if field and field_value and field.convert_value_from_anything:
                         field_value = field.convert_value_from_anything(field_value)
                     if field and field.convert_value_from_anything:
                         try:
                             field_value = field.convert_value_from_anything(field_value)
                         except ValueError:
                             get_publisher().notify_of_exception(sys.exc_info(), context='[PROFILE]')
                             # invalid attribute, do not update it
                             del new_data[field.varname]
                             continue
                     new_user_data[field.id] = field_value
                     # also change initial value to the converted one, as the
                     # initial dictionary is used when sending the profile changes
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-add-fault-tolerance-on-convert_value_from_anything-u.patch