0001-factorize-WorklowStatusItem.compute.patch

Benjamin Dauvergne, 17 février 2017 11:03

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

Subject: [PATCH 1/4] factorize WorklowStatusItem.compute()

 tests/test_workflows.py     | 19 +++++++++----------
 wcs/qommon/misc.py          | 30 ++++++++++++++++++++++++++++++
 wcs/wf/backoffice_fields.py |  3 ++-
 wcs/wf/geolocate.py         |  8 ++++----
 wcs/wf/jump.py              |  3 ++-
 wcs/wf/profile.py           |  4 ++--
 wcs/workflows.py            | 42 +++++++-----------------------------------
 wcs/wscalls.py              |  8 ++++----
 8 files changed, 60 insertions(+), 57 deletions(-)

     from quixote import cleanup, get_response
     from wcs.qommon.http_request import HTTPRequest
     from qommon.form import *
     from qommon.misc import compute
     from wcs.formdef import FormDef
     from wcs import sessions
-...
         formdata.store()
         pub.substitutions.feed(formdata)
         item = JumpWorkflowStatusItem()
         # straight string
         assert item.compute('blah') == 'blah'
         assert compute('blah') == 'blah'
         # ezt string
         assert item.compute('[form_var_foo]') == 'hello'
         assert compute('[form_var_foo]') == 'hello'
         # ezt string, but not ezt asked
         assert item.compute('[form_var_foo]', do_ezt=False) == '[form_var_foo]'
         assert compute('[form_var_foo]', do_ezt=False) == '[form_var_foo]'
         # ezt string, with an error
         assert item.compute('[end]', raises=False) == '[end]'
         assert compute('[end]', raises=False) == '[end]'
         with pytest.raises(Exception):
             item.compute('[end]', raises=True)
             compute('[end]', raises=True)
         # python expression
         assert item.compute('=form_var_foo') == 'hello'
         assert compute('=form_var_foo') == 'hello'
         # python expression, with an error
         assert item.compute('=1/0', raises=False) == '=1/0'
         assert compute('=1/0', raises=False) == '=1/0'
         with pytest.raises(Exception):
             item.compute('=1/0', raises=True)
             compute('=1/0', raises=True)
     def test_jump_nothing(pub):
         FormDef.wipe()

         for chunk in iter(read_chunk, ''):
             digest.update(chunk)
         return digest.hexdigest()
     def compute(expr, do_ezt=True, raises=False, context=None):
         if not isinstance(expr, basestring):
             return expr
         if not expr.startswith('=') and not do_ezt:
             return expr
         vars = get_publisher().substitutions.get_context_variables()
         vars.update(context or {})
         if not expr.startswith('='):
             try:
                 processor = ezt.Template(compress_whitespace=False)
                 processor.parse(expr)
                 fd = StringIO()
                 processor.generate(fd, vars)
                 return fd.getvalue()
             except ezt.EZTException:
                 if raises:
                     raise
                 return expr
         try:
             return eval(expr[1:], get_publisher().get_global_eval_dict(), vars)
         except:
             if raises:
                 raise
             return expr

     from quixote import get_publisher
     from qommon import get_logger
     from qommon.misc import compute
     from qommon.form import (WidgetListAsTable, CompositeWidget, SingleSelectWidget,
             ComputedExpressionWidget, PicklableUpload)
     from wcs.fields import WidgetField
-...
                     continue
                 try:
                     new_value = self.compute(field['value'], raises=True)
                     new_value = compute(field['value'], raises=True)
                 except:
                     get_publisher().notify_of_exception(sys.exc_info())
                     continue

     from qommon import get_logger
     from qommon.form import RadiobuttonsWidget, ComputedExpressionWidget, CheckboxWidget
     from qommon.misc import http_get_page
     from qommon.misc import http_get_page, compute
     from wcs.workflows import WorkflowStatusItem, register_item_class
     class GeolocateWorkflowStatusItem(WorkflowStatusItem):
-...
                 nominatim_url = 'http://nominatim.openstreetmap.org'
             try:
                 address = self.compute(self.address_string, raises=True)
                 address = compute(self.address_string, raises=True)
             except Exception, e:
                 get_logger().error('error in template for address string [%r]', e)
                 return
-...
             return {'lon': float(coords['lon']), 'lat': float(coords['lat'])}
         def geolocate_map_variable(self, formdata):
             value = self.compute(self.map_variable)
             value = compute(self.map_variable)
             if not value:
                 return
-...
                 get_logger().error('error geolocating from file (missing PIL)')
                 return
             value = self.compute(self.photo_variable)
             value = compute(self.photo_variable)
             if not hasattr(value, 'get_file_pointer'):
                 get_logger().error('error geolocating from photo, invalid variable')
                 return

     from qommon import errors
     from qommon.publisher import get_publisher_class
     from qommon.cron import CronJob
     from qommon.misc import compute
     from wcs.workflows import Workflow, WorkflowStatusJumpItem, register_item_class
     from wcs.api import get_user_from_api_query_string, is_url_signed
-...
                 must_jump = must_jump and triggered
             if self.timeout:
                 timeout = int(self.compute(self.timeout))
                 timeout = int(compute(self.timeout))
                 if formdata.evolution:
                     last = formdata.evolution[-1].time
                 else:

     from qommon.form import (CompositeWidget, SingleSelectWidget,
             WidgetListAsTable, ComputedExpressionWidget)
     from qommon.ident.idp import is_idp_managing_user_attributes
     from qommon.misc import http_patch_request
     from qommon.misc import http_patch_request, compute
     from qommon.publisher import get_cfg, get_logger
     from wcs.api_utils import sign_url, get_secret_and_orig, MissingSecret
-...
             get_publisher().substitutions.feed(formdata)
             new_data = {}
             for field in self.fields:
                 new_data[field.get('field_id')] = self.compute(field.get('value'))
                 new_data[field.get('field_id')] = compute(field.get('value'))
             user_formdef = get_publisher().user_class.get_formdef()
             new_user_data = {}

     from quixote import get_request, redirect
     from qommon.misc import C_, get_as_datetime, file_digest
     from qommon.misc import C_, get_as_datetime, file_digest, compute
     from qommon.storage import StorableObject, atomic_write
     from qommon.form import *
     from qommon.humantime import seconds2humanduration
-...
                         value = getattr(self, '%s_parse' % f)(value)
                     setattr(self, f, value)
         @classmethod
         def compute(cls, var, do_ezt=True, raises=False):
             if not isinstance(var, basestring):
                 return var
             if not var.startswith('=') and not do_ezt:
                 return var
             vars = get_publisher().substitutions.get_context_variables()
             if not var.startswith('='):
                 try:
                     processor = ezt.Template(compress_whitespace=False)
                     processor.parse(var)
                     fd = StringIO()
                     processor.generate(fd, vars)
                     return fd.getvalue()
                 except ezt.EZTException:
                     if raises:
                         raise
                     return var
             try:
                 return eval(var[1:], get_publisher().get_global_eval_dict(), vars)
             except:
                 if raises:
                     raise
                 return var
         def get_substitution_variables(self, formdata):
             return {}
-...
                 return _('Change Status (not completed)')
         def fill_form(self, form, formdata, user):
             form.add_submit('button%s' % self.id, self.compute(self.label))
             form.add_submit('button%s' % self.id, compute(self.label))
             if self.require_confirmation:
                 get_response().add_javascript(['jquery.js', '../../i18n.js', 'qommon.js'])
                 widget = form.get_widget('button%s' % self.id)
-...
             url = formdata.get_url()
             try:
                 mail_body = template_on_formdata(formdata, self.compute(self.body, do_ezt=False))
                 mail_body = template_on_formdata(formdata, compute(self.body, do_ezt=False))
             except ezt.EZTException:
                 get_logger().error('error in template for email body [%s], mail could not be generated' % url)
                 return
             try:
                 mail_subject = template_on_formdata(formdata, self.compute(self.subject, do_ezt=False))
                 mail_subject = template_on_formdata(formdata, compute(self.subject, do_ezt=False))
             except ezt.EZTException:
                 get_logger().error('error in template for email subject [%s], mail could not be generated' % url)
                 return
-...
             addresses = []
             for dest in self.to:
                 try:
                     dest = self.compute(dest, raises=True)
                     dest = compute(dest, raises=True)
                 except:
                     continue
-...
             if not self.body:
                 return
             destinations = [self.compute(x) for x in self.to]
             destinations = [compute(x) for x in self.to]
             destinations = [x for x in destinations if x] # ignore empty elements
             if not destinations:
                 return
             try:
                 sms_body = template_on_formdata(formdata, self.compute(self.body, do_ezt=False))
                 sms_body = template_on_formdata(formdata, compute(self.body, do_ezt=False))
             except ezt.EZTException:
                 url = formdata.get_url()
                 get_logger().error('error in template for sms [%s], sms could not be generated' % url)

     from quixote import get_publisher
     from qommon.misc import (simplify, http_get_page, http_post_request,
             get_variadic_url, JSONEncoder, json_loads)
             get_variadic_url, JSONEncoder, json_loads, compute)
     from qommon.xml_storage import XmlStorableObject
     from qommon.form import (CompositeWidget, StringWidget, WidgetDict,
             ComputedExpressionWidget, RadiobuttonsWidget, CheckboxWidget)
-...
             qs = list(urlparse.parse_qsl(parsed.query))
             for key, value in qs_data.iteritems():
                 try:
                     value = WorkflowStatusItem.compute(value, raises=True)
                     value = compute(value, raises=True)
                     value = str(value)
                 except:
                     get_publisher().notify_of_exception(sys.exc_info())
-...
             url = urlparse.urlunparse(parsed[:4] + (qs,) + parsed[5:6])
         if request_signature_key:
             signature_key = WorkflowStatusItem.compute(request_signature_key)
             signature_key = compute(request_signature_key)
             if signature_key:
                 url = sign_url(url, signature_key)
-...
             payload = {}
             for (key, value) in post_data.items():
                 try:
                     payload[key] = WorkflowStatusItem.compute(value, raises=True)
                     payload[key] = compute(value, raises=True)
                 except:
                     get_publisher().notify_of_exception(sys.exc_info())
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-factorize-WorklowStatusItem.compute.patch