Bugfix/mediaconnect create flow include source ingest ip (#5093)

moto/mediaconnect/models.py

@@ -78,6 +78,29 @@ class MediaConnectBackend(BaseBackend):
         self._flows = OrderedDict()
         self._resources = OrderedDict()
 
+    def _add_source_details(self, source, flow_id, ingest_ip="127.0.0.1"):
+        if source:
+            source["sourceArn"] = (
+                f"arn:aws:mediaconnect:{self.region_name}:{ACCOUNT_ID}:source"
+                f":{flow_id}:{source['name']}"
+            )
+            if not source.get("entitlementArn"):
+                source["ingestIp"] = ingest_ip
+
+    def _create_flow_add_details(self, flow):
+        flow_id = uuid4().hex
+
+        flow.description = "A Moto test flow"
+        flow.egress_ip = "127.0.0.1"
+        flow.flow_arn = f"arn:aws:mediaconnect:{self.region_name}:{ACCOUNT_ID}:flow:{flow_id}:{flow.name}"
+
+        for index, _source in enumerate(flow.sources):
+            self._add_source_details(_source, flow_id, f"127.0.0.{index}")
+
+        for index, output in enumerate(flow.outputs):
+            if output.get("protocol") in ["srt-listener", "zixi-pull"]:
+                output["listenerAddress"] = f"{index}.0.0.0"
+
     def reset(self):
         region_name = self.region_name
         self.__dict__ = {}
@@ -94,12 +117,6 @@ class MediaConnectBackend(BaseBackend):
         sources,
         vpc_interfaces,
     ):
-        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,9 +127,7 @@ class MediaConnectBackend(BaseBackend):
             sources=sources,
             vpc_interfaces=vpc_interfaces,
         )
-        flow.description = "A Moto test flow"
+        self._create_flow_add_details(flow)
-        flow.egress_ip = "127.0.0.1"
-        flow.flow_arn = f"arn:aws:mediaconnect:{self.region_name}:{ACCOUNT_ID}:flow:{flow_id}:{name}"
         self._flows[flow.flow_arn] = flow
         return flow
 

moto/mediaconnect/responses.py

@@ -18,9 +18,6 @@ class MediaConnectResponse(BaseResponse):
         entitlements = self._get_param("entitlements")
         name = self._get_param("name")
         outputs = self._get_param("outputs")
-        for index, output in enumerate(outputs):
-            if output.get("protocol") in ["srt-listener", "zixi-pull"]:
-                output["listenerAddress"] = f"{index}.0.0.0"
         source = self._get_param("source")
         source_failover_config = self._get_param("sourceFailoverConfig")
         sources = self._get_param("sources")

tests/test_mediaconnect/test_mediaconnect.py

@@ -11,6 +11,14 @@ from moto.core import ACCOUNT_ID
 region = "eu-west-1"
 
 
+def _source(name="Source-A"):
+    return {
+        "Decryption": {"Algorithm": "aes256", "RoleArn": "some:role"},
+        "Description": "A source",
+        "Name": name,
+    }
+
+
 def _create_flow_config(name, **kwargs):
     availability_zone = kwargs.get("availability_zone", "AZ1")
     entitlements = kwargs.get(
@@ -36,25 +44,24 @@ def _create_flow_config(name, **kwargs):
     )
     source = kwargs.get(
         "source",
-        {
+        _source(),
-            "Decryption": {"Algorithm": "aes256", "RoleArn": "some:role"},
-            "Description": "A source",
-            "Name": "Source-A",
-        },
     )
     source_failover_config = kwargs.get("source_failover_config", {})
     sources = kwargs.get("sources", [])
     vpc_interfaces = kwargs.get("vpc_interfaces", [])
-    flow_config = dict(
+    flow_config = dict(Name=name)
+    optional_flow_config = dict(
         AvailabilityZone=availability_zone,
         Entitlements=entitlements,
-        Name=name,
         Outputs=outputs,
         Source=source,
         SourceFailoverConfig=source_failover_config,
         Sources=sources,
         VpcInterfaces=vpc_interfaces,
     )
+    for key, value in optional_flow_config.items():
+        if value:
+            flow_config[key] = value
     return flow_config
 
 
@@ -83,6 +90,36 @@ def test_create_flow_succeeds():
     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"]["Source"]["IngestIp"].should.equal("127.0.0.0")
+    _check_mediaconnect_arn(
+        type_="source", arn=response["Flow"]["Sources"][0]["SourceArn"], name="Source-A"
+    )
+
+
+@mock_mediaconnect
+def test_create_flow_alternative_succeeds():
+    client = boto3.client("mediaconnect", region_name=region)
+    channel_config = _create_flow_config(
+        "test-Flow-1",
+        source=None,
+        sources=[_source(), _source("Source-B")],
+        source_failover_config={
+            "FailoverMode": "FAILOVER",
+            "SourcePriority": {"PrimarySource": "Source-B"},
+            "State": "ENABLED",
+        },
+    )
+
+    response = client.create_flow(**channel_config)
+
+    response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
+    _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"]["Sources"][0]["IngestIp"].should.equal("127.0.0.0")
+    response["Flow"]["Sources"][1]["IngestIp"].should.equal("127.0.0.1")
     _check_mediaconnect_arn(
         type_="source", arn=response["Flow"]["Sources"][0]["SourceArn"], name="Source-A"
     )