catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Feature cloudwatch log retention (#2199)

Browse source 
* add proper retentionInDays to describe_log_groups response and add support for delete_retention_policy() and put_retention_policy() to log groups

* fix for inline comment formatting

* include check for retentionInDays to verify no retention by default in  test_log_group_create
This commit is contained in:
Jon Beilke 2019-05-25 04:14:23 -05:00 • committed by Terry Cain
commit cb72d1d00e
3 changed files with 68 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

36
tests/test_logs/test_logs.py
View file

@ -17,6 +17,8 @@ def test_log_group_create():
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
# AWS defaults to Never Expire for log group retention
assert response['logGroups'][0].get('retentionInDays') == None
response = conn.delete_log_group(logGroupName=log_group_name)
@ -126,3 +128,37 @@ def test_filter_logs_interleaved():
resulting_event['timestamp'].should.equal(original_message['timestamp'])
resulting_event['message'].should.equal(original_message['message'])
@mock_logs
def test_put_retention_policy():
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'dummy'
response = conn.create_log_group(logGroupName=log_group_name)
response = conn.put_retention_policy(logGroupName=log_group_name, retentionInDays=7)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
assert response['logGroups'][0].get('retentionInDays') == 7
response = conn.delete_log_group(logGroupName=log_group_name)
@mock_logs
def test_delete_retention_policy():
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'dummy'
response = conn.create_log_group(logGroupName=log_group_name)
response = conn.put_retention_policy(logGroupName=log_group_name, retentionInDays=7)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
assert response['logGroups'][0].get('retentionInDays') == 7
response = conn.delete_retention_policy(logGroupName=log_group_name)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
assert response['logGroups'][0].get('retentionInDays') == None
response = conn.delete_log_group(logGroupName=log_group_name)