New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add kinesisvideo #3271
Add kinesisvideo #3271
Conversation
d75b0d9
to
97b1b27
Compare
@bblommers please check when you have time 🙏 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hi @toshitanian, thanks for your PR! Left a few comments
moto/kinesisvideo/models.py
Outdated
self.status = "ACTIVE" | ||
self.version = self._get_random_string() | ||
self.creation_time = datetime.utcnow() | ||
stream_arn = "arn:aws:kinesisvideo:{}:123456789012:stream/{}/1598784211076".format( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm assuming this is the account ID? Let's import that from moto.core
, to avoid magic numbers as much as possible
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
moto/kinesisvideo/models.py
Outdated
): | ||
streams = [_ for _ in self.streams.values() if _.stream_name == stream_name] | ||
if len(streams) > 0: | ||
raise ResourceInUseException("The stream a already exists.") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Typo - or should it say The stream {stream_name} already exists
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for finding out. updated.
6384f37
|
||
|
||
@mock_kinesisvideo | ||
def test_list(): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we split this test up? I prefer to have smaller test cases, i.e. test_list
, test_create
, test_create_with_same_name
, etc.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done: 937dd23
def test_kinesisvideo_list(): | ||
backend = server.create_backend_app("kinesisvideo") | ||
test_client = backend.test_client() | ||
# do test |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this test necessary, if it's already covered by the boto3 test?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's better to have test for moto_server. I added assert of response code to make sure that the server is up.
a7b46c6
@bblommers changed for reviews. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM. Thanks @toshitanian!
This is now part of moto >= 1.3.15.dev1042 |
Added basic operations of
kinesisvideo
create_stream
describe_stream
list_streams
delete_stream
get_data_endpoint