Tests for uid in logs. Currently we test for uid in a response. This covers ApiError.

Currently, we don't have a test for Sentry. Unsure how to do this.
Also added a script, service and test workflow to help. (Also to learn about adding a script and service.)
This commit is contained in:
mike cullerton 2020-12-11 17:47:53 -05:00
parent 856fe445b0
commit 7defc2b02f
4 changed files with 99 additions and 14 deletions

View File

@ -0,0 +1,15 @@
from crc.scripts.script import Script
from crc.services.failing_service import FailingService
class FailingScript(Script):
def get_description(self):
return """It fails"""
def do_task_validate_only(self, task, *args, **kwargs):
pass
def do_task(self, task, *args, **kwargs):
FailingService.fail_as_service()

View File

@ -0,0 +1,11 @@
from crc.api.common import ApiError
class FailingService(object):
@staticmethod
def fail_as_service():
"""It fails"""
raise ApiError(code='bad_service',
message='This is my failing service')

View File

@ -0,0 +1,52 @@
<?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:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_886a64d" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="4.2.0">
<bpmn:process id="Process_FailingWorkflow" name="Failing Workflow" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_0cszvz2</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:sequenceFlow id="Flow_0cszvz2" sourceRef="StartEvent_1" targetRef="Activity_CallFailingScript" />
<bpmn:scriptTask id="Activity_CallFailingScript" name="Call Failing Script">
<bpmn:incoming>Flow_0cszvz2</bpmn:incoming>
<bpmn:outgoing>Flow_1l02umo</bpmn:outgoing>
<bpmn:script>failing_script()</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_1l02umo" sourceRef="Activity_CallFailingScript" targetRef="Activity_PlaceHolder" />
<bpmn:scriptTask id="Activity_PlaceHolder" name="Place Holder">
<bpmn:incoming>Flow_1l02umo</bpmn:incoming>
<bpmn:outgoing>Flow_08zq7mf</bpmn:outgoing>
<bpmn:script>print('I am a placeholder.')</bpmn:script>
</bpmn:scriptTask>
<bpmn:endEvent id="Event_0k0yvmv">
<bpmn:incoming>Flow_08zq7mf</bpmn:incoming>
</bpmn:endEvent>
<bpmn:sequenceFlow id="Flow_08zq7mf" sourceRef="Activity_PlaceHolder" targetRef="Event_0k0yvmv" />
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_FailingWorkflow">
<bpmndi:BPMNEdge id="Flow_0cszvz2_di" bpmnElement="Flow_0cszvz2">
<di:waypoint x="215" y="117" />
<di:waypoint x="270" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_1l02umo_di" bpmnElement="Flow_1l02umo">
<di:waypoint x="370" y="117" />
<di:waypoint x="430" y="117" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_08zq7mf_di" bpmnElement="Flow_08zq7mf">
<di:waypoint x="530" y="117" />
<di:waypoint x="592" 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="Activity_045fev7_di" bpmnElement="Activity_CallFailingScript">
<dc:Bounds x="270" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_0f3nusg_di" bpmnElement="Activity_PlaceHolder">
<dc:Bounds x="430" y="77" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_0k0yvmv_di" bpmnElement="Event_0k0yvmv">
<dc:Bounds x="592" y="99" width="36" height="36" />
</bpmndi:BPMNShape>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

View File

@ -1,21 +1,28 @@
from crc.api.common import ApiError
from tests.base_test import BaseTest
from crc import db
from crc.models.user import UserModel
import json
class TestUserID(BaseTest):
def test_user_id(self):
# try:
# False
# except:
# raise ApiError
def test_user_id_in_request(self):
"""This assures the uid is in response via ApiError"""
# ApiError()
# self.assertEqual(True,False)
workflow = self.create_workflow('failing_workflow')
user_uid = workflow.study.user_uid
user = db.session.query(UserModel).filter_by(uid=user_uid).first()
rv = self.app.get(f'/v1.0/workflow/{workflow.id}'
f'?soft_reset={str(False)}'
f'&hard_reset={str(False)}'
f'&do_engine_steps={str(True)}',
headers=self.logged_in_headers(user),
content_type="application/json")
data = json.loads(rv.data)
self.assertEqual(data['task_user'], user_uid)
# with self.assertRaises(ApiError) as api_error:
# self.assertEqual(2, 3)
workflow = self.create_workflow('email')
first_task = self.get_workflow_api(workflow).next_task
raise ApiError('unknown_approval', 'Please provide a valid Approval ID.')
def test_user_id_in_sentry(self):
"""This assures the uid is in Sentry.
We use this to send errors to Slack."""
# Currently have no clue how to do this :(
pass