test_cw_client.py 文件源码

def test_put_log_messages_no_token(self):
        ''' test put_log_messages() when no sequence token given '''

        messages = [dict(a='abc'), dict(b='xyz', __CURSOR=sentinel.cursor)]
        events = [sentinel.msg1, sentinel.msg2]
        with patch.object(self.client, 'make_message', side_effect=events, autospec=True) as make_message:
            with patch.object(self.client, 'save_cursor', autospec=True) as save_cursor:
                self.client.put_log_messages(sentinel.group, sentinel.stream, None, messages)

        self.cwl.put_log_events.assert_called_once_with(
            logGroupName=sentinel.group,
            logStreamName=sentinel.stream,
            logEvents=events,
        )
        # saves the cursor once finished
        save_cursor.assert_called_once_with(sentinel.cursor)