test_core.py 文件源码

def test_returns_no_tracking_when_no_execution_id_is_found(self):
        self.engine.cwlogs.get_log_events.side_effect = CloudWatchStreamDoesNotExist()
        workflow_state = {e: core.STATE_UNKNOWN for e in self.workflow_events}
        execution_id = "transaction-id-123"

        expected = {
            "events_defined": workflow_state,
            "events_received": [],
            "tracking_summary": {
                "last_received_event": None,
                "subscribers": [],
                "execution_path": self.engine \
                                      ._generate_execution_path(workflow_state)
            }
        }
        actual = self.engine.track(self.workflow_id, self.execution_id)
        nt.assert_equals(expected, actual)