0001-misc-store-LoggedErrors-in-SQL-48925.patch

     from webtest import Upload
     from wcs.qommon.http_request import HTTPRequest
     from wcs.logged_errors import LoggedError
     from wcs.roles import Role
     from wcs.workflows import (
         Workflow, WorkflowCriticalityLevel, DisplayMessageWorkflowStatusItem,
-...
     def test_workflows_wscall_status_error(pub):
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         create_superuser(pub)
         LoggedError.wipe()
         pub.loggederror_class.wipe()
         Workflow.wipe()
         workflow = Workflow(name='foo')
         baz_status = workflow.add_status(name='baz')
-...
         app = login(get_app(pub))
         app.get('/backoffice/workflows/%s/' % workflow.id)
         assert LoggedError.count() == 0
         assert pub.loggederror_class.count() == 0
         # delete foo status
         del workflow.possible_status[1]
         workflow.store()
         app.get('/backoffice/workflows/%s/' % workflow.id)
         assert LoggedError.count() == 1
         error = LoggedError.select()[0]
         assert pub.loggederror_class.count() == 1
         error = pub.loggederror_class.select()[0]
         assert error.tech_id == '%s-reference-to-invalid-status-in-workflow-foo-status-baz-item-webservice' % workflow.id
         assert error.formdef_id is None
         assert error.workflow_id == workflow.id

     from wcs.carddef import CardDef
     from wcs.categories import Category
     from wcs.formdef import FormDef
     from wcs.logged_errors import LoggedError
     from wcs import fields
     from wcs.wscalls import NamedWsCall
-...
     @pytest.mark.parametrize('notify_on_errors', [True, False])
     @pytest.mark.parametrize('record_on_errors', [True, False])
     def test_backoffice_wscall_on_error(http_requests, pub, emails, notify_on_errors, record_on_errors):
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}
         pub.write_cfg()
         create_user(pub)
         create_environment(pub)
         pub.loggederror_class.wipe()
         formdef = FormDef.get_by_urlname('form-title')
         form_class = formdef.data_class()
-...
         else:
             assert emails.count() == 0
         # check LoggedError
         # check pub.loggederror_class
         if record_on_errors:
             assert LoggedError.count() == 1
             LoggedError.wipe()
             assert pub.loggederror_class.count() == 1
             pub.loggederror_class.wipe()
         else:
             assert LoggedError.count() == 0
             assert pub.loggederror_class.count() == 0
     def test_backoffice_wscall_attachment(http_requests, pub):
-...
     def test_backoffice_logged_errors(pub):
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         Workflow.wipe()
         workflow = Workflow.get_default_workflow()
         workflow.id = '12'
-...
         carddef.fields = []
         carddef.store()
         LoggedError.wipe()
         pub.loggederror_class.wipe()
         create_superuser(pub)
         app = login(get_app(pub))
-...
         resp = app.get('/test/')
         resp = resp.form.submit('submit').follow()
         resp = resp.form.submit('submit')
         assert LoggedError.count() == 1
         assert pub.loggederror_class.count() == 1
         app = login(get_app(pub))
         resp = app.get('/backoffice/forms/%s/' % formdef.id)
-...
         assert not 'Acked' in resp.text
         resp = resp.click('Ack').follow()
         assert 'Acked' in resp.text
         assert LoggedError.select()[0].acked is True
         assert pub.loggederror_class.select()[0].acked is True
         resp = resp.click('Delete').follow()
         assert LoggedError.count() == 0
         assert pub.loggederror_class.count() == 0
         pub.cfg.update({'debug': {'error_email': None}})
         pub.write_cfg()
-...
         resp = app.get('/test/')
         resp = resp.form.submit('submit').follow()
         resp = resp.form.submit('submit')
         assert LoggedError.count() == 1
         assert pub.loggederror_class.count() == 1
         app = login(get_app(pub))
         resp = app.get('/backoffice/workflows/%s/' % workflow.id)
-...
         resp = app.get('/test/')
         resp = resp.form.submit('submit').follow()
         resp = resp.form.submit('submit')
         assert LoggedError.count() == 2
         assert pub.loggederror_class.count() == 2
         app = login(get_app(pub))
         resp = app.get('/backoffice/workflows/%s/' % workflow.id)
-...
         source_formdef.store()
         source_formdef.data_class().wipe()
         target_formdef.data_class().wipe()
         LoggedError.wipe()
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
         return locals()
     def test_backoffice_create_formdata_backoffice_submission(create_formdata):
     def test_backoffice_create_formdata_backoffice_submission(pub, create_formdata):
         # create submitting user
         user = create_formdata['pub'].user_class()
         user.name = 'Jean Darmette'
-...
         assert target_data_class.count() == 0
         # resubmit it through backoffice submission
         resp = resp.form.submit(name='button_resubmit')
         assert LoggedError.count() == 0
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 0
         assert target_data_class.count() == 1
         target_formdata = target_data_class.select()[0]
-...
         assert pq('.field-type-file .value').text() == 'bar'
     def test_linked_forms_variables(create_formdata):
     def test_linked_forms_variables(pub, create_formdata):
         # create source formdata
         formdata = create_formdata['source_formdef'].data_class()()
         upload = PicklableUpload('/foo/bar', content_type='text/plain')
-...
         formdata.perform_workflow()
         formdata.store()
         get_publisher().substitutions.reset()
         get_publisher().substitutions.feed(formdata)
         substvars = get_publisher().substitutions.get_context_variables(mode='lazy')
         pub.substitutions.reset()
         pub.substitutions.feed(formdata)
         substvars = pub.substitutions.get_context_variables(mode='lazy')
         assert str(substvars['form_links_resubmitted_form_var_foo_string']) == 'coucou'
         assert 'form_links_resubmitted_form_var_foo_string' in substvars.get_flat_keys()
-...
         assert 'form_links_resubmitted_form_var_foo_string' not in resp
     def test_backoffice_create_formdata_map_fields_by_varname(create_formdata):
     def test_backoffice_create_formdata_map_fields_by_varname(pub, create_formdata):
         create_formdata['create_formdata'].map_fields_by_varname = True
         create_formdata['create_formdata'].mappings  = []
         create_formdata['wf'].store()
-...
         assert target_data_class.count() == 0
         # resubmit it through backoffice submission
         resp = resp.form.submit(name='button_resubmit')
         assert LoggedError.count() == 0
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 0
         assert target_data_class.count() == 1
         target_formdata = target_data_class.select()[0]

     from wcs.carddef import CardDef
     from wcs.categories import CardDefCategory
     from wcs.formdef import FormDef
     from wcs.logged_errors import LoggedError
     from wcs.qommon.http_request import HTTPRequest
     from wcs.wf.wscall import WebserviceCallStatusItem
     from wcs.workflows import ChoiceWorkflowStatusItem, Workflow
-...
     def test_backoffice_cards_wscall_failure_display(http_requests, pub, studio):
         LoggedError.wipe()
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         pub.loggederror_class.wipe()
         user = create_user(pub)
         Workflow.wipe()
-...
         resp = resp.follow()
         assert 'Error during webservice call' in resp.text
         assert LoggedError.count() == 1
         assert LoggedError.select()[0].get_formdata().data == {'1': 'plop'}
         assert pub.loggederror_class.count() == 1
         assert pub.loggederror_class.select()[0].get_formdata().data == {'1': 'plop'}
     def test_block_card_item_link(pub, studio, blocks_feature):

     from wcs.data_sources import NamedDataSource
     from wcs.wscalls import NamedWsCall
     from wcs import fields
     from wcs.logged_errors import LoggedError
     from wcs.forms.root import PublicFormStatusPage
     from utilities import get_app, login, create_temporary_pub, clean_temporary_pub
-...
         assert formdef.data_class().get(data_id).status == 'wf-%s' % st2.id
         # jump to a nonexistent status == do not jump, but add a LoggedError
         LoggedError.wipe()
         assert LoggedError.count() == 0
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
             assert pub.loggederror_class.count() == 0
         editable.status = 'deleted_status_id'
         workflow.store()
         # go back to st1
-...
         resp = resp.forms[0].submit('submit')
         resp = resp.follow()
         assert formdef.data_class().get(data_id).status == 'wf-%s' % st1.id  # stay on st1
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.formdata_id == str(formdata.id)
         assert logged_error.formdef_id == formdef.id
         assert logged_error.workflow_id == workflow.id
         assert logged_error.status_id == st1.id
         assert logged_error.status_item_id == editable.id
         assert logged_error.occurences_count == 1
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 1
             logged_error = pub.loggederror_class.select()[0]
             assert logged_error.formdata_id == str(formdata.id)
             assert logged_error.formdef_id == formdef.id
             assert logged_error.workflow_id == workflow.id
             assert logged_error.status_id == st1.id
             assert logged_error.status_item_id == editable.id
             assert logged_error.occurences_count == 1
         # do it again: increment logged_error.occurences_count
         page = login(get_app(pub), username='foo', password='foo').get('/test/%s/' % data_id)
-...
         resp = resp.forms[0].submit('submit')
         resp = resp.follow()
         assert formdef.data_class().get(data_id).status == 'wf-%s' % st1.id  # stay on st1
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.occurences_count == 2
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 1
             logged_error = pub.loggederror_class.select()[0]
             assert logged_error.occurences_count == 2
     def test_form_edit_autocomplete_list(pub):
-...
     def test_logged_errors(pub):
         if not pub.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         Workflow.wipe()
         workflow = Workflow.get_default_workflow()
         workflow.id = '12'
-...
         formdef.fields = []
         formdef.store()
         LoggedError.wipe()
         pub.loggederror_class.wipe()
         app = get_app(pub)
         resp = app.get('/test/')
         resp = resp.form.submit('submit').follow()
         resp = resp.form.submit('submit')
         assert LoggedError.count() == 1
         assert pub.loggederror_class.count() == 1
         resp = app.get('/test/')
         resp = resp.form.submit('submit').follow()
         resp = resp.form.submit('submit')
         assert LoggedError.count() == 1
         assert pub.loggederror_class.count() == 1
         error = LoggedError.get_on_index(
             '34-12-just_submitted-_jump-failed-to-evaluate-condition-ZeroDivisionError-integer-division-or-modulo-by-zero',
             'tech_id')
         error = list(pub.loggederror_class.get_with_indexed_value(
             'tech_id',
             '34-12-just_submitted-_jump-failed-to-evaluate-condition-ZeroDivisionError-integer-division-or-modulo-by-zero'))[0]
         assert error.occurences_count == 2
         assert len(LoggedError.get_ids_with_indexed_value('formdef_id', '34')) == 1
         assert len(LoggedError.get_ids_with_indexed_value('formdef_id', 'X')) == 0
         assert len(list(pub.loggederror_class.get_with_indexed_value('formdef_id', '34'))) == 1
         assert len(list(pub.loggederror_class.get_with_indexed_value('formdef_id', 'X'))) == 0
         assert len(LoggedError.get_ids_with_indexed_value('workflow_id', '12')) == 1
         assert len(LoggedError.get_ids_with_indexed_value('workflow_id', 'X')) == 0
         assert len(list(pub.loggederror_class.get_with_indexed_value('workflow_id', '12'))) == 1
         assert len(list(pub.loggederror_class.get_with_indexed_value('workflow_id', 'X'))) == 0
     def test_formdata_named_wscall(http_requests, pub):
-...
         assert pq('.linked .foo_string').text() == 'zob'
     def test_create_formdata_empty_item_ds_with_id_parameter(create_formdata):
         LoggedError.wipe()
     def test_create_formdata_empty_item_ds_with_id_parameter(pub, create_formdata):
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
         NamedDataSource.wipe()
         data_source = NamedDataSource(name='foobar')
         data_source.data_source = {
-...
             resp = resp.form.submit('submit')  # -> submission
             resp = resp.follow()
             assert create_formdata['target_formdef'].data_class().count() == 0
             assert LoggedError.count() == 0
             if pub.is_using_postgresql():
                 assert pub.loggederror_class.count() == 0
             resp = resp.form.submit('button_resubmit')
             assert LoggedError.count() == 0
             if pub.is_using_postgresql():
                 assert pub.loggederror_class.count() == 0
     def test_js_libraries(pub):

     from wcs.conditions import Condition
     from wcs.formdef import FormDef
     from wcs.formdata import Evolution
     from wcs.logged_errors import LoggedError
     from wcs.roles import Role
     from wcs import sessions
     from wcs.variables import LazyFormData
-...
         assert queryset.count == 4
         queryset = lazy_formdata.objects.filter_by('foo_foo').apply_filter_value('X')
         assert queryset.count == 0
         LoggedError.wipe()
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
         queryset = lazy_formdata.objects.filter_by('unknown').apply_filter_value('X')
         assert queryset.count == 0
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Invalid filter "unknown"'
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 1
             logged_error = pub.loggederror_class.select()[0]
             assert logged_error.summary == 'Invalid filter "unknown"'
         # filter function on backoffice field
         queryset = lazy_formdata.objects.filter_by('backoffice_blah').apply_filter_value('plop1')

     from quixote import cleanup
     from wcs.formdef import FormDef
     from wcs.fields import FileField
     from wcs.logged_errors import LoggedError
     from wcs.mail_templates import MailTemplate
     from wcs.workflows import Workflow, SendmailWorkflowStatusItem
     from wcs.qommon.http_request import HTTPRequest
-...
         assert workflow.possible_status[0].items[0].mail_template == 'test-mail-template'
     def test_workflow_send_mail_template(pub, superuser, mail_templates_option, emails):
     def test_workflow_send_mail_template_with_sql(superuser, mail_templates_option, emails):
         pub = create_temporary_pub(sql_mode=True)
         req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
         pub.set_app_dir(req)
         pub._set_request(req)
         Workflow.wipe()
         MailTemplate.wipe()
-...
         # check nothing is sent and an error is logged if the mail template is
         # missing
         emails.empty()
         LoggedError.wipe()
         pub.loggederror_class.wipe()
         MailTemplate.wipe()
         item.perform(formdata)
         pub.get_request().response.process_after_jobs()
         assert emails.count() == 0
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert pub.loggederror_class.count() == 1
         logged_error = pub.loggederror_class.select()[0]
         assert logged_error.summary == 'reference to invalid mail template test-mail-template in status Status1'

             ItemsField, CommentField, EmailField, PageField, TitleField,
             SubtitleField, TextField, BoolField, TableField)
     from wcs.formdata import Evolution
     from wcs.logged_errors import LoggedError
     from wcs.roles import Role
     from wcs.workflows import (Workflow, WorkflowStatusItem,
             SendmailWorkflowStatusItem, SendSMSWorkflowStatusItem,
-...
         assert item.must_jump(formdata) is False
     def test_jump_bad_python_condition(pub):
     def test_jump_bad_python_condition(two_pubs):
         if not two_pubs.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'foobar'
         formdef.store()
         pub.substitutions.feed(formdef)
         two_pubs.substitutions.feed(formdef)
         formdef.data_class().wipe()
         formdata = formdef.data_class()()
         item = JumpWorkflowStatusItem()
         LoggedError.wipe()
         two_pubs.loggederror_class.wipe()
         item.condition = {'type': 'python', 'value': 'form_var_foobar == 0'}
         assert item.must_jump(formdata) is False
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert two_pubs.loggederror_class.count() == 1
         logged_error = two_pubs.loggederror_class.select()[0]
         assert logged_error.summary == 'Failed to evaluate condition'
         assert logged_error.exception_class == 'NameError'
         assert logged_error.exception_message == "name 'form_var_foobar' is not defined"
         assert logged_error.expression == 'form_var_foobar == 0'
         assert logged_error.expression_type == 'python'
         LoggedError.wipe()
         two_pubs.loggederror_class.wipe()
         item.condition = {'type': 'python', 'value': '~ invalid ~'}
         assert item.must_jump(formdata) is False
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert two_pubs.loggederror_class.count() == 1
         logged_error = two_pubs.loggederror_class.select()[0]
         assert logged_error.summary == 'Failed to evaluate condition'
         assert logged_error.exception_class == 'SyntaxError'
         assert logged_error.exception_message == 'unexpected EOF while parsing (<string>, line 1)'
-...
         assert logged_error.expression_type == 'python'
     def test_jump_django_conditions(pub):
     def test_jump_django_conditions(two_pubs):
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'foobar'
-...
         formdef.store()
         formdata = formdef.data_class()()
         formdata.data = {'1': 'hello'}
         pub.substitutions.feed(formdata)
         two_pubs.substitutions.feed(formdata)
         item = JumpWorkflowStatusItem()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         item.condition = {'type': 'django', 'value': '1 < 2'}
         assert item.must_jump(formdata) is True
-...
         item.condition = {'type': 'django', 'value': 'form_var_foo|first|upper == "X"'}
         assert item.must_jump(formdata) is False
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         item.condition = {'type': 'django', 'value': '~ invalid ~'}
         assert item.must_jump(formdata) is False
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Failed to evaluate condition'
         assert logged_error.exception_class == 'TemplateSyntaxError'
         assert logged_error.exception_message == "Could not parse the remainder: '~' from '~'"
         assert logged_error.expression == '~ invalid ~'
         assert logged_error.expression_type == 'django'
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.summary == 'Failed to evaluate condition'
             assert logged_error.exception_class == 'TemplateSyntaxError'
             assert logged_error.exception_message == "Could not parse the remainder: '~' from '~'"
             assert logged_error.expression == '~ invalid ~'
             assert logged_error.expression_type == 'django'
     def test_check_auth(pub):
-...
     def test_dispatch_auto(pub):
         LoggedError.wipe()
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
         formdef = FormDef()
         formdef.name = 'baz'
-...
         pub.substitutions.feed(formdata)
         item.perform(formdata)
         assert not formdata.workflow_roles
         assert LoggedError.count() == 1
         error = LoggedError.select()[0]
         assert error.tech_id == '%s-_default-error-in-dispatch-missing-role-foobar' % formdef.id
         assert error.formdef_id == formdef.id
         assert error.workflow_id == '_default'
         assert error.summary == 'error in dispatch, missing role (foobar)'
         assert error.occurences_count == 1
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 1
             error = pub.loggederror_class.select()[0]
             assert error.tech_id == '%s-_default-error-in-dispatch-missing-role-foobar' % formdef.id
             assert error.formdef_id == formdef.id
             assert error.workflow_id == '_default'
             assert error.summary == 'error in dispatch, missing role (foobar)'
             assert error.occurences_count == 1
     def test_dispatch_computed(pub):
         LoggedError.wipe()
         if pub.is_using_postgresql():
             pub.loggederror_class.wipe()
         formdef = FormDef()
         formdef.name = 'baz'
-...
         item.role_id = '="foobar"'
         item.perform(formdata)
         assert not formdata.workflow_roles
         assert LoggedError.count() == 1
         error = LoggedError.select()[0]
         assert error.tech_id == '%s-_default-error-in-dispatch-missing-role-foobar' % formdef.id
         assert error.formdef_id == formdef.id
         assert error.workflow_id == '_default'
         assert error.summary == 'error in dispatch, missing role (="foobar")'
         assert error.occurences_count == 1
         if pub.is_using_postgresql():
             assert pub.loggederror_class.count() == 1
             error = pub.loggederror_class.select()[0]
             assert error.tech_id == '%s-_default-error-in-dispatch-missing-role-foobar' % formdef.id
             assert error.formdef_id == formdef.id
             assert error.workflow_id == '_default'
             assert error.summary == 'error in dispatch, missing role (="foobar")'
             assert error.occurences_count == 1
     def test_roles(pub):
-...
     def test_jump_missing_previous_mark(two_pubs):
         if not two_pubs.is_using_postgresql():
             pytest.skip('this requires SQL')
             return
         workflow = Workflow(name='jump-mark')
         st1 = workflow.add_status('Status1', 'st1')
-...
         formdata.store()
         time.sleep(0.3)
         LoggedError.wipe()
         two_pubs.loggederror_class.wipe()
         _apply_timeouts(two_pubs)
         assert LoggedError.count() == 1
         assert two_pubs.loggederror_class.count() == 1
     def test_sms(pub, sms_mocking):
-...
     def test_set_backoffice_field(http_requests, two_pubs):
         Workflow.wipe()
         FormDef.wipe()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         wf = Workflow(name='xxx')
         wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)
         wf.backoffice_fields_formdef.fields = [
-...
         formdata = formdef.data_class().get(formdata.id)
         assert formdata.data['bo1'] == ''
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         item.fields = [{'field_id': 'bo1', 'value': '= ~ invalid python ~'}]
         item.perform(formdata)
         formdata = formdef.data_class().get(formdata.id)
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Failed to compute Python expression'
         assert logged_error.formdata_id == str(formdata.id)
         assert logged_error.expression == ' ~ invalid python ~'
         assert logged_error.expression_type == 'python'
         assert logged_error.exception_class == 'SyntaxError'
         assert logged_error.exception_message == 'invalid syntax (<string>, line 1)'
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.summary == 'Failed to compute Python expression'
             assert logged_error.formdata_id == str(formdata.id)
             assert logged_error.expression == ' ~ invalid python ~'
             assert logged_error.expression_type == 'python'
             assert logged_error.exception_class == 'SyntaxError'
             assert logged_error.exception_message == 'invalid syntax (<string>, line 1)'
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         item.fields = [{'field_id': 'bo1', 'value': '{% if bad django %}'}]
         item.perform(formdata)
         formdata = formdef.data_class().get(formdata.id)
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Failed to compute template'
         assert logged_error.formdata_id == str(formdata.id)
         assert logged_error.expression == '{% if bad django %}'
         assert logged_error.expression_type == 'template'
         assert logged_error.exception_class == 'TemplateError'
         assert logged_error.exception_message.startswith('syntax error in Django template')
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.summary == 'Failed to compute template'
             assert logged_error.formdata_id == str(formdata.id)
             assert logged_error.expression == '{% if bad django %}'
             assert logged_error.expression_type == 'template'
             assert logged_error.exception_class == 'TemplateError'
             assert logged_error.exception_message.startswith('syntax error in Django template')
     def test_set_backoffice_field_file(http_requests, two_pubs):
-...
             item.parent = st1
             item.fields = [{'field_id': 'bo1', 'value': value}]
             LoggedError.wipe()
             if two_pubs.is_using_postgresql():
                 two_pubs.loggederror_class.wipe()
             item.perform(formdata)
             formdata = formdef.data_class().get(formdata.id)
             assert formdata.data['bo1'].base_filename == 'hello.txt'
             assert formdata.data['bo1'].get_content() == b'HELLO WORLD'
             assert LoggedError.count() == 1
             logged_error = LoggedError.select()[0]
             assert logged_error.summary.startswith('Failed to convert')
             assert logged_error.formdata_id == str(formdata.id)
             assert logged_error.exception_class == 'ValueError'
             if two_pubs.is_using_postgresql():
                 assert two_pubs.loggederror_class.count() == 1
                 logged_error = two_pubs.loggederror_class.select()[0]
                 assert logged_error.summary.startswith('Failed to convert')
                 assert logged_error.formdata_id == str(formdata.id)
                 assert logged_error.exception_class == 'ValueError'
         # check wrong field
         item = SetBackofficeFieldsWorkflowStatusItem()
-...
     def test_set_backoffice_field_date(two_pubs):
         Workflow.wipe()
         FormDef.wipe()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         wf = Workflow(name='xxx')
         wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)
         st1 = wf.add_status('Status1')
-...
         assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)
         # invalid values => do nothing
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         for value in ('plop', '={}', '=[]'):
             item = SetBackofficeFieldsWorkflowStatusItem()
             item.parent = st1
             item.fields = [{'field_id': 'bo1', 'value': value}]
             LoggedError.wipe()
             if two_pubs.is_using_postgresql():
                 two_pubs.loggederror_class.wipe()
             item.perform(formdata)
             formdata = formdef.data_class().get(formdata.id)
             assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)
             assert LoggedError.count() == 1
             assert LoggedError.select()[0].summary.startswith('Failed to convert')
             if two_pubs.is_using_postgresql():
                 assert two_pubs.loggederror_class.count() == 1
                 assert two_pubs.loggederror_class.select()[0].summary.startswith('Failed to convert')
         # None : empty date
         item = SetBackofficeFieldsWorkflowStatusItem()
-...
     def test_set_backoffice_field_boolean(two_pubs):
         Workflow.wipe()
         FormDef.wipe()
         LoggedError.wipe()
         wf = Workflow(name='xxx')
         wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)
         st1 = wf.add_status('Status1')
-...
     def test_workflow_action_condition(two_pubs):
         pub = two_pubs
         pub._set_request(None)  # to avoid after jobs
         two_pubs._set_request(None)  # to avoid after jobs
         workflow = Workflow(name='jump condition migration')
         st1 = workflow.add_status('Status1', 'st1')
         workflow.store()
-...
         role = Role(name='bar1')
         role.store()
         user = pub.user_class()
         user = two_pubs.user_class()
         user.roles = [role.id]
         user.store()
-...
         choice.condition = {'type': 'python', 'value': 'form_var_foo == "foo"'}
         workflow.store()
         with pub.substitutions.temporary_feed(formdata1):
         with two_pubs.substitutions.temporary_feed(formdata1):
             assert FormDef.get(formdef.id).data_class().get(formdata1.id).get_actions_roles() == set([role.id])
         with pub.substitutions.temporary_feed(formdata2):
         with two_pubs.substitutions.temporary_feed(formdata2):
             assert FormDef.get(formdef.id).data_class().get(formdata2.id).get_actions_roles() == set()
         assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 1
-...
         assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 2
         # bad condition
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         choice.condition = {'type': 'python', 'value': 'foobar == barfoo'}
         workflow.store()
         assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 0
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.occurences_count > 1  # should be 2... == 12 with pickle, 4 with sql
         assert logged_error.summary == 'Failed to evaluate condition'
         assert logged_error.exception_class == 'NameError'
         assert logged_error.exception_message == "name 'foobar' is not defined"
         assert logged_error.expression == 'foobar == barfoo'
         assert logged_error.expression_type == 'python'
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.occurences_count > 1  # should be 2... == 12 with pickle, 4 with sql
             assert logged_error.summary == 'Failed to evaluate condition'
             assert logged_error.exception_class == 'NameError'
             assert logged_error.exception_message == "name 'foobar' is not defined"
             assert logged_error.expression == 'foobar == barfoo'
             assert logged_error.expression_type == 'python'
     def test_notifications(pub, http_requests):
-...
         formdef.fields = []
         formdef.workflow_id = workflow.id
         formdef.store()
         formdef.data_class().wipe()
         for i in range(5):
             formdata = formdef.data_class()()
-...
         assert 'http://example.net/foobar/%s/status (New)' % formdata.id in emails.emails['New arrivals']['payload']
     def test_create_formdata(pub):
     def test_create_formdata(two_pubs):
         FormDef.wipe()
         LoggedError.wipe()
         pub.tracking_code_class.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         two_pubs.tracking_code_class.wipe()
         target_formdef = FormDef()
         target_formdef.name = 'target form'
-...
         formdata.just_created()
         assert target_formdef.data_class().count() == 0
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         # check unconfigure action do nothing
         formdata.perform_workflow()
         assert target_formdef.data_class().count() == 0
-...
         formdata.perform_workflow()
         assert target_formdef.data_class().count() == 1
         errors = LoggedError.select()
         assert len(errors) == 2
         assert any('form_var_toto_string' in (error.exception_message or '') for error in errors)
         assert any('Missing field' in error.summary for error in errors)
         if two_pubs.is_using_postgresql():
             errors = two_pubs.loggederror_class.select()
             assert len(errors) == 2
             assert any('form_var_toto_string' in (error.exception_message or '') for error in errors)
             assert any('Missing field' in error.summary for error in errors)
         # no tracking code has been created
         created_formdata = target_formdef.data_class().select()[0]
         assert created_formdata.tracking_code is None
         assert pub.tracking_code_class.count() == 0
         assert two_pubs.tracking_code_class.count() == 0
         # now we want one
         target_formdef.enable_tracking_codes = True
         target_formdef.store()
-...
         assert target_formdef.data_class().count() == 1
         created_formdata = target_formdef.data_class().select()[0]
         assert created_formdata.tracking_code is not None
         assert pub.tracking_code_class.count() == 1
         assert pub.tracking_code_class.select()[0].formdef_id == target_formdef.id
         assert pub.tracking_code_class.select()[0].formdata_id == created_formdata.id
         assert two_pubs.tracking_code_class.count() == 1
         assert two_pubs.tracking_code_class.select()[0].formdef_id == target_formdef.id
         assert two_pubs.tracking_code_class.select()[0].formdata_id == str(created_formdata.id)
         create.condition = {'type': 'python', 'value': '1 == 2'}
         wf.store()
-...
         assert target_formdef.data_class().count() == 0
     def test_create_carddata(pub):
     def test_create_carddata(two_pubs):
         CardDef.wipe()
         FormDef.wipe()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         carddef = CardDef()
         carddef.name = 'My card'
-...
         assert carddef.data_class().count() == 1
         errors = LoggedError.select()
         assert len(errors) == 2
         assert any('form_var_undefined' in (error.exception_message or '') for error in errors)
         assert any('invalid date value' in (error.exception_message or '') for error in errors)
         if two_pubs.is_using_postgresql():
             errors = two_pubs.loggederror_class.select()
             assert len(errors) == 2
             assert any('form_var_undefined' in (error.exception_message or '') for error in errors)
             assert any('invalid date value' in (error.exception_message or '') for error in errors)
         formdata = formdef.data_class()()
         today = datetime.date.today()
         formdata.data = {'1': 'item1',
                          '1_display': 'item1',
                          '2': today}
                          '2': today.timetuple()}
         formdata.just_created()
         formdata.perform_workflow()
-...
         assert carddef.data_class().count() == 0
     def test_call_external_workflow_with_evolution_linked_object(pub):
     def test_call_external_workflow_with_evolution_linked_object(two_pubs):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         external_wf = Workflow(name='External Workflow')
         st1 = external_wf.add_status(name='New')
-...
         # remove external formdata
         perform_items([action], formdata)
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert external_formdef.data_class().count() == 0
         assert external_carddef.data_class().count() == 1
         # formdata is already deleted: cannot find it again
         perform_items([action], formdata)
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Could not find linked "External Form" object by id %s' % external_formdata.id
         assert logged_error.exception_class == 'KeyError'
         assert logged_error.status_item_id == action.id
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.summary == 'Could not find linked "External Form" object by id %s' % external_formdata.id
             assert logged_error.exception_class == 'KeyError'
             assert logged_error.status_item_id == action.id
         # try remove an unexisting carddef: do nothing
         unused_carddef = CardDef()
-...
         unused_carddef.fields = []
         unused_carddef.workflow = external_wf
         unused_carddef.store()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         action.slug = 'carddef:%s' % unused_carddef.url_name
         wf.store()
         perform_items([action], formdata)
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert external_formdef.data_class().count() == 0
         assert external_carddef.data_class().count() == 1
         # remove the right carddef
         action.slug = 'carddef:%s' % external_carddef.url_name
         wf.store()
         perform_items([action], formdata)
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert external_formdef.data_class().count() == 0
         assert external_carddef.data_class().count() == 0
     def test_call_external_workflow_with_data_sourced_object(pub):
     def test_call_external_workflow_with_data_sourced_object(two_pubs):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         if two_pubs.is_using_postgresql():
             two_pubs.loggederror_class.wipe()
         carddef_wf = Workflow(name='Carddef Workflow')
         carddef_wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(carddef_wf)
-...
         formdef.workflow = wf
         formdef.store()
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert carddef.data_class().count() == 1
         formdata = formdef.data_class()()
-...
         formdata.perform_workflow()
         perform_items([update_action], formdata)
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert carddef.data_class().count() == 1
         data = carddef.data_class().select()[0]
         assert data.data['bo0'] == '1'
-...
         assert data.data['bo0'] == '2'
         perform_items([delete_action], formdata)
         assert LoggedError.count() == 0
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 0
         assert carddef.data_class().count() == 0
         perform_items([delete_action], formdata)
         assert LoggedError.count() == 1
         logged_error = LoggedError.select()[0]
         assert logged_error.summary == 'Could not find linked "Data" object by id %s' % carddata.id
         assert logged_error.exception_class == 'KeyError'
         if two_pubs.is_using_postgresql():
             assert two_pubs.loggederror_class.count() == 1
             logged_error = two_pubs.loggederror_class.select()[0]
             assert logged_error.summary == 'Could not find linked "Data" object by id %s' % carddata.id
             assert logged_error.exception_class == 'KeyError'
     def test_call_external_workflow_with_parent_object(pub):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         # carddef workflow, with global action to increment a counter in its
         # backoffice fields.
-...
     def test_call_external_workflow_use_caller_variable(pub):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         # carddef workflow, with global action to set a value in a backoffice field
         carddef_wf = Workflow(name='Carddef Workflow')
-...
     def test_edit_carddata_with_data_sourced_object(pub):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         datasource = {
             'type': 'formula',
-...
     def test_edit_carddata_with_linked_object(pub):
         FormDef.wipe()
         CardDef.wipe()
         LoggedError.wipe()
         carddef = CardDef()
         carddef.name = 'Parent'

     from wcs import fields
     from wcs.formdef import FormDef
     from wcs.logged_errors import LoggedError
     from wcs.qommon.http_request import HTTPRequest
     from wcs.qommon.template import Template
     from wcs.wscalls import NamedWsCall
-...
     def test_named_wscall(pub):
         # create object
         NamedWsCall.wipe()
         wscall = NamedWsCall()
         wscall.name = 'Hello'
         wscall.request = {'url': 'http://example.net', 'qs_data': {'a': 'b'}}
-...
     @pytest.mark.parametrize('notify_on_errors', [True, False])
     @pytest.mark.parametrize('record_on_errors', [True, False])
     def test_webservice_on_error(http_requests, pub, emails, notify_on_errors, record_on_errors):
     def test_webservice_on_error_with_sql(http_requests, emails, notify_on_errors, record_on_errors):
         pub = create_temporary_pub(sql_mode=True)
         pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}
         pub.write_cfg()
         NamedWsCall.wipe()
         LoggedError.wipe()
         pub.loggederror_class.wipe()
         FormDef.wipe()
         wscall = NamedWsCall()
-...
             resp = get_app(pub).get('/foobar/')
             assert 'Foo Bar ' in resp.text
             assert emails.count() == 0
             assert LoggedError.count() == 0
             assert pub.loggederror_class.count() == 0
         for url_part in ['400', '400-json', '404', '404-json', '500', 'json-err1', 'json-errheader1']:
             status_code = 200
-...
             else:
                 assert emails.count() == 0
             if record_on_errors:
                 assert LoggedError.count() == 1
                 LoggedError.wipe()
                 assert pub.loggederror_class.count() == 1
                 pub.loggederror_class.wipe()
             else:
                 assert LoggedError.count() == 0
                 assert pub.loggederror_class.count() == 0

     import tempfile
     import random
     import psycopg2
     import pytest
     import shutil
     import sys
     import threading
     from wcs import sql, sessions, custom_views
-...
     from wcs.qommon import force_str
     import wcs
     import wcs.wsgi
     from wcs import publisher, compat
     from wcs.qommon.http_request import HTTPRequest
     from wcs import compat
     from wcs.users import User
     from wcs.tracking_code import TrackingCode
     import wcs.qommon.emails
-...
             pub.session_class = sql.Session
             pub.custom_view_class = sql.CustomView
             pub.snapshot_class = sql.Snapshot
             pub.loggederror_class = sql.LoggedError
             pub.is_using_postgresql = lambda: True
         else:
             pub.user_class = User
             pub.tracking_code_class = TrackingCode
             pub.session_class = sessions.BasicSession
             pub.custom_view_class = custom_views.CustomView
             pub.snapshot_class = None
             pub.is_using_postgresql = lambda: False
         pub.session_manager_class = sessions.StorageSessionManager
-...
             sql.do_session_table()
             sql.do_custom_views_table()
             sql.do_snapshots_table()
             sql.do_loggederrors_table()
             sql.do_meta_table()
             conn.close()

     from wcs.qommon import errors, get_cfg
     from wcs.qommon.misc import localstrftime
     from wcs.logged_errors import LoggedError
     class LoggedErrorDirectory(Directory):
         _q_exports = ['', 'delete', 'ack']
-...
             return redirect('.')
         def delete(self):
             self.error.remove_self()
             get_publisher().loggederror_class.remove_object(self.error.id)
             return redirect('..')
-...
         @classmethod
         def get_errors(cls, formdef_class=None, formdef_id=None, workflow_id=None):
             errors = []
             if not get_publisher().loggederror_class:
                 return errors
             if formdef_id and formdef_class:
                 errors = [e for e in LoggedError.get_with_indexed_value('formdef_id', formdef_id) if e.formdef_class == formdef_class.__name__]
                 errors = [
                     e for e in get_publisher().loggederror_class.get_with_indexed_value('formdef_id', formdef_id)
                     if e.formdef_class == formdef_class.__name__]
             elif workflow_id:
                 errors = LoggedError.get_with_indexed_value('workflow_id', workflow_id)
                 errors = get_publisher().loggederror_class.get_with_indexed_value('workflow_id', workflow_id)
             return list(errors)
         @classmethod
-...
         def _q_lookup(self, component):
             try:
                 error = LoggedError.get(component)
                 error = get_publisher().loggederror_class.get(component)
             except KeyError:
                 raise errors.TraversalError()
             get_response().breadcrumb.append(('logged-errors/', _('Logged Errors')))

                 if self.log_errors:
                     get_logger().warning('failed to evaluate %r (%r)', self, e)
                 if self.record_errors:
                     from wcs.logged_errors import LoggedError
                     summary = _('Failed to evaluate condition')
                     LoggedError.record(summary,
                                        formdata=self.context.get('formdata'),
                                        status_item=self.context.get('status_item'),
                                        expression=self.value, expression_type=self.type,
                                        exception=e)
                     get_publisher().record_error(
                         summary,
                         formdata=self.context.get('formdata'),
                         status_item=self.context.get('status_item'),
                         expression=self.value, expression_type=self.type,
                         exception=e)
                 raise RuntimeError()
         def evaluate_python(self, local_variables):

             url += param_name + '=' + urllib.quote(param_value)
             def find_item(items, name, value):
                 from wcs.logged_errors import LoggedError
                 for item in items:
                     if str(item.get(name)) == str(value):
                         return item
                 # not found
                 LoggedError.record(_('Could not find element by id "%s"') % value)
                 get_publisher().record_error(_('Could not find element by id "%s"') % value)
                 return None
             request = get_request()

                 except TemplateError:
                     return ('', explicit_lock)
                 except AttributeError as e:
                     from wcs.logged_errors import LoggedError
                     LoggedError.record(_('Failed to evaluate prefill on field "%s"') % self.label, formdef=getattr(self, 'formdef', None), exception=e)
                     get_publisher().record_error(
                         _('Failed to evaluate prefill on field "%s"') % self.label,
                         formdef=getattr(self, 'formdef', None),
                         exception=e)
                     return ('', explicit_lock)
             elif t == 'user' and user:

             if status_id == '_previous':
                 previous_status = self.pop_previous_marked_status()
                 if not previous_status:
                     from wcs.logged_errors import LoggedError
                     summary = _('Failed to compute previous status')
                     LoggedError.record(summary, formdata=self)
                     get_publisher().record_error(summary, formdata=self)
                     return
                 status_id = previous_status.id
             status = 'wf-%s' % status_id
-...
         def iter_target_datas(self, objectdef=None, object_type=None, status_item=None):
             # objectdef, object_type and status_item are provided when called from a workflow action
             from wcs.wf.create_formdata import LinkedFormdataEvolutionPart
             from wcs.logged_errors import LoggedError
             from .carddef import CardDef
             from .formdef import FormDef
-...
                         yield objectdef.data_class().get(target_id)
                     except KeyError as e:
                         # use custom error message depending on target type
                         LoggedError.record(_('Could not find linked "%(object_name)s" object by id %(object_id)s') % {
                             'object_name': objectdef.name, 'object_id': target_id},
                         get_publisher().record_error(
                             _('Could not find linked "%(object_name)s" object by id %(object_id)s') % {
                                 'object_name': objectdef.name, 'object_id': target_id},
                             formdata=self, status_item=status_item, exception=e)
                 else:
                     # inspect page

     import datetime
     from .qommon.misc import simplify
     from .qommon.xml_storage import XmlStorableObject
     from wcs.carddef import CardDef
     from wcs.formdef import FormDef
     from wcs.workflows import Workflow
     class LoggedError(XmlStorableObject):
     class LoggedError(object):
         _names = 'logged-errors'
         xml_root_node = 'error'
         _indexes = ['tech_id']
         _hashed_indexes = ['formdef_id', 'workflow_id']
         id = None
         tech_id = None
         summary = None
         formdef_class = 'FormDef'
         formdata_id = None
-...
         latest_occurence_timestamp = None
         acked = False
         # declarations for serialization
         XML_NODES = [
                 ('summary', 'str'), ('traceback', 'str'),
                 ('exception_class', 'str'), ('exception_message', 'str'),
                 ('expression', 'str'), ('expression_type', 'str'),
                 ('formdata_id', 'str'), ('formdef_id', 'str'), ('workflow_id', 'str'),
                 ('formdef_class', 'str'),
                 ('status_id', 'str'), ('status_item_id', 'str'),
                 ('occurences_count', 'int'),
                 ('first_occurence_timestamp', 'datetime'),
                 ('latest_occurence_timestamp', 'datetime'),
                 ('acked', 'bool')]
         @classmethod
         def record(cls, error_summary, plain_error_msg=None, formdata=None,
                    formdef=None, workflow=None, status=None, status_item=None,
-...
                 error.status_id = status.id
             error.first_occurence_timestamp = datetime.datetime.now()
             error.id = '%s-%s' % (
                     error.first_occurence_timestamp.strftime('%Y%m%d-%H%M%S'),
                     error.tech_id,
+                    )
             existing_error = cls.get_on_index(error.tech_id, 'tech_id', ignore_errors=True)
             if existing_error:
                 error = existing_error
             error.tech_id = error.build_tech_id()
             existing_errors = list(cls.get_with_indexed_value('tech_id', error.tech_id))
             if existing_errors:
                 error = existing_errors[0]
             error.occurences_count += 1
             error.latest_occurence_timestamp = datetime.datetime.now()
             error.store()
-...
         def record_exception(cls, error_summary, plain_error_msg, publisher):
             try:
                 context = publisher.substitutions.get_context_variables()
             except Exception as e:
             except Exception:
                 return
             formdata_id = context.get('form_number_raw')
             formdef_urlname = context.get('form_slug')
-...
                 klass = CardDef
             formdef = klass.get_by_urlname(formdef_urlname)
             formdata = formdef.data_class().get(formdata_id, ignore_errors=True)
             return cls.record(error_summary, plain_error_msg, formdata=formdata,
                        formdef=formdef, workflow=formdef.workflow)
             return cls.record(
                 error_summary, plain_error_msg, formdata=formdata,
                 formdef=formdef, workflow=formdef.workflow)
         @property
         def tech_id(self):
         def build_tech_id(self):
             tech_id = ''
             if self.formdef_id:
                 tech_id += '%s-' % self.formdef_id

     from .users import User
     from .tracking_code import TrackingCode
     from .logged_errors import LoggedError
     import pickle
-...
                 self.session_class = sql.Session
                 self.custom_view_class = sql.CustomView
                 self.snapshot_class = sql.Snapshot
                 self.loggederror_class = sql.LoggedError
                 sql.get_connection(new=True)
             else:
                 self.user_class = User
-...
                 self.session_class = sessions.BasicSession
                 self.custom_view_class = custom_views.CustomView
                 self.snapshot_class = None
                 self.loggederror_class = None
             self.session_manager_class = sessions.StorageSessionManager
             self.set_session_manager(self.session_manager_class(session_class=self.session_class))
-...
             sql.do_tracking_code_table()
             sql.do_custom_views_table()
             sql.do_snapshots_table()
             sql.do_loggederrors_table()
             sql.do_meta_table()
             from .formdef import FormDef
             from .carddef import CardDef
-...
         def log_internal_error(self, error_summary, plain_error_msg, record=False, notify=True):
             tech_id = None
             if record:
                 logged_exception = LoggedError.record_exception(
             if record and self.loggederror_class:
                 logged_exception = self.loggederror_class.record_exception(
                         error_summary, plain_error_msg, publisher=self)
                 if logged_exception:
                     tech_id = logged_exception.tech_id
-...
                 # this could happen on file descriptor exhaustion
                 pass
         def record_error(self, *args, **kwargs):
             if self.loggederror_class:
                 self.loggederror_class.record(*args, **kwargs)
         def apply_global_action_timeouts(self):
             from wcs.workflows import Workflow, WorkflowGlobalActionTimeoutTrigger
             for workflow in Workflow.select():

     import wcs.carddata
     import wcs.custom_views
     import wcs.formdata
     import wcs.logged_errors
     import wcs.snapshots
     import wcs.tracking_code
     import wcs.users
-...
         cur.close()
     def do_loggederrors_table(concurrently=False):
         conn, cur = get_connection_and_cursor()
         table_name = 'loggederrors'
         cur.execute('''SELECT COUNT(*) FROM information_schema.tables
                         WHERE table_schema = 'public'
                           AND table_name = %s''', (table_name,))
         if cur.fetchone()[0] == 0:
             cur.execute('''CREATE TABLE %s (id SERIAL PRIMARY KEY,
                                             tech_id VARCHAR UNIQUE,
                                             summary VARCHAR,
                                             formdef_class VARCHAR,
                                             formdata_id VARCHAR,
                                             formdef_id VARCHAR,
                                             workflow_id VARCHAR,
                                             status_id VARCHAR,
                                             status_item_id VARCHAR,
                                             expression VARCHAR,
                                             expression_type VARCHAR,
                                             traceback TEXT,
                                             exception_class VARCHAR,
                                             exception_message VARCHAR,
                                             occurences_count INTEGER,
                                             first_occurence_timestamp TIMESTAMP WITH TIME ZONE,
                                             latest_occurence_timestamp TIMESTAMP WITH TIME ZONE,
                                             acked BOOLEAN
                                             )''' % table_name)
         cur.execute('''SELECT column_name FROM information_schema.columns
                         WHERE table_schema = 'public'
                           AND table_name = %s''', (table_name,))
         existing_fields = set([x[0] for x in cur.fetchall()])
         needed_fields = set([x[0] for x in LoggedError._table_static_fields])
         # delete obsolete fields
         for field in (existing_fields - needed_fields):
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         create_index = 'CREATE INDEX'
         if concurrently:
             create_index = 'CREATE INDEX CONCURRENTLY'
         # build indexes
         existing_indexes = set()
         cur.execute('''SELECT indexname
                          FROM pg_indexes
                         WHERE schemaname = 'public'
                           AND tablename = %s''', (table_name,))
         existing_indexes = set([x[0] for x in cur.fetchall()])
         for attr in ('formdef_id', 'workflow_id'):
             if not table_name + '_' + attr + '_idx' in existing_indexes:
                 cur.execute('%(create_index)s %(table_name)s_%(attr)s_idx ON %(table_name)s (%(attr)s)' % {
                     'create_index': create_index,
                     'table_name': table_name,
                     'attr': attr})
         conn.commit()
         cur.close()
     @guard_postgres
     def do_meta_table(conn=None, cur=None, insert_current_sql_level=True):
         own_conn = False
-...
             return cls.get(row[0])
     class LoggedError(SqlMixin, wcs.logged_errors.LoggedError):
         _table_name = 'loggederrors'
         _table_static_fields = [
             ('id', 'serial'),
             ('tech_id', 'varchar'),
             ('summary', 'varchar'),
             ('formdef_class', 'varchar'),
             ('formdata_id', 'varchar'),
             ('formdef_id', 'varchar'),
             ('workflow_id', 'varchar'),
             ('status_id', 'varchar'),
             ('status_item_id', 'varchar'),
             ('expression', 'varchar'),
             ('expression_type', 'varchar'),
             ('traceback', 'text'),
             ('exception_class', 'varchar'),
             ('exception_message', 'varchar'),
             ('occurences_count', 'integer'),
             ('first_occurence_timestamp', 'timestamptz'),
             ('latest_occurence_timestamp', 'timestamptz'),
             ('acked', 'boolean'),
+        ]
         _numerical_id = False
         @guard_postgres
         @invalidate_substitution_cache
         def store(self):
             sql_dict = {x: getattr(self, x) for x, y in self._table_static_fields}
             conn, cur = get_connection_and_cursor()
             if not self.id:
                 column_names = [x for x in sql_dict.keys() if x != 'id']
                 sql_statement = '''INSERT INTO %s (%s)
                                    VALUES (%s)
                                    RETURNING id''' % (
                                            self._table_name,
                                            ', '.join(column_names),
                                            ', '.join(['%%(%s)s' % x for x in column_names]))
                 cur.execute(sql_statement, sql_dict)
                 self.id = cur.fetchone()[0]
             else:
                 column_names = sql_dict.keys()
                 sql_statement = '''UPDATE %s SET %s WHERE id = %%(id)s RETURNING id''' % (
                                            self._table_name,
                                            ', '.join(['%s = %%(%s)s' % (x, x) for x in column_names]))
                 cur.execute(sql_statement, sql_dict)
                 if cur.fetchone() is None:
                     raise AssertionError()
             conn.commit()
             cur.close()
         @classmethod
         def _row2ob(cls, row, **kwargs):
             o = cls()
             for field, value in zip(cls._table_static_fields, tuple(row)):
                 if field[1] in ('varchar', 'text'):
                     setattr(o, field[0], str_encode(value))
                 else:
                     setattr(o, field[0], value)
             return o
         @classmethod
         def get_data_fields(cls):
             return []
         @classmethod
         @guard_postgres
         def fix_sequences(cls):
             conn, cur = get_connection_and_cursor()
             sql_statement = '''select max(id) from %s''' % cls._table_name
             cur.execute(sql_statement)
             max_id = cur.fetchone()[0]
             if max_id is not None:
                 sql_statement = '''ALTER SEQUENCE %s_id_seq RESTART %s''' % (
                         cls._table_name, max_id+1)
                 cur.execute(sql_statement)
             conn.commit()
             cur.close()
     class classproperty(object):
         def __init__(self, f):
             self.f = f
-...
     # latest migration, number + description (description is not used
     # programmaticaly but will make sure git conflicts if two migrations are
     # separately added with the same number)
     SQL_LEVEL = (46, 'add index on formdata(status) - fix')
     SQL_LEVEL = (47, 'use SQL to store LoggedError')
     def migrate_global_views(conn, cur):
-...
         if sql_level < 42:
             # 42: create snapshots table
             do_snapshots_table()
         if sql_level < 47:
             # 47: store LoggedErrors in SQL
             do_loggederrors_table()
         cur.execute('''UPDATE wcs_meta SET value = %s WHERE key = %s''', (
             str(SQL_LEVEL[0]), 'sql_level'))

     from pyproj import Geod
     from .logged_errors import LoggedError
     from .qommon import misc, force_str, _
     from .qommon.evalutils import make_datetime
     from .qommon.templatetags.qommon import parse_datetime
-...
                     from wcs import sql
                     criteria = Equal(sql.get_field_id(field), value)
                     return self._clone(self._criterias + [criteria])
             LoggedError.record(_('Invalid filter "%s"') % self.pending_attr, formdata=self._formdata)
             get_publisher().record_error(_('Invalid filter "%s"') % self.pending_attr, formdata=self._formdata)
             return self.none()
         def __getattr__(self, attribute):

     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     import xml.etree.ElementTree as ET
     from quixote import get_publisher
     from quixote.html import htmltext
     from ..qommon import _, N_
-...
                     try:
                         new_value = formdef_field.convert_value_from_anything(new_value)
                     except ValueError as e:
                         from wcs.logged_errors import LoggedError
                         summary = _('Failed to convert %(class)s value to %(kind)s field (%(id)s)') % {
                                         'class': type(new_value),
                                         'kind': _(getattr(formdef_field, 'description', 'unknown')),
                                         'id': field['field_id'],
+                                    }
                         expression_dict = self.get_expression(field['value'])
                         LoggedError.record(summary, formdata=formdata, status_item=self,
                                            expression=expression_dict['value'],
                                            expression_type=expression_dict['type'],
                                            exception=e)
                         get_publisher().record_error(
                             summary, formdata=formdata, status_item=self,
                             expression=expression_dict['value'],
                             expression_type=expression_dict['type'],
                             exception=e)
                         continue
                 formdata.data['%s' % field['field_id']] = new_value

                                  SingleSelectWidget, ComputedExpressionWidget,
                                  CheckboxWidget, VarnameWidget, HtmlWidget)
     from wcs.logged_errors import LoggedError
     from wcs.workflows import WorkflowStatusItem, register_item_class
     from wcs.formdef import FormDef
-...
                             field=dest_field,
                             value=src.data.get(field.id))
                     except Exception as e:
                         LoggedError.record(_('Could not copy field by varname for "%s"') % field.varname,
                                            formdata=src, status_item=self, exception=e)
                         get_publisher().record_error(
                             _('Could not copy field by varname for "%s"') % field.varname,
                             formdata=src, status_item=self, exception=e)
             # field.id can be serialized to xml, so we must always convert them to
             # str when matching
-...
                         value=value)
                 except Exception as e:
                     expression = self.get_expression(mapping.expression)
                     LoggedError.record(_('Could not assign value to field "%s"') % dest_field.label,
                                        formdata=src, status_item=self,
                                        expression=expression['value'], expression_type=expression['type'],
                                        exception=e)
                     get_publisher().record_error(
                         _('Could not assign value to field "%s"') % dest_field.label,
                         formdata=src, status_item=self,
                         expression=expression['value'], expression_type=expression['type'],
                         exception=e)
             if missing_fields:
                 summary = _('Missing field %r') % missing_fields
                 LoggedError.record(summary, formdata=src, status_item=self)
                 get_publisher().record_error(summary, formdata=src, status_item=self)
         def _set_value(self, formdata, field, value):
             if field.convert_value_from_anything:

             return htmltext('<ul class="rules">%s</ul>') % htmltext('').join(result)
         def perform(self, formdata):
             from wcs.logged_errors import LoggedError
             if not formdata.workflow_roles:
                 formdata.workflow_roles = {}
-...
                     return
                 new_role_id = self.get_computed_role_id(self.role_id)
                 if not new_role_id:
                     LoggedError.record(_('error in dispatch, missing role (%s)') % self.role_id, formdata=formdata)
                     get_publisher().record_error(_('error in dispatch, missing role (%s)') % self.role_id, formdata=formdata)
             elif self.dispatch_type == 'automatic':
                 if not (self.role_key and self.variable and self.rules):
                     return
-...
             if new_role_id:
                 if not Role.has_key(new_role_id):
                     LoggedError.record(_('error in dispatch, missing role (%s)') % new_role_id, formdata=formdata)
                     get_publisher().record_error(_('error in dispatch, missing role (%s)') % new_role_id, formdata=formdata)
                 else:
                     formdata.workflow_roles[self.role_key] = str(new_role_id)
                     formdata.store()

     from wcs.qommon import _
     from wcs.qommon.form import SingleSelectWidget
     from wcs.logged_errors import LoggedError
     from wcs.workflows import WorkflowStatusItem, perform_items, register_item_class
     from wcs.workflows import WorkflowGlobalActionWebserviceTrigger, Workflow
     from wcs.carddef import CardDef
-...
             trigger = self.get_trigger(objectdef.workflow)
             if not trigger:
                 LoggedError.record(_('No trigger with id "%s" found in workflow') % self.trigger_id,
                         formdata=formdata, status_item=self)
                 get_publisher().record_error(
                     _('No trigger with id "%s" found in workflow') % self.trigger_id,
                     formdata=formdata, status_item=self)
                 return
             class CallerSource:

                 try:
                     target = self.parent.parent.get_status(value)
                 except KeyError:
                     from wcs.logged_errors import LoggedError
                     message = _('reference to invalid status in workflow %(workflow)s, status %(status)s, item %(item)s') % {
                         'workflow': self.parent.parent.name,
                         'status': self.parent.name,
                         'item': self.description,
+                    }
                     LoggedError.record(message, workflow=self.parent.parent)
                     get_publisher().record_error(message, workflow=self.parent.parent)
                     continue
                 targets.append(target)
             return targets

             vars.update(context or {})
             def log_exception(exception):
                 from wcs.logged_errors import LoggedError
                 if expression['type'] == 'template':
                     summary = _('Failed to compute template')
                 else:
                     summary = _('Failed to compute Python expression')
                 LoggedError.record(summary, formdata=formdata, status_item=status_item,
                                    expression=expression['value'],
                                    expression_type=expression['type'],
                                    exception=exception)
                 get_publisher().record_error(
                     summary, formdata=formdata, status_item=status_item,
                     expression=expression['value'],
                     expression_type=expression['type'],
                     exception=exception)
             if expression['type'] == 'template':
                 try:
-...
             targets = [x for x in self.parent.parent.possible_status if x.id == self.status]
             if not targets and formdata:  # do not log in presentation context: formdata is needed
                 from wcs.logged_errors import LoggedError
                 message = _('reference to invalid status %(target)s in status %(status)s, '
                             'action %(status_item)s') % {
                                 'target': self.status,
                                 'status': self.parent.name,
                                 'status_item': _(self.description)
+                            }
                 LoggedError.record(message, formdata=formdata, status_item=self)
                 get_publisher().record_error(message, formdata=formdata, status_item=self)
             return targets
-...
                     subject = mail_template.subject
                     extra_attachments = mail_template.attachments
                 else:
                     from wcs.logged_errors import LoggedError
                     message = _('reference to invalid mail template %(mail_template)s in status %(status)s') % {
                                 'status': self.parent.name,
                                 'mail_template': self.mail_template,
+                                }
                     LoggedError.record(message, formdata=formdata, status_item=self)
                     get_publisher().record_error(message, formdata=formdata, status_item=self)
                     return
             try:
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.