cr-connect-workflow/tests/workflow/test_workflow_service.py

96 lines
3.4 KiB
Python
Raw Normal View History

import json
import unittest
from tests.base_test import BaseTest
from crc.services.workflow_processor import WorkflowProcessor
from crc.services.workflow_service import WorkflowService
from SpiffWorkflow import Task as SpiffTask, WorkflowException
from example_data import ExampleDataLoader
from crc import db
from crc.models.task_event import TaskEventModel
from crc.models.api_models import Task
from crc.api.common import ApiError
class TestWorkflowService(BaseTest):
def test_documentation_processing_handles_replacements(self):
self.load_example_data()
workflow = self.create_workflow('random_fact')
processor = WorkflowProcessor(workflow)
processor.do_engine_steps()
task = processor.next_task()
task.task_spec.documentation = "Some simple docs"
docs = WorkflowService._process_documentation(task)
self.assertEqual("Some simple docs", docs)
task.data = {"replace_me": "new_thing"}
task.task_spec.documentation = "{{replace_me}}"
docs = WorkflowService._process_documentation(task)
self.assertEqual("new_thing", docs)
documentation = """
# Bigger Test
* bullet one
* bullet two has {{replace_me}}
2020-05-24 05:53:48 +00:00
# other stuff.
"""
expected = """
# Bigger Test
* bullet one
* bullet two has new_thing
2020-05-24 05:53:48 +00:00
# other stuff.
"""
task.task_spec.documentation = documentation
result = WorkflowService._process_documentation(task)
self.assertEqual(expected, result)
def test_documentation_processing_handles_conditionals(self):
self.load_example_data()
workflow = self.create_workflow('random_fact')
processor = WorkflowProcessor(workflow)
processor.do_engine_steps()
task = processor.next_task()
task.task_spec.documentation = "This test {% if works == 'yes' %}works{% endif %}"
docs = WorkflowService._process_documentation(task)
self.assertEqual("This test ", docs)
task.data = {"works": 'yes'}
docs = WorkflowService._process_documentation(task)
self.assertEqual("This test works", docs)
def test_enum_options_from_file(self):
self.load_example_data()
workflow = self.create_workflow('enum_options_from_file')
processor = WorkflowProcessor(workflow)
processor.do_engine_steps()
task = processor.next_task()
WorkflowService.process_options(task, task.task_spec.form.fields[0])
options = task.task_spec.form.fields[0].options
2020-06-05 18:08:46 +00:00
self.assertEqual(28, len(options))
self.assertEqual('1000', options[0]['id'])
self.assertEqual("UVA - INTERNAL - GM USE ONLY", options[0]['name'])
def test_random_data_populate_form_on_auto_complete(self):
self.load_example_data()
workflow = self.create_workflow('enum_options_with_search')
processor = WorkflowProcessor(workflow)
processor.do_engine_steps()
task = processor.next_task()
task_api = WorkflowService.spiff_task_to_api_task(task, add_docs_and_forms=True)
WorkflowService.populate_form_with_random_data(task, task_api, required_only=False)
self.assertTrue(isinstance(task.data["sponsor"], dict))
def test_dmn_evaluation_errors_in_oncomplete_raise_api_errors_during_validation(self):
workflow_spec_model = self.load_test_spec("decision_table_invalid")
with self.assertRaises(ApiError):
WorkflowService.test_spec(workflow_spec_model.id)