0002-forms-redisplay-page-when-datasource-is-unavailable-.patch

Benjamin Dauvergne, 11 novembre 2021 09:58

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

Subject: [PATCH 2/2] forms: redisplay page when datasource is unavailable
 (#56824)

 tests/form_pages/test_all.py  | 179 ++++++++++++++++++++++++++++++++++
 tests/form_pages/test_live.py |  39 ++++++++
 wcs/forms/root.py             |  96 +++++++++++++-----
 3 files changed, 288 insertions(+), 26 deletions(-)

             assert submit_item_data_source_field(ds) == {'0': '1', '0_display': 'un'}
     @pytest.mark.parametrize('fail_after_count_page', range(2, 8))
     @pytest.mark.parametrize('fail_after_count_validation', range(0, 2))
     @mock.patch('wcs.qommon.misc.urlopen')
     def test_form_item_data_source_error(
         urlopen, pub, monkeypatch, fail_after_count_page, fail_after_count_validation
     ):
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {'type': 'json', 'value': 'http://www.example.net/plop'}
         data_source.id_parameter = 'id'
         data_source.store()
         normal_get_structured_value = NamedDataSource.get_structured_value
         class failing_get_structured_value:
             def __init__(self, fail_after_count):
                 self.fail_after_count = fail_after_count
                 self.count = 0
             def __call__(self, *args):
                 import inspect
                 for frame in inspect.stack():
                     if frame.function in ['store_display_value', 'store_structured_value']:
                         count = self.count
                         self.count += 1
                         if count >= self.fail_after_count:
                             return None
                 return normal_get_structured_value(*args)
             @property
             def method(self):
                 def f(*args):
                     return self(*args)
                 return f
         data = {'data': [{'id': '1', 'text': 'un'}, {'id': '2', 'text': 'deux'}]}
         urlopen.side_effect = lambda *args: io.StringIO(json.dumps(data))
         formdef = create_formdef()
         formdef.fields = [
             fields.PageField(id='0', label='1st page', type='page'),
             fields.ItemField(id='1', label='string', data_source={'type': 'foobar'}),
+        ]
         formdef.store()
         resp = get_app(pub).get('/test/')
         formdef.data_class().wipe()
         resp.forms[0]['f1'] = '1'
         # fail in get_structured_value
         monkeypatch.setattr(
             NamedDataSource, 'get_structured_value', failing_get_structured_value(fail_after_count_page).method
+        )
         resp = resp.forms[0].submit('submit')
         assert 'Technical error, please try again' in resp.text
         # fix transient failure
         monkeypatch.setattr(NamedDataSource, 'get_structured_value', normal_get_structured_value)
         resp = resp.forms[0].submit('submit')
         assert 'Check values then click submit.' in resp.text
         # fail in get_structured_value
         monkeypatch.setattr(
             NamedDataSource,
             'get_structured_value',
             failing_get_structured_value(fail_after_count_validation).method,
+        )
         resp = resp.forms[0].submit('submit')
         assert 'Technical error, please try again' in resp.text
         # fix transient failure
         monkeypatch.setattr(NamedDataSource, 'get_structured_value', normal_get_structured_value)
         resp = resp.forms[0].submit('submit')
         assert resp.status_int == 302
         resp = resp.follow()
         assert 'The form has been recorded' in resp.text
         assert formdef.data_class().count() == 1
         data_id = formdef.data_class().select()[0].id
         return formdef.data_class().get(data_id).data
     @pytest.mark.parametrize('fail_after_count_page', range(2, 8))
     @mock.patch('wcs.qommon.misc.urlopen')
     def test_form_item_data_source_error_no_confirmation(urlopen, pub, monkeypatch, fail_after_count_page):
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {'type': 'json', 'value': 'http://www.example.net/plop'}
         data_source.id_parameter = 'id'
         data_source.store()
         normal_get_structured_value = NamedDataSource.get_structured_value
         class failing_get_structured_value:
             def __init__(self, fail_after_count):
                 self.fail_after_count = fail_after_count
                 self.count = 0
             def __call__(self, *args):
                 import inspect
                 for frame in inspect.stack():
                     if frame.function in ['store_display_value', 'store_structured_value']:
                         count = self.count
                         self.count += 1
                         if count >= self.fail_after_count:
                             return None
                 return normal_get_structured_value(*args)
             @property
             def method(self):
                 def f(*args):
                     return self(*args)
                 return f
         data = {'data': [{'id': '1', 'text': 'un'}, {'id': '2', 'text': 'deux'}]}
         urlopen.side_effect = lambda *args: io.StringIO(json.dumps(data))
         formdef = create_formdef()
         formdef.confirmation = False
         formdef.fields = [
             fields.PageField(id='0', label='1st page', type='page'),
             fields.ItemField(id='1', label='string', data_source={'type': 'foobar'}),
+        ]
         formdef.store()
         resp = get_app(pub).get('/test/')
         formdef.data_class().wipe()
         resp.forms[0]['f1'] = '1'
         # fail in get_structured_value
         monkeypatch.setattr(
             NamedDataSource, 'get_structured_value', failing_get_structured_value(fail_after_count_page).method
+        )
         resp = resp.forms[0].submit('submit')
         assert 'Technical error, please try again' in resp.text
         # fix transient failure
         monkeypatch.setattr(NamedDataSource, 'get_structured_value', normal_get_structured_value)
         resp = resp.forms[0].submit('submit')
         assert resp.status_int == 302
         resp = resp.follow()
         assert 'The form has been recorded' in resp.text
         assert formdef.data_class().count() == 1
         data_id = formdef.data_class().select()[0].id
         return formdef.data_class().get(data_id).data
     def test_form_items_data_source_field_submit(pub):
         def submit_items_data_source_field(ds):
             formdef = create_formdef()
-...
         assert resp.forms[1]['f1'].value == 'foobar'  # not a valid email
     @mock.patch('wcs.qommon.misc.urlopen')
     def test_form_autosave_item_field_data_source_error(urlopen, pub):
         ds = {'type': 'json', 'value': 'http://www.example.net/plop'}
         formdef = create_formdef()
         formdef.fields = [
             fields.ItemField(id='1', label='string', data_source=ds),
+        ]
         formdef.enable_tracking_codes = True
         formdef.store()
         data = {'data': [{'id': '1', 'text': 'un'}, {'id': '2', 'text': 'deux'}]}
         urlopen.side_effect = lambda *args: io.StringIO(json.dumps(data))
         formdef.data_class().wipe()
         app = get_app(pub)
         resp = app.get('/test/')
         resp.form['f1'] = '1'  # not a valid email
         # make the ds fails
         with mock.patch.object(NamedDataSource, 'get_structured_value', lambda *args: None):
             autosave_resp = app.post('/test/autosave', params=resp.form.submit_fields())
         assert autosave_resp.json == {
             'reason': 'form deserialization failed: a datasource is unavailable',
             'result': 'error',
+        }
         autosave_resp = app.post('/test/autosave', params=resp.form.submit_fields())
         assert autosave_resp.json['result'] == 'success'
     def test_form_autosave_with_parameterized_datasource(pub):
         formdef = create_formdef()
         formdef.fields = [

         resp = resp.form.submit('submit')  # -> 3rd page
         resp = resp.form.submit('previous')  # -> 2nd page
         assert 'live value: attr1' in resp
     @mock.patch('wcs.qommon.misc.urlopen')
     def test_field_live_condition_data_source_error(urlopen, pub):
         ds = {'type': 'json', 'value': 'http://www.example.net/plop'}
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'Foo'
         formdef.fields = [
             fields.ItemField(id='1', label='string', data_source=ds, varname='foo'),
             fields.StringField(
                 id='2',
                 label='bar',
                 required=True,
                 varname='bar',
                 condition={'type': 'django', 'value': 'form_var_foo_x == "bye"'},
             ),
+        ]
         formdef.store()
         data = {'data': [{'id': '1', 'text': 'un'}, {'id': '2', 'text': 'deux', 'x': 'bye'}]}
         urlopen.side_effect = lambda *args: io.StringIO(json.dumps(data))
         app = get_app(pub)
         resp = app.get('/foo/')
         assert 'f1' in resp.form.fields
         assert 'f2' not in resp.form.fields
         resp.form['f1'] = '2'
         with mock.patch.object(NamedDataSource, 'get_structured_value', lambda *args: None):
             live_resp = app.post('/foo/live', params=resp.form.submit_fields())
             assert live_resp.json == {
                 'reason': 'form deserialization failed: a datasource is unavailable',
                 'result': 'error',
+            }
         live_resp = app.post('/foo/live', params=resp.form.submit_fields())
         assert live_resp.json == {'result': {'1': {'visible': True}, '2': {'visible': True}}}

     from quixote.util import randbytes
     from wcs.categories import Category
     from wcs.fields import SetValueError
     from wcs.formdata import Evolution, FormData
     from wcs.formdef import FormDef
     from wcs.forms.common import FormStatusPage, FormTemplateMixin
-...
                 if had_prefill:
                     # include prefilled data
                     transient_formdata = self.get_transient_formdata(magictoken)
                     # XXX: Should we handle datasources error here ?
                     transient_formdata.data.update(self.formdef.get_data(form))
                     if self.has_draft_support() and not (req.is_from_application() or req.is_from_bot()):
                         # save to get prefilling data in database
-...
                     # reset locked data with newly submitted values, this allows
                     # for templates referencing fields from the sampe page.
                     self.reset_locked_data(form)
                     data = self.formdef.get_data(form)
                     try:
                         data = self.formdef.get_data(form, raise_on_error=True)
                     except SetValueError:
                         return self.page(
                             page,
                             page_change=False,
                             page_error_messages=[_('Technical error, please try again')],
                             transient_formdata=transient_formdata,
+                        )
                     computed_data = self.handle_computed_fields(magictoken, submitted_fields)
                 form_data.update(data)
-...
                     # a new ConditionsVars will get added to the substitution
                     # variables.
                     form_data = copy.copy(session.get_by_magictoken(magictoken, {}))
                     data = self.formdef.get_data(form)
                     try:
                         data = self.formdef.get_data(form, raise_on_error=True)
                     except SetValueError:
                         return self.page(
                             page,
                             page_change=False,
                             page_error_messages=[_('Technical error, please try again')],
                             transient_formdata=transient_formdata,
+                        )
                     form_data.update(data)
                     form_data.update(computed_data)
                     for i, post_condition in enumerate(post_conditions):
-...
                 form_data = session.get_by_magictoken(magictoken, {})
                 with get_publisher().substitutions.temporary_feed(transient_formdata, force_mode='lazy'):
                     data = self.formdef.get_data(form)
                     try:
                         data = self.formdef.get_data(form, raise_on_error=True)
                     except SetValueError:
                         return self.page(
                             page,
                             page_change=False,
                             page_error_messages=[_('Technical error, please try again')],
                             transient_formdata=transient_formdata,
+                        )
                 form_data.update(data)
                 form_data.update(computed_data)
-...
                                 v = field.convert_value_to_str(v)
                             req.form['f%s' % k] = v
                     if self.edit_mode:
                         form = self.create_view_form(form_data, use_tokens=False)
                         return self.submitted_existing(form)
                         view_form = self.create_view_form(form_data, use_tokens=False)
                         try:
                             return self.submitted_existing(view_form)
                         except SetValueError:
                             return self.page(
                                 page,
                                 page_change=False,
                                 page_error_messages=[_('Technical error, please try again')],
                                 transient_formdata=transient_formdata,
+                            )
                     if self.has_confirmation_page():
                         return self.validating(form_data)
                     else:
-...
             self.reset_locked_data(form)
             if step == 1:
                 form.add_submit('previous')
                 magictoken = form.get_widget('magictoken').parse()
                 if form.get_submit() == 'previous':
                     return self.previous_page(len(self.pages), magictoken)
                 magictoken = form.get_widget('magictoken').parse()
                 form_data = session.get_by_magictoken(magictoken, {})
                 data = self.formdef.get_data(form)
                 form_data.update(data)
                 session.add_magictoken(magictoken, form_data)
                 step = 2  # so it will flow to submit
                 form = Form()
                 form.add_hidden('step', '-1')
                 form.add_hidden('page', '-1')
                 form.add_hidden('magictoken', '-1')
                 form.add_submit('cancel')
             if step == 2:
                 form.add_submit('previous')
                 if 'previous' not in form:
                     form.add_submit('previous')
                 magictoken = form.get_widget('magictoken').parse()
                 self.feed_current_data(magictoken)
                 form_data = session.get_by_magictoken(magictoken, {})
-...
                     # submitted a second time
                     return template.error_page(_('This form has already been submitted.'))
                 return self.submitted(form, existing_formdata)
                 try:
                     return self.submitted(form, existing_formdata)
                 except SetValueError:
                     if get_request().form.get('step') == '2':
                         # submit came from the validation page
                         return self.validating(
                             form_data, page_error_messages=[_('Technical error, please try again')]
+                        )
                     else:
                         # last page
                         return self.page(
                             page,
                             page_change=False,
                             page_error_messages=[_('Technical error, please try again')],
                             transient_formdata=transient_formdata,
+                        )
         def reset_locked_data(self, form):
             # reset locked fields, making sure the user cannot alter them.
-...
                 # on webservice results, there can be (temporary?) inconsistencies.
                 return result_error('ouf ot range page_no')
             form = self.create_form(page=page)
             data = self.formdef.get_data(form)
             try:
                 data = self.formdef.get_data(form, raise_on_error=True)
             except SetValueError as e:
                 return result_error('form deserialization failed: %s' % e)
             if not data:
                 return result_error('nothing to save')
-...
             displayed_fields = []
             with get_publisher().substitutions.temporary_feed(formdata, force_mode='lazy'):
                 form = self.create_form(page=page, displayed_fields=displayed_fields, transient_formdata=formdata)
             formdata.data.update(self.formdef.get_data(form))
             try:
                 formdata.data.update(self.formdef.get_data(form, raise_on_error=True))
             except SetValueError as e:
                 return result_error('form deserialization failed: %s' % e)
             return FormStatusPage.live_process_fields(form, formdata, displayed_fields)
         def clean_submission_context(self):
-...
                 filled = self.get_current_draft() or self.formdef.data_class()()
                 filled.just_created()
             filled.data = self.formdef.get_data(form)
             filled.data = self.formdef.get_data(form, raise_on_error=True)
             magictoken = get_request().form['magictoken']
             computed_values = get_session().get_by_magictoken('%s-computed' % magictoken, {})
             filled.data.update(computed_values)
-...
             code.formdata = formdata  # this will .store() the code
         def submitted_existing(self, form):
             new_data = self.formdef.get_data(form)
             new_data = self.formdef.get_data(form, raise_on_error=True)
             magictoken = get_request().form['magictoken']
             computed_values = get_session().get_by_magictoken('%s-computed' % magictoken, {})
             new_data.update(computed_values)
-...
                     return thumbnail
             return get_session().get_tempfile_content(t).get_file_pointer().read()
         def validating(self, data):
         def validating(self, data, page_error_messages=None):
             self.on_validation_page = True
             get_request().view_name = 'validation'
             self.html_top(self.formdef.name)
-...
             # over in rendering.
             get_request().environ['REQUEST_METHOD'] = 'GET'
             form = self.create_view_form(data)
             if page_error_messages:
                 form.add_global_errors(page_error_messages)
             token_widget = form.get_widget(form.TOKEN_NAME)
             token_widget._parsed = True
             if self.formdef.has_captcha and not (get_session().get_user() or get_session().won_captcha):
-...
+            }
             if self.has_draft_support() and data:
                 context['tracking_code_box'] = lambda: self.tracking_code_box(data, magictoken)
             return template.QommonTemplateResponse(
                 templates=list(self.get_formdef_template_variants(self.validation_templates)), context=context
+            )
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0002-forms-redisplay-page-when-datasource-is-unavailable-.patch