Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

mediaconnect: fix flow and source arn format and update tests to use … (#5070)

Browse Source
This commit is contained in:
nickphbrown 2022-04-28 10:52:48 +01:00 committed by GitHub
parent 56806fb310
commit 9ab77d505d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 45 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
moto/mediaconnect/models.py
View File

@ -1,7 +1,7 @@
from collections import OrderedDict
from uuid import uuid4
from moto.core import BaseBackend, BaseModel
from moto.core import ACCOUNT_ID, BaseBackend, BaseModel
from moto.core.utils import BackendDict
from moto.mediaconnect.exceptions import NotFoundException
@ -94,10 +94,12 @@ class MediaConnectBackend(BaseBackend):
sources,
vpc_interfaces,
):
if isinstance(source, dict) and source.get("name"):
source["sourceArn"] = "arn:aws:mediaconnect:source:{}".format(
source["name"]
)
flow_id = uuid4().hex
source_name = source.get("name")
if isinstance(source, dict) and source_name:
source[
"sourceArn"
] = f"arn:aws:mediaconnect:{self.region_name}:{ACCOUNT_ID}:source:{flow_id}:{source_name}"
flow = Flow(
availability_zone=availability_zone,
entitlements=entitlements,
@ -110,8 +112,7 @@ class MediaConnectBackend(BaseBackend):
)
flow.description = "A Moto test flow"
flow.egress_ip = "127.0.0.1"
flow_id = uuid4().hex
flow.flow_arn = "arn:aws:mediaconnect:flow:{}".format(flow_id)
flow.flow_arn = f"arn:aws:mediaconnect:{self.region_name}:{ACCOUNT_ID}:flow:{flow_id}:{name}"
self._flows[flow.flow_arn] = flow
return flow

60
tests/test_mediaconnect/test_mediaconnect.py
View File

@ -1,9 +1,12 @@
from uuid import UUID
import boto3
import pytest
import sure # noqa # pylint: disable=unused-import
from botocore.exceptions import ClientError
from moto import mock_mediaconnect
from moto.core import ACCOUNT_ID
region = "eu-west-1"
@ -18,7 +21,7 @@ def _create_flow_config(name, **kwargs):
"Description": "An entitlement",
"Encryption": {"Algorithm": "aes256", "RoleArn": "some:role"},
"EntitlementStatus": "ENABLED",
"Name": "Entitlement A",
"Name": "Entitlement-A",
"Subscribers": [],
}
],
@ -26,9 +29,9 @@ def _create_flow_config(name, **kwargs):
outputs = kwargs.get(
"outputs",
[
{"Name": "Output 1", "Protocol": "zixi-push"},
{"Name": "Output 2", "Protocol": "zixi-pull"},
{"Name": "Output 3", "Protocol": "srt-listener"},
{"Name": "Output-1", "Protocol": "zixi-push"},
{"Name": "Output-2", "Protocol": "zixi-pull"},
{"Name": "Output-3", "Protocol": "srt-listener"},
],
)
source = kwargs.get(
@ -36,7 +39,7 @@ def _create_flow_config(name, **kwargs):
{
"Decryption": {"Algorithm": "aes256", "RoleArn": "some:role"},
"Description": "A source",
"Name": "Source A",
"Name": "Source-A",
},
)
source_failover_config = kwargs.get("source_failover_config", {})
@ -55,30 +58,41 @@ def _create_flow_config(name, **kwargs):
return flow_config
def _check_mediaconnect_arn(type_, arn, name):
_arn_list = str.split(arn, ":")
_arn_list[:6].should.equal(
["arn", "aws", "mediaconnect", region, ACCOUNT_ID, type_]
)
UUID(_arn_list[6])
_arn_list[-1].should.equal(name)
@mock_mediaconnect
def test_create_flow_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
response = client.create_flow(**channel_config)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response["Flow"]["FlowArn"][:26].should.equal("arn:aws:mediaconnect:flow:")
response["Flow"]["Name"].should.equal("test Flow 1")
_check_mediaconnect_arn(
type_="flow", arn=response["Flow"]["FlowArn"], name="test-Flow-1"
)
response["Flow"]["Name"].should.equal("test-Flow-1")
response["Flow"]["Status"].should.equal("STANDBY")
response["Flow"]["Outputs"][0].should.equal({"Name": "Output 1"})
response["Flow"]["Outputs"][0].should.equal({"Name": "Output-1"})
response["Flow"]["Outputs"][1]["ListenerAddress"].should.equal("1.0.0.0")
response["Flow"]["Outputs"][2]["ListenerAddress"].should.equal("2.0.0.0")
response["Flow"]["Sources"][0]["SourceArn"].should.equal(
"arn:aws:mediaconnect:source:Source A"
_check_mediaconnect_arn(
type_="source", arn=response["Flow"]["Sources"][0]["SourceArn"], name="Source-A"
)
@mock_mediaconnect
def test_list_flows_succeeds():
client = boto3.client("mediaconnect", region_name=region)
flow_1_config = _create_flow_config("test Flow 1")
flow_2_config = _create_flow_config("test Flow 2")
flow_1_config = _create_flow_config("test-Flow-1")
flow_2_config = _create_flow_config("test-Flow-2")
client.create_flow(**flow_1_config)
client.create_flow(**flow_2_config)
@ -86,12 +100,12 @@ def test_list_flows_succeeds():
response = client.list_flows()
len(response["Flows"]).should.equal(2)
response["Flows"][0]["Name"].should.equal("test Flow 1")
response["Flows"][0]["Name"].should.equal("test-Flow-1")
response["Flows"][0]["AvailabilityZone"].should.equal("AZ1")
response["Flows"][0]["SourceType"].should.equal("OWNED")
response["Flows"][0]["Status"].should.equal("STANDBY")
response["Flows"][1]["Name"].should.equal("test Flow 2")
response["Flows"][1]["Name"].should.equal("test-Flow-2")
response["Flows"][1]["AvailabilityZone"].should.equal("AZ1")
response["Flows"][1]["SourceType"].should.equal("OWNED")
response["Flows"][1]["Status"].should.equal("STANDBY")
@ -100,20 +114,20 @@ def test_list_flows_succeeds():
@mock_mediaconnect
def test_describe_flow_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
flow_arn = create_response["Flow"]["FlowArn"]
describe_response = client.describe_flow(FlowArn=flow_arn)
describe_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
describe_response["Flow"]["Name"].should.equal("test Flow 1")
describe_response["Flow"]["Name"].should.equal("test-Flow-1")
@mock_mediaconnect
def test_delete_flow_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -127,7 +141,7 @@ def test_delete_flow_succeeds():
@mock_mediaconnect
def test_start_stop_flow_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -168,7 +182,7 @@ def test_tag_resource_succeeds():
@mock_mediaconnect
def test_add_flow_vpc_interfaces_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -215,7 +229,7 @@ def test_add_flow_vpc_interfaces_fails():
@mock_mediaconnect
def test_remove_flow_vpc_interface_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -260,7 +274,7 @@ def test_remove_flow_vpc_interface_fails():
@mock_mediaconnect
def test_add_flow_outputs_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@ -307,7 +321,7 @@ def test_remove_flow_output_fails():
@mock_mediaconnect
def test_remove_flow_output_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
channel_config = _create_flow_config("test-Flow-1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)