Implement describe_log_groups() method for CloudWatchLogs

This patch teaches `LogsResponse` class how to handle the DescribeLogGroups
request, so that we can mock out the `boto.describe_log_groups()` call.

With this change in place, we can write as below:

    @mock_logs
    def test_log_group():
        conn = boto3.client('logs', 'us-west-2')

        some_method_to_init_log_groups()

        resp = conn.describe_log_groups(logGroupNamePrefix='myapp')
        assert ...

This should be fairly useful for a number of programs which handles
CloudWatchLogs.

Signed-off-by: Fujimoto Seiji <fujimoto@clear-code.com>
This commit is contained in:
Fujimoto Seiji 2018-04-24 11:12:17 +09:00
parent 21a264c337
commit ac016a7bb3
3 changed files with 44 additions and 0 deletions

View File

@ -132,6 +132,9 @@ class LogGroup:
def __init__(self, region, name, tags):
self.name = name
self.region = region
self.arn = "arn:aws:logs:{region}:1:log-group:{log_group}".format(
region=region, log_group=name)
self.creationTime = unix_time_millis()
self.tags = tags
self.streams = dict() # {name: LogStream}
@ -197,6 +200,16 @@ class LogGroup:
searched_streams = [{"logStreamName": stream.logStreamName, "searchedCompletely": True} for stream in streams]
return events_page, next_token, searched_streams
def to_describe_dict(self):
return {
"arn": self.arn,
"creationTime": self.creationTime,
"logGroupName": self.name,
"metricFilterCount": 0,
"retentionInDays": 30,
"storedBytes": sum(s.storedBytes for s in self.streams.values()),
}
class LogsBackend(BaseBackend):
def __init__(self, region_name):
@ -223,6 +236,21 @@ class LogsBackend(BaseBackend):
raise ResourceNotFoundException()
del self.groups[log_group_name]
def describe_log_groups(self, limit, log_group_name_prefix, next_token):
if log_group_name_prefix is None:
log_group_name_prefix = ''
if next_token is None:
next_token = 0
groups = sorted(group.to_describe_dict() for name, group in self.groups.items() if name.startswith(log_group_name_prefix))
groups_page = groups[next_token:next_token + limit]
next_token += limit
if next_token >= len(groups):
next_token = None
return groups_page, next_token
def create_log_stream(self, log_group_name, log_stream_name):
if log_group_name not in self.groups:
raise ResourceNotFoundException()

View File

@ -33,6 +33,18 @@ class LogsResponse(BaseResponse):
self.logs_backend.delete_log_group(log_group_name)
return ''
def describe_log_groups(self):
log_group_name_prefix = self._get_param('logGroupNamePrefix')
next_token = self._get_param('nextToken')
limit = self._get_param('limit', 50)
assert limit <= 50
groups, next_token = self.logs_backend.describe_log_groups(
limit, log_group_name_prefix, next_token)
return json.dumps({
"logGroups": groups,
"nextToken": next_token
})
def create_log_stream(self):
log_group_name = self._get_param('logGroupName')
log_stream_name = self._get_param('logStreamName')

View File

@ -13,6 +13,10 @@ def test_log_group_create():
conn = boto3.client('logs', 'us-west-2')
log_group_name = 'dummy'
response = conn.create_log_group(logGroupName=log_group_name)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response['logGroups']) == 1
response = conn.delete_log_group(logGroupName=log_group_name)