mirror of
https://github.com/sartography/cr-connect-workflow.git
synced 2025-02-23 13:18:35 +00:00
Dan Funk
GitHub
commit
c3f9ab28d0
66
crc/scripts/modify_spreadsheet.py
Normal file
66
crc/scripts/modify_spreadsheet.py
Normal file
@ -0,0 +1,66 @@
|
||||
from crc import session
|
||||
from crc.api.common import ApiError
|
||||
from crc.models.file import FileModel, FileDataModel
|
||||
from crc.scripts.script import Script
|
||||
|
||||
from io import BytesIO
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.writer.excel import save_virtual_workbook
|
||||
|
||||
|
||||
class ModifySpreadsheet(Script):
|
||||
|
||||
@staticmethod
|
||||
def get_parameters(args, kwargs):
|
||||
parameters = {}
|
||||
if len(args) == 3 or ('irb_doc_code' in kwargs and 'cell' in kwargs and 'text' in kwargs):
|
||||
if 'irb_doc_code' in kwargs and 'line' in kwargs and 'text' in kwargs:
|
||||
parameters['irb_doc_code'] = (kwargs['irb_doc_code'])
|
||||
parameters['cell'] = (kwargs['cell'])
|
||||
parameters['text'] = (kwargs['text'])
|
||||
else:
|
||||
parameters['irb_doc_code'] = (args[0])
|
||||
parameters['cell'] = (args[1])
|
||||
parameters['text'] = (args[2])
|
||||
return parameters
|
||||
|
||||
def get_description(self):
|
||||
return """Script to modify an existing spreadsheet.
|
||||
It inserts text into a spreadsheet in the cell indicated.
|
||||
Requires 'irb_doc_code', 'cell', and 'text' parameters.
|
||||
Example: modify_spreadsheet('Finance_BCA', 'C4', 'This is my inserted text')
|
||||
Example: modify_spreadsheet(irb_doc_code='Finance_BCA', cell='C4', text='This is my inserted text')
|
||||
"""
|
||||
|
||||
def do_task_validate_only(self, task, study_id, workflow_id, *args, **kwargs):
|
||||
parameters = self.get_parameters(args, kwargs)
|
||||
if len(parameters) == 3:
|
||||
return True
|
||||
else:
|
||||
raise ApiError(code='missing_parameters',
|
||||
message='The modify_spreadsheet script requires 3 parameters: irb_doc_code, cell, and text')
|
||||
|
||||
def do_task(self, task, study_id, workflow_id, *args, **kwargs):
|
||||
parameters = self.get_parameters(args, kwargs)
|
||||
if len(parameters) == 3:
|
||||
|
||||
spreadsheet = session.query(FileModel). \
|
||||
filter(FileModel.workflow_id == workflow_id). \
|
||||
filter(FileModel.irb_doc_code == parameters['irb_doc_code']).\
|
||||
first()
|
||||
if spreadsheet:
|
||||
spreadsheet_data = session.query(FileDataModel).\
|
||||
filter(FileDataModel.file_model_id==spreadsheet.id).\
|
||||
first()
|
||||
workbook = load_workbook(BytesIO(spreadsheet_data.data))
|
||||
sheet = workbook.active
|
||||
sheet[parameters['cell']] = parameters['text']
|
||||
data_string = save_virtual_workbook(workbook)
|
||||
spreadsheet_data.data = data_string
|
||||
session.commit()
|
||||
else:
|
||||
raise ApiError(code='missing_spreadsheet',
|
||||
message=f"The spreadshhet you want to modify does not exist. Workflow ID is {workflow_id}, and IRB Doc Code is {parameters['irb_doc_code']}")
|
||||
else:
|
||||
raise ApiError(code='missing_parameters',
|
||||
message='The modify_spreadsheet script requires 3 parameters: irb_doc_code, cell, and text')
|
||||
88
tests/data/modify_spreadsheet/modify_spreadsheet.bpmn
Normal file
88
tests/data/modify_spreadsheet/modify_spreadsheet.bpmn
Normal file
@ -0,0 +1,88 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:camunda="http://camunda.org/schema/1.0/bpmn" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_1wrf54p" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="4.2.0">
|
||||
<bpmn:process id="Process_ModifySpreadsheet" name="Modify Spreadsheet" isExecutable="true">
|
||||
<bpmn:startEvent id="StartEvent_1">
|
||||
<bpmn:outgoing>Flow_03wyga3</bpmn:outgoing>
|
||||
</bpmn:startEvent>
|
||||
<bpmn:sequenceFlow id="Flow_03wyga3" sourceRef="StartEvent_1" targetRef="Activity_FileUpload" />
|
||||
<bpmn:sequenceFlow id="Flow_0msewj5" sourceRef="Activity_FileUpload" targetRef="Activity_GetModifyData" />
|
||||
<bpmn:sequenceFlow id="Flow_0fpc32g" sourceRef="Activity_GetModifyData" targetRef="Activity_ModifySpreadsheet" />
|
||||
<bpmn:sequenceFlow id="Flow_10ulj3l" sourceRef="Activity_ModifySpreadsheet" targetRef="Event_1d07dls" />
|
||||
<bpmn:endEvent id="Event_1d07dls">
|
||||
<bpmn:incoming>Flow_10ulj3l</bpmn:incoming>
|
||||
</bpmn:endEvent>
|
||||
<bpmn:userTask id="Activity_FileUpload" name="Upload Spreadsheet" camunda:formKey="FileUpload">
|
||||
<bpmn:extensionElements>
|
||||
<camunda:formData>
|
||||
<camunda:formField id="Finance_BCA" label="'File Upload'" type="file">
|
||||
<camunda:validation>
|
||||
<camunda:constraint name="required" config="True" />
|
||||
</camunda:validation>
|
||||
</camunda:formField>
|
||||
</camunda:formData>
|
||||
</bpmn:extensionElements>
|
||||
<bpmn:incoming>Flow_03wyga3</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_0msewj5</bpmn:outgoing>
|
||||
</bpmn:userTask>
|
||||
<bpmn:scriptTask id="Activity_ModifySpreadsheet" name="Modify Spreadsheet">
|
||||
<bpmn:incoming>Flow_0fpc32g</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_10ulj3l</bpmn:outgoing>
|
||||
<bpmn:script>modify_spreadsheet('Finance_BCA', cell_indicator, input_text)</bpmn:script>
|
||||
</bpmn:scriptTask>
|
||||
<bpmn:userTask id="Activity_GetModifyData" name="Get Modify Data" camunda:formKey="ModifyData">
|
||||
<bpmn:documentation>## File Upload
|
||||
{{ Finance_BCA }}</bpmn:documentation>
|
||||
<bpmn:extensionElements>
|
||||
<camunda:formData>
|
||||
<camunda:formField id="cell_indicator" label="Cell Indicator" type="string">
|
||||
<camunda:validation>
|
||||
<camunda:constraint name="required" config="True" />
|
||||
</camunda:validation>
|
||||
</camunda:formField>
|
||||
<camunda:formField id="input_text" label="Input Text" type="string">
|
||||
<camunda:validation>
|
||||
<camunda:constraint name="required" config="True" />
|
||||
</camunda:validation>
|
||||
</camunda:formField>
|
||||
</camunda:formData>
|
||||
</bpmn:extensionElements>
|
||||
<bpmn:incoming>Flow_0msewj5</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_0fpc32g</bpmn:outgoing>
|
||||
</bpmn:userTask>
|
||||
</bpmn:process>
|
||||
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
|
||||
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_ModifySpreadsheet">
|
||||
<bpmndi:BPMNEdge id="Flow_10ulj3l_di" bpmnElement="Flow_10ulj3l">
|
||||
<di:waypoint x="691" y="117" />
|
||||
<di:waypoint x="912" y="117" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_0fpc32g_di" bpmnElement="Flow_0fpc32g">
|
||||
<di:waypoint x="530" y="117" />
|
||||
<di:waypoint x="591" y="117" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_0msewj5_di" bpmnElement="Flow_0msewj5">
|
||||
<di:waypoint x="370" y="117" />
|
||||
<di:waypoint x="430" y="117" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_03wyga3_di" bpmnElement="Flow_03wyga3">
|
||||
<di:waypoint x="215" y="117" />
|
||||
<di:waypoint x="270" y="117" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
|
||||
<dc:Bounds x="179" y="99" width="36" height="36" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Event_1d07dls_di" bpmnElement="Event_1d07dls">
|
||||
<dc:Bounds x="912" y="99" width="36" height="36" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_0jut75d_di" bpmnElement="Activity_FileUpload">
|
||||
<dc:Bounds x="270" y="77" width="100" height="80" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_0uk32ds_di" bpmnElement="Activity_ModifySpreadsheet">
|
||||
<dc:Bounds x="591" y="77" width="100" height="80" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_0hn4ju2_di" bpmnElement="Activity_GetModifyData">
|
||||
<dc:Bounds x="430" y="77" width="100" height="80" />
|
||||
</bpmndi:BPMNShape>
|
||||
</bpmndi:BPMNPlane>
|
||||
</bpmndi:BPMNDiagram>
|
||||
</bpmn:definitions>
|
||||
BIN
tests/data/modify_spreadsheet/test_spreadsheet.xlsx
Normal file
BIN
tests/data/modify_spreadsheet/test_spreadsheet.xlsx
Normal file
Binary file not shown.
78
tests/scripts/test_modify_spreadsheet.py
Normal file
78
tests/scripts/test_modify_spreadsheet.py
Normal file
@ -0,0 +1,78 @@
|
||||
from tests.base_test import BaseTest
|
||||
|
||||
from crc import app, session
|
||||
from crc.models.file import FileModel, FileDataModel
|
||||
from crc.services.user_file_service import UserFileService
|
||||
|
||||
from io import BytesIO
|
||||
from openpyxl import load_workbook
|
||||
|
||||
import os
|
||||
|
||||
|
||||
class TestModifySpreadsheet(BaseTest):
|
||||
|
||||
def upload_spreadsheet(self, workflow, task, irb_doc_code):
|
||||
filepath = os.path.join(app.root_path, '..', 'tests', 'data',
|
||||
'modify_spreadsheet', 'test_spreadsheet.xlsx')
|
||||
with open(filepath, 'br') as f_open:
|
||||
ss_data = f_open.read()
|
||||
|
||||
file_model = UserFileService.add_workflow_file(workflow_id=workflow.id,
|
||||
task_spec_name=task.name,
|
||||
name="test_spreadsheet.xlsx", content_type="text",
|
||||
binary_data=ss_data, irb_doc_code=irb_doc_code)
|
||||
workflow_api = self.complete_form(workflow, task, {irb_doc_code: {'id': file_model.id}})
|
||||
return workflow_api
|
||||
|
||||
@staticmethod
|
||||
def get_sheet(workflow_id, irb_doc_code):
|
||||
spreadsheet = session.query(FileModel). \
|
||||
filter(FileModel.workflow_id == workflow_id). \
|
||||
filter(FileModel.irb_doc_code == irb_doc_code). \
|
||||
first()
|
||||
spreadsheet_data = session.query(FileDataModel). \
|
||||
filter(FileDataModel.file_model_id == spreadsheet.id). \
|
||||
first()
|
||||
workbook = load_workbook(BytesIO(spreadsheet_data.data))
|
||||
sheet = workbook.active
|
||||
return sheet
|
||||
|
||||
def test_modify_spreadsheet(self):
|
||||
irb_doc_code = 'Finance_BCA'
|
||||
cell_indicator = 'C4'
|
||||
input_text = 'This is my input text.'
|
||||
|
||||
workflow = self.create_workflow('modify_spreadsheet')
|
||||
workflow_api = self.get_workflow_api(workflow)
|
||||
task = workflow_api.next_task
|
||||
|
||||
workflow_api = self.upload_spreadsheet(workflow, task, irb_doc_code)
|
||||
|
||||
sheet = self.get_sheet(workflow.id, irb_doc_code)
|
||||
self.assertEqual(None, sheet[cell_indicator].value)
|
||||
|
||||
task = workflow_api.next_task
|
||||
self.complete_form(workflow, task, {'cell_indicator': cell_indicator,
|
||||
'input_text': input_text})
|
||||
|
||||
sheet = self.get_sheet(workflow.id, irb_doc_code)
|
||||
self.assertEqual(input_text, sheet[cell_indicator].value)
|
||||
|
||||
def test_missing_spreadsheet(self):
|
||||
"""The modify_spreadsheet has Finance_BCA hard-coded as the spreadsheet to modify.
|
||||
In this test we upload a spreadsheet with a different doc code,
|
||||
and assert that we raise an error when the Finance_BCA spreadsheet does not exist"""
|
||||
irb_doc_code = 'Finance_GPRF'
|
||||
cell_indicator = 'C4'
|
||||
input_text = 'This is my input text.'
|
||||
|
||||
workflow = self.create_workflow('modify_spreadsheet')
|
||||
workflow_api = self.get_workflow_api(workflow)
|
||||
task = workflow_api.next_task
|
||||
|
||||
workflow_api = self.upload_spreadsheet(workflow, task, irb_doc_code)
|
||||
task = workflow_api.next_task
|
||||
with self.assertRaises(AssertionError):
|
||||
self.complete_form(workflow, task, {'cell_indicator': cell_indicator,
|
||||
'input_text': input_text})
|
||||
Loading…
x
Reference in New Issue
Block a user