Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #2493 from edekadigital/filter_logs_raises_if_filter_pattern

Browse Source 
add NotImplementedError when filter_pattern is given to filter_log_events
This commit is contained in:
Jack Danger 2019-10-23 15:23:45 -07:00 committed by GitHub
commit 902a35d5b9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 36 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
moto/logs/models.py
View File

@ -125,6 +125,9 @@ class LogStream:
return events_page, get_paging_token_from_index(back_index, True), get_paging_token_from_index(next_index) return events_page, get_paging_token_from_index(back_index, True), get_paging_token_from_index(next_index)
def filter_log_events(self, log_group_name, log_stream_names, start_time, end_time, limit, next_token, filter_pattern, interleaved): def filter_log_events(self, log_group_name, log_stream_names, start_time, end_time, limit, next_token, filter_pattern, interleaved):
if filter_pattern:
raise NotImplementedError('filter_pattern is not yet implemented')
def filter_func(event): def filter_func(event):
if start_time and event.timestamp < start_time: if start_time and event.timestamp < start_time:
return False return False

33
tests/test_logs/test_logs.py
View File

@ -1,10 +1,12 @@
import boto3 import boto3
import os
import sure # noqa import sure # noqa
import six import six
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_logs, settings from moto import mock_logs, settings
from nose.tools import assert_raises from nose.tools import assert_raises
from nose import SkipTest
_logs_region = 'us-east-1' if settings.TEST_SERVER_MODE else 'us-west-2' _logs_region = 'us-east-1' if settings.TEST_SERVER_MODE else 'us-west-2'
@ -128,6 +130,36 @@ def test_filter_logs_interleaved():
resulting_event['timestamp'].should.equal(original_message['timestamp']) resulting_event['timestamp'].should.equal(original_message['timestamp'])
resulting_event['message'].should.equal(original_message['message']) resulting_event['message'].should.equal(original_message['message'])
@mock_logs
def test_filter_logs_raises_if_filter_pattern():
if os.environ.get('TEST_SERVER_MODE', 'false').lower() == 'true':
raise SkipTest('Does not work in server mode due to error in Workzeug')
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'dummy'
log_stream_name = 'stream'
conn.create_log_group(logGroupName=log_group_name)
conn.create_log_stream(
logGroupName=log_group_name,
logStreamName=log_stream_name
)
messages = [
{'timestamp': 0, 'message': 'hello'},
{'timestamp': 0, 'message': 'world'}
]
conn.put_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
logEvents=messages
)
with assert_raises(NotImplementedError):
conn.filter_log_events(
logGroupName=log_group_name,
logStreamNames=[log_stream_name],
filterPattern='{$.message = "hello"}',
)
@mock_logs @mock_logs
def test_put_retention_policy(): def test_put_retention_policy():
conn = boto3.client('logs', 'us-west-2') conn = boto3.client('logs', 'us-west-2')
@ -142,6 +174,7 @@ def test_put_retention_policy():
response = conn.delete_log_group(logGroupName=log_group_name) response = conn.delete_log_group(logGroupName=log_group_name)
@mock_logs @mock_logs
def test_delete_retention_policy(): def test_delete_retention_policy():
conn = boto3.client('logs', 'us-west-2') conn = boto3.client('logs', 'us-west-2')