inspect résultat test (#87244) #1176
|
@ -14,6 +14,7 @@ from wcs.qommon.http_request import HTTPRequest
|
|||
from wcs.qommon.upload_storage import PicklableUpload
|
||||
from wcs.testdef import TestDef, TestResult, WebserviceResponse
|
||||
from wcs.workflow_tests import WorkflowTests
|
||||
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
|
||||
from wcs.wscalls import NamedWsCall
|
||||
|
||||
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
|
||||
|
@ -999,6 +1000,87 @@ def test_tests_result_error_field(pub):
|
|||
assert 'deleted' in resp.text
|
||||
|
||||
|
||||
def test_tests_result_inspect(pub):
|
||||
user = create_superuser(pub)
|
||||
|
||||
role = pub.role_class(name='test role')
|
||||
role.store()
|
||||
user.roles = [role.id]
|
||||
user.store()
|
||||
|
||||
workflow = Workflow(name='Workflow One')
|
||||
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
|
||||
workflow.backoffice_fields_formdef.fields = [
|
||||
fields.StringField(id='1', label='Text BO', varname='text_bo'),
|
||||
]
|
||||
|
||||
new_status = workflow.add_status(name='New status')
|
||||
set_backoffice_fields = new_status.add_action('set-backoffice-fields')
|
||||
set_backoffice_fields.fields = [{'field_id': '1', 'value': 'goodbye'}]
|
||||
|
||||
jump = new_status.add_action('choice')
|
||||
jump.label = 'Loop on status'
|
||||
jump.status = new_status.id
|
||||
jump.by = [role.id]
|
||||
|
||||
workflow.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test title'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='Text Field', varname='text'),
|
||||
]
|
||||
formdef.workflow = workflow
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
formdata.data['0'] = 'hello'
|
||||
|
||||
testdef = TestDef.create_from_formdata(formdef, formdata)
|
||||
testdef.name = 'First test'
|
||||
testdef.agent_id = user.id
|
||||
testdef.workflow_tests.actions = [
|
||||
workflow_tests.ButtonClick(id='1', button_name='Loop on status'),
|
||||
]
|
||||
testdef.store()
|
||||
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/forms/1/tests/results/run')
|
||||
result_url = resp.location
|
||||
resp = resp.follow()
|
||||
resp = resp.click('Display inspect')
|
||||
|
||||
assert 'form_var_text' in resp.text
|
||||
assert 'form_var_text_bo' in resp.text
|
||||
|
||||
assert [x.text_content() for x in resp.pyquery('div#inspect-timeline a')] == [
|
||||
'New status',
|
||||
'Backoffice Data',
|
||||
'Action button - Manual Jump Loop on status',
|
||||
'Backoffice Data',
|
||||
]
|
||||
|
||||
resp.form['django-condition'] = 'form_var_text == "hello"'
|
||||
resp = resp.form.submit()
|
||||
assert 'Condition result' in resp.text
|
||||
assert 'result-true' in resp.text
|
||||
|
||||
resp.form['django-condition'] = 'form_var_text_bo == "goodbye"'
|
||||
resp = resp.form.submit()
|
||||
assert 'Condition result' in resp.text
|
||||
assert 'result-true' in resp.text
|
||||
|
||||
# check inspect is not accessible for old results
|
||||
light_test_result = TestResult.select()[-1]
|
||||
test_result = TestResult.get(light_test_result.id)
|
||||
del test_result.results[0]['formdata']
|
||||
test_result.store()
|
||||
|
||||
resp = app.get(result_url)
|
||||
assert 'Display inspect' not in resp.text
|
||||
|
||||
|
||||
def test_tests_run_order(pub):
|
||||
create_superuser(pub)
|
||||
|
||||
|
|
|
@ -25,6 +25,7 @@ from quixote.directory import Directory
|
|||
from quixote.html import TemplateIO, htmltext
|
||||
|
||||
from wcs.admin.workflow_tests import WorkflowTestsDirectory
|
||||
from wcs.api import posted_json_data_to_formdata_data
|
||||
from wcs.backoffice.management import FormBackofficeEditPage, FormBackOfficeStatusPage
|
||||
from wcs.backoffice.pagination import pagination_links
|
||||
from wcs.forms.common import FormStatusPage
|
||||
|
@ -43,6 +44,7 @@ from wcs.qommon.form import (
|
|||
from wcs.sql_criterias import Equal, Null, StrictNotEqual
|
||||
from wcs.testdef import TestDef, TestError, TestResult, WebserviceResponse
|
||||
from wcs.workflow_tests import WorkflowTestError
|
||||
from wcs.workflow_traces import WorkflowTrace
|
||||
|
||||
|
||||
class TestEditPage(FormBackofficeEditPage):
|
||||
|
@ -435,7 +437,7 @@ class TestsDirectory(Directory):
|
|||
|
||||
|
||||
class TestResultDetailPage(Directory):
|
||||
_q_exports = ['']
|
||||
_q_exports = ['', 'inspect', ('inspect-tool', 'inspect_tool')]
|
||||
|
||||
def __init__(self, component, test_result, formdef):
|
||||
self.result_index = component
|
||||
|
@ -509,6 +511,55 @@ class TestResultDetailPage(Directory):
|
|||
field.url = self.formdef.get_field_admin_url(field)
|
||||
return field
|
||||
|
||||
def inspect(self):
|
||||
formdata_json = json.loads(self.result['formdata'])
|
||||
formdata = self.import_formdata_from_json(formdata_json)
|
||||
|
||||
return FormBackOfficeStatusPage(self.formdef, formdata).inspect()
|
||||
|
||||
def inspect_tool(self):
|
||||
formdata_json = json.loads(self.result['formdata'])
|
||||
formdata = self.import_formdata_from_json(formdata_json)
|
||||
|
||||
return FormBackOfficeStatusPage(self.formdef, formdata).inspect_tool()
|
||||
|
||||
def import_formdata_from_json(self, formdata_json):
|
||||
formdata = self.formdef.data_class()()
|
||||
formdata.receipt_time = formdata_json['receipt_time']
|
||||
formdata.user_id = formdata_json.get('user', {}).get('id')
|
||||
formdata.digests = formdata_json['digests']
|
||||
formdata.backoffice_submission = formdata_json['submission']['backoffice']
|
||||
formdata.submission_channel = formdata_json['submission']['channel']
|
||||
formdata.submission_agent_id = formdata_json['submission'].get('agent', {}).get('id')
|
||||
formdata.geolocations = formdata_json.get('geolocations')
|
||||
formdata.criticality_level = formdata_json['criticality_level']
|
||||
formdata.anonymised = formdata_json['anonymised']
|
||||
formdata.set_auto_fields()
|
||||
|
||||
# load fields
|
||||
formdata.data = posted_json_data_to_formdata_data(self.formdef, formdata_json['fields'])
|
||||
|
||||
# load backoffice fields if any
|
||||
if 'fields' in (formdata_json.get('workflow') or {}):
|
||||
backoffice_data_dict = posted_json_data_to_formdata_data(
|
||||
self.formdef, formdata_json['workflow']['fields']
|
||||
)
|
||||
formdata.data.update(backoffice_data_dict)
|
||||
|
||||
# set status
|
||||
formdata.status = formdata_json['workflow']['real_status']['id']
|
||||
|
||||
formdata.workflow_traces = [
|
||||
WorkflowTrace.import_from_json_dict(x) for x in formdata_json['workflow_traces']
|
||||
]
|
||||
|
||||
def get_workflow_traces():
|
||||
return formdata.workflow_traces
|
||||
|
||||
formdata.get_workflow_traces = get_workflow_traces
|
||||
|
||||
return formdata
|
||||
|
||||
|
||||
class TestResultPage(Directory):
|
||||
_q_exports = ['']
|
||||
|
@ -627,6 +678,11 @@ class TestsAfterJob(AfterJob):
|
|||
test.error = str(e)
|
||||
test.exception = e
|
||||
|
||||
formdata_json = test.formdata.get_json_export_dict()
|
||||
formdata_json['criticality_level'] = test.formdata.criticality_level
|
||||
formdata_json['anonymised'] = test.formdata.anonymised
|
||||
formdata_json['workflow_traces'] = [x.get_json_export_dict() for x in test.formdata.workflow_traces]
|
||||
|
||||
test_result = TestResult()
|
||||
test_result.object_type = objectdef.get_table_name()
|
||||
test_result.object_id = objectdef.id
|
||||
|
@ -638,6 +694,7 @@ class TestsAfterJob(AfterJob):
|
|||
'id': test.id,
|
||||
'name': str(test),
|
||||
'error': getattr(test, 'error', None),
|
||||
'formdata': json.dumps(formdata_json, cls=misc.JSONEncoder),
|
||||
'details': {
|
||||
'recorded_errors': test.recorded_errors,
|
||||
'missing_required_fields': test.missing_required_fields,
|
||||
|
|
|
@ -20,6 +20,7 @@
|
|||
<thead>
|
||||
<th>{% trans "Name" %}</th>
|
||||
<th>{% trans "Result" %}</th>
|
||||
<th>{% trans "Inspect" %}</th>
|
||||
<th>{% trans "Details" %}</th>
|
||||
</thead>
|
||||
<tbody>
|
||||
|
@ -27,6 +28,11 @@
|
|||
<tr>
|
||||
<td><a {% if test.url %}href="{{ test.url }}"{% else %}disabled{% endif %}>{{ test.name }}</a></td>
|
||||
<td>{% firstof test.error _("Success!") %}</td>
|
||||
<td>
|
||||
{% if test.formdata %}
|
||||
<a href="{{ forloop.counter0 }}/inspect">{% trans "Display inspect" %}</a>
|
||||
{% endif %}
|
||||
</td>
|
||||
<td>
|
||||
{% if test.has_details %}
|
||||
<a rel="popup" data-selector="div.section" href="{{ forloop.counter0 }}/">{% trans "Display details" %}</a>
|
||||
|
|
|
@ -217,6 +217,8 @@ class TestDef(sql.TestDef):
|
|||
formdata = objectdef.data_class()()
|
||||
formdata.just_created()
|
||||
|
||||
formdata.workflow_traces = []
|
||||
|
||||
if self.data['user']:
|
||||
formdata.set_user_from_json(self.data['user'])
|
||||
|
||||
|
@ -287,7 +289,7 @@ class TestDef(sql.TestDef):
|
|||
self.workflow_tests.run(formdata, agent_user)
|
||||
|
||||
def run_form_fill(self, objectdef):
|
||||
formdata = self.build_formdata(objectdef)
|
||||
self.formdata = formdata = self.build_formdata(objectdef)
|
||||
|
||||
get_publisher().reset_formdata_state()
|
||||
get_publisher().substitutions.feed(objectdef)
|
||||
|
|
|
@ -59,10 +59,7 @@ class WorkflowTests(XmlStorableObject):
|
|||
self.actions = []
|
||||
|
||||
def run(self, formdata, agent_user):
|
||||
# mock methods so nothing is stored
|
||||
formdata.record_workflow_event = lambda *args, **kwargs: None
|
||||
formdata.record_workflow_action = lambda *args, **kwargs: None
|
||||
formdata.store = lambda *args, **kwargs: None
|
||||
self.mock_formdata_methods(formdata)
|
||||
|
||||
# mark formdata as running workflow tests
|
||||
formdata.workflow_test = True
|
||||
|
@ -81,6 +78,20 @@ class WorkflowTests(XmlStorableObject):
|
|||
e.details.append(_('Form status when error occured: %s') % status.name)
|
||||
raise e
|
||||
|
||||
def mock_formdata_methods(self, formdata):
|
||||
from wcs.workflow_traces import WorkflowTrace
|
||||
|
||||
def record_workflow_event(event, **kwargs):
|
||||
formdata.workflow_traces.append(WorkflowTrace(formdata=formdata, event=event, event_args=kwargs))
|
||||
|
||||
def record_workflow_action(action):
|
||||
formdata.workflow_traces.append(WorkflowTrace(formdata=formdata, action=action))
|
||||
|
||||
formdata.record_workflow_event = record_workflow_event
|
||||
formdata.record_workflow_action = record_workflow_action
|
||||
|
||||
formdata.store = lambda *args, **kwargs: None
|
||||
|
||||
def get_new_action_id(self):
|
||||
if not self.actions:
|
||||
return '1'
|
||||
|
|
|
@ -17,7 +17,7 @@
|
|||
from quixote.html import TemplateIO, htmltext
|
||||
|
||||
from wcs import sql
|
||||
from wcs.qommon import _
|
||||
from wcs.qommon import _, misc
|
||||
|
||||
|
||||
class WorkflowTrace(sql.WorkflowTrace):
|
||||
|
@ -284,3 +284,19 @@ class WorkflowTrace(sql.WorkflowTrace):
|
|||
status_admin_base_url,
|
||||
status_label,
|
||||
)
|
||||
|
||||
def get_json_export_dict(self):
|
||||
return {field: getattr(self, field) for field, _ in self._table_static_fields}
|
||||
|
||||
@classmethod
|
||||
def import_from_json_dict(cls, data):
|
||||
workflow_trace = cls.__new__(cls)
|
||||
|
||||
for field, kind in cls._table_static_fields:
|
||||
value = data.get(field)
|
||||
if value and kind == 'timestamptz':
|
||||
value = misc.get_as_datetime(value)
|
||||
|
||||
setattr(workflow_trace, field, value)
|
||||
|
||||
return workflow_trace
|
||||
|
|
Loading…
Reference in New Issue