test_aws.py 文件源码

def test_successfully_gets_log_events(self):
        mocked_timestamp = 1476826208 * 1000
        mocked_message = '{"foo": "bar"}'
        mocked_events = {
            "events": [
                {
                    "timestamp": mocked_timestamp,
                    "message": mocked_message
                }
            ],
            "nextForwardToken": None
        }
        self.logs.cwlogs.get_log_events.return_value = mocked_events
        expected = [{
            "timestamp": utils.format_datetime(datetime.fromtimestamp(mocked_timestamp / 1000)),
            "data": json.loads(mocked_message)
        }]
        actual = self.logs.get_log_events(self.log_group, self.log_stream)
        nt.assert_equals(expected, actual)