wscall: fix post_data xml export/import (#8172)

tests/test_workflow_import.py

@@ -7,6 +7,7 @@ from quixote import cleanup
 from wcs import publisher
 
 from wcs.workflows import Workflow, CommentableWorkflowStatusItem
+from wcs.wf.wscall import WebserviceCallStatusItem
 from wcs.roles import Role
 from wcs.fields import StringField
 
@@ -277,3 +278,26 @@ def test_variables_formdef():
     wf.variables_formdef.fields.append(StringField(label='Test', type='string'))
     wf2 = assert_import_export_works(wf)
     assert wf2.variables_formdef.fields[0].label == 'Test'
+
+
+def test_wscall_action():
+    wf = Workflow(name='status')
+    st1 = wf.add_status('Status1', 'st1')
+
+    wscall = WebserviceCallStatusItem()
+    wscall.id = '_wscall'
+    wscall.url = 'http://test/'
+    wscall.varname = 'varname'
+    wscall.post = False
+    wscall.request_signature_key = 'key'
+    wscall.post_data = {'one': '1', 'two': '=2', 'good:name': 'ok'}
+    st1.items.append(wscall)
+    wscall.parent = st1
+
+    wf2 = assert_import_export_works(wf)
+    wscall2 = wf2.possible_status[0].items[0]
+    assert wscall2.url == 'http://test/'
+    assert wscall2.varname == 'varname'
+    assert wscall2.post == False
+    assert wscall2.request_signature_key == 'key'
+    assert wscall2.post_data == {'one': '1', 'two': '=2', 'good:name': 'ok'}

wcs/wf/wscall.py

@@ -16,6 +16,7 @@
 
 import json
 import sys
+import xml.etree.ElementTree as ET
 
 from qommon.form import *
 from qommon.misc import http_get_page, http_post_request, get_variadic_url, JSONEncoder
@@ -122,4 +123,33 @@ class WebserviceCallStatusItem(WorkflowStatusItem):
         if (status // 100) not in (2, 3):
             raise Exception("Call to webservice gave %s as status code" % status)
 
+    def post_data_export_to_xml(self, xml_item, charset, include_id=False):
+        if not self.post_data:
+            return
+        el = ET.SubElement(xml_item, 'post_data')
+        for (key, value) in self.post_data.items():
+            item = ET.SubElement(el, 'item')
+            if type(key) is unicode:
+                ET.SubElement(item, 'name').text = key
+            elif type(key) is str:
+                ET.SubElement(item, 'name').text = unicode(key, charset, 'replace')
+            else:
+                raise AssertionError('unknown type for key (%r)' % key)
+            if type(value) is unicode:
+                ET.SubElement(item, 'value').text = value
+            elif type(value) is str:
+                ET.SubElement(item, 'value').text = unicode(value, charset, 'replace')
+            else:
+                raise AssertionError('unknown type for value (%r)' % key)
+
+
+    def post_data_init_with_xml(self, elem, charset, include_id=False):
+        if elem is None:
+            return
+        self.post_data = {}
+        for item in elem.findall('item'):
+            key = item.find('name').text.encode(charset)
+            value = item.find('value').text.encode(charset)
+            self.post_data[key] = value
+
 register_item_class(WebserviceCallStatusItem)