Techdebt: Replace sure with regular assertions in MediaLive (#6651)

This commit is contained in:
Bert Blommers 2023-08-13 16:59:51 +00:00 committed by GitHub
parent dc8015c91d
commit 5f35156d3d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 83 additions and 109 deletions

View File

@ -1,5 +1,4 @@
import boto3 import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_medialive from moto import mock_medialive
from uuid import uuid4 from uuid import uuid4
@ -111,21 +110,16 @@ def test_create_channel_succeeds():
channel_config = _create_channel_config("test channel 1") channel_config = _create_channel_config("test channel 1")
response = client.create_channel(**channel_config) response = client.create_channel(**channel_config)
assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) channel = response["Channel"]
response["Channel"]["Arn"].should.equal( assert channel["Arn"] == f"arn:aws:medialive:channel:{response['Channel']['Id']}"
f"arn:aws:medialive:channel:{response['Channel']['Id']}" assert channel["Destinations"] == channel_config["Destinations"]
) assert channel["EncoderSettings"] == channel_config["EncoderSettings"]
response["Channel"]["Destinations"].should.equal(channel_config["Destinations"]) assert channel["InputAttachments"] == channel_config["InputAttachments"]
response["Channel"]["EncoderSettings"].should.equal( assert channel["Name"] == "test channel 1"
channel_config["EncoderSettings"] assert channel["State"] == "CREATING"
) assert channel["Tags"]["Customer"] == "moto"
response["Channel"]["InputAttachments"].should.equal(
channel_config["InputAttachments"]
)
response["Channel"]["Name"].should.equal("test channel 1")
response["Channel"]["State"].should.equal("CREATING")
response["Channel"]["Tags"]["Customer"].should.equal("moto")
@mock_medialive @mock_medialive
@ -139,15 +133,15 @@ def test_list_channels_succeeds():
client.create_channel(**channel2_config) client.create_channel(**channel2_config)
response = client.list_channels() response = client.list_channels()
len(response["Channels"]).should.equal(2) assert len(response["Channels"]) == 2
response["Channels"][0]["Name"].should.equal("test channel 1") assert response["Channels"][0]["Name"] == "test channel 1"
response["Channels"][0]["ChannelClass"].should.equal("STANDARD") assert response["Channels"][0]["ChannelClass"] == "STANDARD"
response["Channels"][0]["PipelinesRunningCount"].should.equal(2) assert response["Channels"][0]["PipelinesRunningCount"] == 2
response["Channels"][1]["Name"].should.equal("test channel 2") assert response["Channels"][1]["Name"] == "test channel 2"
response["Channels"][1]["ChannelClass"].should.equal("SINGLE_PIPELINE") assert response["Channels"][1]["ChannelClass"] == "SINGLE_PIPELINE"
response["Channels"][1]["PipelinesRunningCount"].should.equal(1) assert response["Channels"][1]["PipelinesRunningCount"] == 1
@mock_medialive @mock_medialive
@ -156,11 +150,11 @@ def test_delete_channel_moves_channel_in_deleted_state():
channel_name = "test channel X" channel_name = "test channel X"
channel_config = _create_channel_config(channel_name) channel_config = _create_channel_config(channel_name)
create_response = client.create_channel(**channel_config) channel_id = client.create_channel(**channel_config)["Channel"]["Id"]
delete_response = client.delete_channel(ChannelId=create_response["Channel"]["Id"]) delete_response = client.delete_channel(ChannelId=channel_id)
delete_response["Name"].should.equal(channel_name) assert delete_response["Name"] == channel_name
delete_response["State"].should.equal("DELETING") assert delete_response["State"] == "DELETING"
@mock_medialive @mock_medialive
@ -169,22 +163,16 @@ def test_describe_channel_succeeds():
channel_name = "test channel X" channel_name = "test channel X"
channel_config = _create_channel_config(channel_name) channel_config = _create_channel_config(channel_name)
create_response = client.create_channel(**channel_config) channel_id = client.create_channel(**channel_config)["Channel"]["Id"]
describe_response = client.describe_channel( channel = client.describe_channel(ChannelId=channel_id)
ChannelId=create_response["Channel"]["Id"]
)
describe_response["Arn"].should.equal( assert channel["Arn"] == f"arn:aws:medialive:channel:{channel['Id']}"
f"arn:aws:medialive:channel:{describe_response['Id']}" assert channel["Destinations"] == channel_config["Destinations"]
) assert channel["EncoderSettings"] == channel_config["EncoderSettings"]
describe_response["Destinations"].should.equal(channel_config["Destinations"]) assert channel["InputAttachments"] == channel_config["InputAttachments"]
describe_response["EncoderSettings"].should.equal(channel_config["EncoderSettings"]) assert channel["Name"] == channel_name
describe_response["InputAttachments"].should.equal( assert channel["State"] == "IDLE"
channel_config["InputAttachments"] assert channel["Tags"]["Customer"] == "moto"
)
describe_response["Name"].should.equal(channel_name)
describe_response["State"].should.equal("IDLE")
describe_response["Tags"]["Customer"].should.equal("moto")
@mock_medialive @mock_medialive
@ -193,15 +181,12 @@ def test_start_channel_succeeds():
channel_name = "testchan1" channel_name = "testchan1"
channel_config = _create_channel_config(channel_name) channel_config = _create_channel_config(channel_name)
create_response = client.create_channel(**channel_config) channel_id = client.create_channel(**channel_config)["Channel"]["Id"]
start_response = client.start_channel(ChannelId=create_response["Channel"]["Id"]) start_response = client.start_channel(ChannelId=channel_id)
start_response["Name"].should.equal(channel_name) assert start_response["Name"] == channel_name
start_response["State"].should.equal("STARTING") assert start_response["State"] == "STARTING"
describe_response = client.describe_channel( assert client.describe_channel(ChannelId=channel_id)["State"] == "RUNNING"
ChannelId=create_response["Channel"]["Id"]
)
describe_response["State"].should.equal("RUNNING")
@mock_medialive @mock_medialive
@ -210,18 +195,14 @@ def test_stop_channel_succeeds():
channel_name = "testchan2" channel_name = "testchan2"
channel_config = _create_channel_config(channel_name) channel_config = _create_channel_config(channel_name)
create_response = client.create_channel(**channel_config) channel_id = client.create_channel(**channel_config)["Channel"]["Id"]
channel_id = create_response["Channel"]["Id"]
assert len(channel_id) > 1 assert len(channel_id) > 1
client.start_channel(ChannelId=channel_id) client.start_channel(ChannelId=channel_id)
stop_response = client.stop_channel(ChannelId=channel_id) stop_response = client.stop_channel(ChannelId=channel_id)
stop_response["Name"].should.equal(channel_name) assert stop_response["Name"] == channel_name
stop_response["State"].should.equal("STOPPING") assert stop_response["State"] == "STOPPING"
describe_response = client.describe_channel( assert client.describe_channel(ChannelId=channel_id)["State"] == "IDLE"
ChannelId=create_response["Channel"]["Id"]
)
describe_response["State"].should.equal("IDLE")
@mock_medialive @mock_medialive
@ -230,19 +211,17 @@ def test_update_channel_succeeds():
channel_name = "Original Channel" channel_name = "Original Channel"
channel_config = _create_channel_config(channel_name) channel_config = _create_channel_config(channel_name)
create_response = client.create_channel(**channel_config) channel_id = client.create_channel(**channel_config)["Channel"]["Id"]
channel_id = create_response["Channel"]["Id"]
assert len(channel_id) > 1
update_response = client.update_channel( updated_channel = client.update_channel(
ChannelId=channel_id, Name="Updated Channel" ChannelId=channel_id, Name="Updated Channel"
) )["Channel"]
update_response["Channel"]["State"].should.equal("UPDATING") assert updated_channel["State"] == "UPDATING"
update_response["Channel"]["Name"].should.equal("Updated Channel") assert updated_channel["Name"] == "Updated Channel"
describe_response = client.describe_channel(ChannelId=channel_id) channel = client.describe_channel(ChannelId=channel_id)
describe_response["State"].should.equal("IDLE") assert channel["State"] == "IDLE"
describe_response["Name"].should.equal("Updated Channel") assert channel["Name"] == "Updated Channel"
@mock_medialive @mock_medialive
@ -252,24 +231,24 @@ def test_create_input_succeeds():
input_config = _create_input_config(input_name) input_config = _create_input_config(input_name)
create_response = client.create_input(**input_config) create_response = client.create_input(**input_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert create_response["ResponseMetadata"]["HTTPStatusCode"] == 200
r_input = create_response["Input"] r_input = create_response["Input"]
input_id = r_input["Id"] input_id = r_input["Id"]
assert len(input_id) > 1 assert len(input_id) > 1
r_input["Arn"].should.equal(f"arn:aws:medialive:input:{r_input['Id']}") assert r_input["Arn"] == f"arn:aws:medialive:input:{r_input['Id']}"
r_input["Name"].should.equal(input_name) assert r_input["Name"] == input_name
r_input["AttachedChannels"].should.equal([]) assert r_input["AttachedChannels"] == []
r_input["Destinations"].should.equal(input_config["Destinations"]) assert r_input["Destinations"] == input_config["Destinations"]
r_input["InputClass"].should.equal("STANDARD") assert r_input["InputClass"] == "STANDARD"
r_input["InputDevices"].should.equal(input_config["InputDevices"]) assert r_input["InputDevices"] == input_config["InputDevices"]
r_input["InputSourceType"].should.equal("STATIC") assert r_input["InputSourceType"] == "STATIC"
r_input["MediaConnectFlows"].should.equal(input_config["MediaConnectFlows"]) assert r_input["MediaConnectFlows"] == input_config["MediaConnectFlows"]
r_input["RoleArn"].should.equal(input_config["RoleArn"]) assert r_input["RoleArn"] == input_config["RoleArn"]
r_input["SecurityGroups"].should.equal([]) assert r_input["SecurityGroups"] == []
r_input["Sources"].should.equal(input_config["Sources"]) assert r_input["Sources"] == input_config["Sources"]
r_input["State"].should.equal("CREATING") assert r_input["State"] == "CREATING"
r_input["Tags"].should.equal(input_config["Tags"]) assert r_input["Tags"] == input_config["Tags"]
r_input["Type"].should.equal(input_config["Type"]) assert r_input["Type"] == input_config["Type"]
@mock_medialive @mock_medialive
@ -279,16 +258,14 @@ def test_describe_input_succeeds():
input_config = _create_input_config(input_name) input_config = _create_input_config(input_name)
create_response = client.create_input(**input_config) create_response = client.create_input(**input_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert create_response["ResponseMetadata"]["HTTPStatusCode"] == 200
create_response["Input"]["State"].should.equal("CREATING") assert create_response["Input"]["State"] == "CREATING"
describe_response = client.describe_input(InputId=create_response["Input"]["Id"]) describe_response = client.describe_input(InputId=create_response["Input"]["Id"])
describe_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert describe_response["ResponseMetadata"]["HTTPStatusCode"] == 200
describe_response["Name"].should.equal(input_name) assert describe_response["Name"] == input_name
describe_response["State"].should.equal("DETACHED") assert describe_response["State"] == "DETACHED"
describe_response["MediaConnectFlows"].should.equal( assert describe_response["MediaConnectFlows"] == input_config["MediaConnectFlows"]
input_config["MediaConnectFlows"]
)
@mock_medialive @mock_medialive
@ -299,11 +276,11 @@ def test_list_inputs_succeeds():
input_config2 = _create_input_config("Input Two") input_config2 = _create_input_config("Input Two")
client.create_input(**input_config2) client.create_input(**input_config2)
response = client.list_inputs() inputs = client.list_inputs()["Inputs"]
len(response["Inputs"]).should.equal(2) assert len(inputs) == 2
response["Inputs"][0]["Name"].should.equal("Input One") assert inputs[0]["Name"] == "Input One"
response["Inputs"][1]["Name"].should.equal("Input Two") assert inputs[1]["Name"] == "Input Two"
@mock_medialive @mock_medialive
@ -312,13 +289,13 @@ def test_delete_input_moves_input_in_deleted_state():
input_name = "test input X" input_name = "test input X"
input_config = _create_input_config(input_name) input_config = _create_input_config(input_name)
create_response = client.create_input(**input_config) input_id = client.create_input(**input_config)["Input"]["Id"]
delete_response = client.delete_input(InputId=create_response["Input"]["Id"]) response = client.delete_input(InputId=input_id)
delete_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
describe_response = client.describe_input(InputId=create_response["Input"]["Id"]) input_ = client.describe_input(InputId=input_id)
describe_response["Name"].should.equal(input_name) assert input_["Name"] == input_name
describe_response["State"].should.equal("DELETED") assert input_["State"] == "DELETED"
@mock_medialive @mock_medialive
@ -327,8 +304,6 @@ def test_update_input_succeeds():
input_name = "test input X" input_name = "test input X"
input_config = _create_input_config(input_name) input_config = _create_input_config(input_name)
create_response = client.create_input(**input_config) input_id = client.create_input(**input_config)["Input"]["Id"]
update_response = client.update_input( input_ = client.update_input(InputId=input_id, Name="test input U")
InputId=create_response["Input"]["Id"], Name="test input U" assert input_["Input"]["Name"] == "test input U"
)
update_response["Input"]["Name"].should.equal("test input U")

View File

@ -1,5 +1,4 @@
import sure # noqa # pylint: disable=unused-import import json
import moto.server as server import moto.server as server
from moto import mock_medialive from moto import mock_medialive
@ -15,7 +14,7 @@ def test_medialive_list_channels():
res = test_client.get("/prod/channels") res = test_client.get("/prod/channels")
result = res.data.decode("utf-8") result = res.data.decode("utf-8")
result.should.contain('"channels": []') assert json.loads(result) == {"channels": [], "nextToken": None}
@mock_medialive @mock_medialive
@ -26,4 +25,4 @@ def test_medialive_list_inputs():
res = test_client.get("/prod/inputs") res = test_client.get("/prod/inputs")
result = res.data.decode("utf-8") result = res.data.decode("utf-8")
result.should.contain('"inputs": []') assert json.loads(result) == {"inputs": [], "nextToken": None}