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

Add outputs and vpc interfaces to a mediaconnect flow (#4034)

Browse source 
* Add outputs and vpc interfaces to a mediaconnect flow

* Add negative tests for add_flow_outputs and add_flow_vpc_interfaces

* fix: fstring to format

* MediaConnect - add appropriate URLs for ServerMode tests

Co-authored-by: Bert Blommers <info@bertblommers.nl>
This commit is contained in:
Alexandre Blanchet 2021-06-25 16:31:05 +02:00 • committed by GitHub
commit 167423777b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 136 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

95
tests/test_mediaconnect/test_mediaconnect.py
View file

@ -1,9 +1,12 @@
from __future__ import unicode_literals
import boto3
import botocore
import pytest
import sure # noqa
from moto import mock_mediaconnect
from botocore.exceptions import ClientError
from moto import mock_mediaconnect
region = "eu-west-1"
@ -153,3 +156,93 @@ def test_tag_resource_succeeds():
list_response = client.list_tags_for_resource(ResourceArn="some-arn")
list_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
list_response["Tags"].should.equal({"Tag1": "Value1"})
@mock_mediaconnect
def test_add_flow_vpc_interfaces_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
create_response["Flow"]["Status"].should.equal("STANDBY")
flow_arn = create_response["Flow"]["FlowArn"]
client.add_flow_vpc_interfaces(
FlowArn=flow_arn,
VpcInterfaces=[
{
"Name": "VPCInterface",
"SubnetId": "",
"SecurityGroupIds": [],
"RoleArn": "",
}
],
)
describe_response = client.describe_flow(FlowArn=flow_arn)
describe_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
describe_response["Flow"]["VpcInterfaces"].should.equal(
[
{
"Name": "VPCInterface",
"RoleArn": "",
"SecurityGroupIds": [],
"SubnetId": "",
}
]
)
@mock_mediaconnect
def test_add_flow_vpc_interfaces_fails():
client = boto3.client("mediaconnect", region_name=region)
flow_arn = "unknown-flow"
with pytest.raises(ClientError) as err:
client.add_flow_vpc_interfaces(
FlowArn=flow_arn, VpcInterfaces=[],
)
err = err.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"flow with arn=unknown-flow not found".format(str(flow_arn))
)
@mock_mediaconnect
def test_add_flow_outputs_succeeds():
client = boto3.client("mediaconnect", region_name=region)
channel_config = _create_flow_config("test Flow 1")
create_response = client.create_flow(**channel_config)
create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
create_response["Flow"]["Status"].should.equal("STANDBY")
flow_arn = create_response["Flow"]["FlowArn"]
client.add_flow_outputs(
FlowArn=flow_arn,
Outputs=[
{"Description": "string", "Name": "string", "Port": 123, "Protocol": "rist"}
],
)
describe_response = client.describe_flow(FlowArn=flow_arn)
describe_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
describe_response["Flow"]["Outputs"].should.equal(
[{"Description": "string", "Name": "string", "Port": 123}]
)
@mock_mediaconnect
def test_add_flow_outputs_fails():
client = boto3.client("mediaconnect", region_name=region)
flow_arn = "unknown-flow"
with pytest.raises(ClientError) as err:
client.add_flow_outputs(
FlowArn=flow_arn, Outputs=[],
)
err = err.value.response["Error"]
err["Code"].should.equal("NotFoundException")
err["Message"].should.equal(
"flow with arn=unknown-flow not found".format(str(flow_arn))
)