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

add implemented kinesis method describe_stream_summary (#2023)

Browse source
This commit is contained in:
Alexey Firsov 2019-05-21 02:02:36 +03:00 • committed by Terry Cain
commit f13e4e41cd
4 changed files with 59 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

18
moto/kinesis/models.py
View file

@ -116,10 +116,12 @@ class Stream(BaseModel):
def __init__(self, stream_name, shard_count, region):
self.stream_name = stream_name
self.shard_count = shard_count
self.creation_datetime = datetime.datetime.now()
self.region = region
self.account_number = "123456789012"
self.shards = {}
self.tags = {}
self.status = "ACTIVE"
if six.PY3:
izip_longest = itertools.zip_longest
@ -183,12 +185,23 @@ class Stream(BaseModel):
"StreamDescription": {
"StreamARN": self.arn,
"StreamName": self.stream_name,
"StreamStatus": "ACTIVE",
"StreamStatus": self.status,
"HasMoreShards": False,
"Shards": [shard.to_json() for shard in self.shards.values()],
}
}
def to_json_summary(self):
return {
"StreamDescriptionSummary": {
"StreamARN": self.arn,
"StreamName": self.stream_name,
"StreamStatus": self.status,
"StreamCreationTimestamp": six.text_type(self.creation_datetime),
"OpenShardCount": self.shard_count,
}
}
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
properties = cloudformation_json['Properties']
@ -309,6 +322,9 @@ class KinesisBackend(BaseBackend):
else:
raise StreamNotFoundError(stream_name)
def describe_stream_summary(self, stream_name):
return self.describe_stream(stream_name)
def list_streams(self):
return self.streams.values()