Techdebt: Replace sure with regular assertions in KinesisVideo (#6589)

This commit is contained in:
Bert Blommers 2023-08-02 09:39:45 +00:00 committed by GitHub
parent 28837234c5
commit f69d0b2b5f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 21 additions and 28 deletions

View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
from moto import mock_kinesisvideo
from botocore.exceptions import ClientError
@ -13,7 +12,7 @@ def test_create_stream():
# stream can be created
res = client.create_stream(StreamName=stream_name, DeviceName=device_name)
res.should.have.key("StreamARN").which.should.contain(stream_name)
assert stream_name in res["StreamARN"]
@mock_kinesisvideo
@ -36,7 +35,7 @@ def test_describe_stream():
device_name = "random-device"
res = client.create_stream(StreamName=stream_name, DeviceName=device_name)
res.should.have.key("StreamARN").which.should.contain(stream_name)
assert stream_name in res["StreamARN"]
stream_arn = res["StreamARN"]
# cannot create with existing stream name
@ -45,19 +44,17 @@ def test_describe_stream():
# stream can be described with name
res = client.describe_stream(StreamName=stream_name)
res.should.have.key("StreamInfo")
stream_info = res["StreamInfo"]
stream_info.should.have.key("StreamARN").which.should.contain(stream_name)
stream_info.should.have.key("StreamName").which.should.equal(stream_name)
stream_info.should.have.key("DeviceName").which.should.equal(device_name)
assert stream_name in stream_info["StreamARN"]
assert stream_info["StreamName"] == stream_name
assert stream_info["DeviceName"] == device_name
# stream can be described with arn
res = client.describe_stream(StreamARN=stream_arn)
res.should.have.key("StreamInfo")
stream_info = res["StreamInfo"]
stream_info.should.have.key("StreamARN").which.should.contain(stream_name)
stream_info.should.have.key("StreamName").which.should.equal(stream_name)
stream_info.should.have.key("DeviceName").which.should.equal(device_name)
assert stream_name in stream_info["StreamARN"]
assert stream_info["StreamName"] == stream_name
assert stream_info["DeviceName"] == device_name
@mock_kinesisvideo
@ -82,9 +79,8 @@ def test_list_streams():
# streams can be listed
res = client.list_streams()
res.should.have.key("StreamInfoList")
streams = res["StreamInfoList"]
streams.should.have.length_of(2)
assert len(streams) == 2
@mock_kinesisvideo
@ -102,7 +98,7 @@ def test_delete_stream():
client.delete_stream(StreamARN=stream_2_arn)
res = client.list_streams()
streams = res["StreamInfoList"]
streams.should.have.length_of(1)
assert len(streams) == 1
@mock_kinesisvideo
@ -134,4 +130,4 @@ def test_data_endpoint():
api_name = "GET_MEDIA"
client.create_stream(StreamName=stream_name, DeviceName=device_name)
res = client.get_data_endpoint(StreamName=stream_name, APIName=api_name)
res.should.have.key("DataEndpoint")
assert "DataEndpoint" in res

View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
import moto.server as server
from moto import mock_kinesisvideo
@ -13,4 +11,4 @@ def test_kinesisvideo_server_is_up():
backend = server.create_backend_app("kinesisvideo")
test_client = backend.test_client()
res = test_client.post("/listStreams")
res.status_code.should.equal(200)
assert res.status_code == 200

View File

@ -1,5 +1,4 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_kinesisvideoarchivedmedia
from moto import mock_kinesisvideo
from datetime import datetime, timedelta
@ -23,8 +22,9 @@ def test_get_hls_streaming_session_url():
endpoint_url=data_endpoint,
)
res = client.get_hls_streaming_session_url(StreamName=stream_name)
reg_exp = rf"^{data_endpoint}/hls/v1/getHLSMasterPlaylist.m3u8\?SessionToken\=.+$"
res.should.have.key("HLSStreamingSessionURL").which.should.match(reg_exp)
assert res["HLSStreamingSessionURL"].startswith(
f"{data_endpoint}/hls/v1/getHLSMasterPlaylist.m3u8?SessionToken="
)
@mock_kinesisvideo
@ -45,8 +45,9 @@ def test_get_dash_streaming_session_url():
endpoint_url=data_endpoint,
)
res = client.get_dash_streaming_session_url(StreamName=stream_name)
reg_exp = rf"^{data_endpoint}/dash/v1/getDASHManifest.mpd\?SessionToken\=.+$"
res.should.have.key("DASHStreamingSessionURL").which.should.match(reg_exp)
assert res["DASHStreamingSessionURL"].startswith(
f"{data_endpoint}/dash/v1/getDASHManifest.mpd?SessionToken="
)
@mock_kinesisvideo
@ -78,5 +79,5 @@ def test_get_clip():
},
},
)
res.should.have.key("ContentType").which.should.match("video/mp4")
res.should.have.key("Payload")
assert res["ContentType"] == "video/mp4"
assert "Payload" in res

View File

@ -1,5 +1,3 @@
import sure # noqa # pylint: disable=unused-import
import moto.server as server
from moto import mock_kinesisvideoarchivedmedia
@ -14,4 +12,4 @@ def test_kinesisvideoarchivedmedia_server_is_up():
test_client = backend.test_client()
res = test_client.post("/getHLSStreamingSessionURL")
# Just checking server is up
res.status_code.should.equal(404)
assert res.status_code == 404