diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index e0444a787..f143d52dd 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/IMPLEMENTATION_COVERAGE.md @@ -5616,25 +5616,25 @@ ## medialive
-0% implemented +26% implemented - [ ] batch_update_schedule -- [ ] create_channel -- [ ] create_input +- [x] create_channel +- [x] create_input - [ ] create_input_security_group - [ ] create_multiplex - [ ] create_multiplex_program - [ ] create_tags -- [ ] delete_channel -- [ ] delete_input +- [x] delete_channel +- [x] delete_input - [ ] delete_input_security_group - [ ] delete_multiplex - [ ] delete_multiplex_program - [ ] delete_reservation - [ ] delete_schedule - [ ] delete_tags -- [ ] describe_channel -- [ ] describe_input +- [x] describe_channel +- [x] describe_input - [ ] describe_input_device - [ ] describe_input_security_group - [ ] describe_multiplex @@ -5642,23 +5642,23 @@ - [ ] describe_offering - [ ] describe_reservation - [ ] describe_schedule -- [ ] list_channels +- [x] list_channels - [ ] list_input_devices - [ ] list_input_security_groups -- [ ] list_inputs +- [x] list_inputs - [ ] list_multiplex_programs - [ ] list_multiplexes - [ ] list_offerings - [ ] list_reservations - [ ] list_tags_for_resource - [ ] purchase_offering -- [ ] start_channel +- [x] start_channel - [ ] start_multiplex -- [ ] stop_channel +- [x] stop_channel - [ ] stop_multiplex -- [ ] update_channel +- [x] update_channel - [ ] update_channel_class -- [ ] update_input +- [x] update_input - [ ] update_input_device - [ ] update_input_security_group - [ ] update_multiplex diff --git a/moto/__init__.py b/moto/__init__.py index fd467cbf8..972d0836b 100644 --- a/moto/__init__.py +++ b/moto/__init__.py @@ -119,6 +119,7 @@ mock_kinesisvideo = lazy_load(".kinesisvideo", "mock_kinesisvideo") mock_kinesisvideoarchivedmedia = lazy_load( ".kinesisvideoarchivedmedia", "mock_kinesisvideoarchivedmedia" ) +mock_medialive = lazy_load(".medialive", "mock_medialive") # import logging # logging.getLogger('boto').setLevel(logging.CRITICAL) diff --git a/moto/backends.py b/moto/backends.py index c8bac72fc..ac9384136 100644 --- a/moto/backends.py +++ b/moto/backends.py @@ -71,6 +71,7 @@ BACKENDS = { "transcribe": ("transcribe", "transcribe_backends"), "xray": ("xray", "xray_backends"), "kinesisvideo": ("kinesisvideo", "kinesisvideo_backends"), + "medialive": ("medialive", "medialive_backends"), "kinesis-video-archived-media": ( "kinesisvideoarchivedmedia", "kinesisvideoarchivedmedia_backends", diff --git a/moto/core/responses.py b/moto/core/responses.py index 1149ab0be..68493c87d 100644 --- a/moto/core/responses.py +++ b/moto/core/responses.py @@ -324,7 +324,12 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin): def _convert(elem, is_last): if not re.match("^{.*}$", elem): return elem - name = elem.replace("{", "").replace("}", "").replace("+", "") + name = ( + elem.replace("{", "") + .replace("}", "") + .replace("+", "") + .replace("-", "_") + ) if is_last: return "(?P<%s>[^/]*)" % name return "(?P<%s>.*)" % name diff --git a/moto/medialive/__init__.py b/moto/medialive/__init__.py new file mode 100644 index 000000000..d25d2beda --- /dev/null +++ b/moto/medialive/__init__.py @@ -0,0 +1,6 @@ +from __future__ import unicode_literals +from .models import medialive_backends +from ..core.models import base_decorator + +medialive_backend = medialive_backends["us-east-1"] +mock_medialive = base_decorator(medialive_backends) diff --git a/moto/medialive/exceptions.py b/moto/medialive/exceptions.py new file mode 100644 index 000000000..baffc4882 --- /dev/null +++ b/moto/medialive/exceptions.py @@ -0,0 +1 @@ +from __future__ import unicode_literals diff --git a/moto/medialive/models.py b/moto/medialive/models.py new file mode 100644 index 000000000..023fe5bb9 --- /dev/null +++ b/moto/medialive/models.py @@ -0,0 +1,302 @@ +from __future__ import unicode_literals + +from collections import OrderedDict +from uuid import uuid4 + +from boto3 import Session + +from moto.core import BaseBackend, BaseModel + + +class Input(BaseModel): + def __init__(self, *args, **kwargs): + self.arn = kwargs.get("arn") + self.attached_channels = kwargs.get("attached_channels", []) + self.destinations = kwargs.get("destinations", []) + self.input_id = kwargs.get("input_id") + self.input_class = kwargs.get("input_class", "STANDARD") + self.input_devices = kwargs.get("input_devices", []) + self.input_source_type = kwargs.get("input_source_type", "STATIC") + self.media_connect_flows = kwargs.get("media_connect_flows", []) + self.name = kwargs.get("name") + self.role_arn = kwargs.get("role_arn") + self.security_groups = kwargs.get("security_groups", []) + self.sources = kwargs.get("sources", []) + # Possible states: 'CREATING'|'DETACHED'|'ATTACHED'|'DELETING'|'DELETED' + self.state = kwargs.get("state") + self.tags = kwargs.get("tags") + self.input_type = kwargs.get("input_type") + + def to_dict(self): + data = { + "arn": self.arn, + "attachedChannels": self.attached_channels, + "destinations": self.destinations, + "id": self.input_id, + "inputClass": self.input_class, + "inputDevices": self.input_devices, + "inputSourceType": self.input_source_type, + "mediaConnectFlows": self.media_connect_flows, + "name": self.name, + "roleArn": self.role_arn, + "securityGroups": self.security_groups, + "sources": self.sources, + "state": self.state, + "tags": self.tags, + "type": self.input_type, + } + return data + + def _resolve_transient_states(self): + # Resolve transient states before second call + # (to simulate AWS taking its sweet time with these things) + if self.state in ["CREATING"]: + self.state = "DETACHED" # or ATTACHED + elif self.state == "DELETING": + self.state = "DELETED" + + +class Channel(BaseModel): + def __init__(self, *args, **kwargs): + self.arn = kwargs.get("arn") + self.cdi_input_specification = kwargs.get("cdi_input_specification") + self.channel_class = kwargs.get("channel_class", "STANDARD") + self.destinations = kwargs.get("destinations") + self.egress_endpoints = kwargs.get("egress_endpoints", []) + self.encoder_settings = kwargs.get("encoder_settings") + self.channel_id = kwargs.get("channel_id") + self.input_attachments = kwargs.get("input_attachments") + self.input_specification = kwargs.get("input_specification") + self.log_level = kwargs.get("log_level") + self.name = kwargs.get("name") + self.pipeline_details = kwargs.get("pipeline_details", []) + self.role_arn = kwargs.get("role_arn") + self.state = kwargs.get("state") + self.tags = kwargs.get("tags") + self._previous_state = None + + def to_dict(self, exclude=None): + data = { + "arn": self.arn, + "cdiInputSpecification": self.cdi_input_specification, + "channelClass": self.channel_class, + "destinations": self.destinations, + "egressEndpoints": self.egress_endpoints, + "encoderSettings": self.encoder_settings, + "id": self.channel_id, + "inputAttachments": self.input_attachments, + "inputSpecification": self.input_specification, + "logLevel": self.log_level, + "name": self.name, + "pipelineDetails": self.pipeline_details, + "pipelinesRunningCount": 1 + if self.channel_class == "SINGLE_PIPELINE" + else 2, + "roleArn": self.role_arn, + "state": self.state, + "tags": self.tags, + } + if exclude: + for key in exclude: + del data[key] + return data + + def _resolve_transient_states(self): + # Resolve transient states before second call + # (to simulate AWS taking its sweet time with these things) + if self.state in ["CREATING", "STOPPING"]: + self.state = "IDLE" + elif self.state == "STARTING": + self.state = "RUNNING" + elif self.state == "DELETING": + self.state = "DELETED" + elif self.state == "UPDATING": + self.state = self._previous_state + self._previous_state = None + + +class MediaLiveBackend(BaseBackend): + def __init__(self, region_name=None): + super(MediaLiveBackend, self).__init__() + self.region_name = region_name + self._channels = OrderedDict() + self._inputs = OrderedDict() + + def reset(self): + region_name = self.region_name + self.__dict__ = {} + self.__init__(region_name) + + def create_channel( + self, + cdi_input_specification, + channel_class, + destinations, + encoder_settings, + input_attachments, + input_specification, + log_level, + name, + request_id, + reserved, + role_arn, + tags, + ): + channel_id = uuid4().hex + arn = "arn:aws:medialive:channel:{}".format(channel_id) + channel = Channel( + arn=arn, + cdi_input_specification=cdi_input_specification, + channel_class=channel_class or "STANDARD", + destinations=destinations, + egress_endpoints=[], + encoder_settings=encoder_settings, + channel_id=channel_id, + input_attachments=input_attachments, + input_specification=input_specification, + log_level=log_level, + name=name, + pipeline_details=[], + role_arn=role_arn, + state="CREATING", + tags=tags, + ) + self._channels[channel_id] = channel + return channel + + def list_channels(self, max_results, next_token): + channels = list(self._channels.values()) + if max_results is not None: + channels = channels[:max_results] + response_channels = [ + c.to_dict(exclude=["encoderSettings", "pipelineDetails"]) for c in channels + ] + return response_channels, next_token + + def describe_channel(self, channel_id): + channel = self._channels[channel_id] + channel._resolve_transient_states() + return channel.to_dict() + + def delete_channel(self, channel_id): + channel = self._channels[channel_id] + channel.state = "DELETING" + return channel.to_dict() + + def start_channel(self, channel_id): + channel = self._channels[channel_id] + channel.state = "STARTING" + return channel.to_dict() + + def stop_channel(self, channel_id): + channel = self._channels[channel_id] + channel.state = "STOPPING" + return channel.to_dict() + + def update_channel( + self, + channel_id, + cdi_input_specification, + destinations, + encoder_settings, + input_attachments, + input_specification, + log_level, + name, + role_arn, + ): + channel = self._channels[channel_id] + channel.cdi_input_specification = cdi_input_specification + channel.destinations = destinations + channel.encoder_settings = encoder_settings + channel.input_attachments = input_attachments + channel.input_specification = input_specification + channel.log_level = log_level + channel.name = name + channel.role_arn = role_arn + + channel._resolve_transient_states() + channel._previous_state = channel.state + channel.state = "UPDATING" + + return channel + + def create_input( + self, + destinations, + input_devices, + input_security_groups, + media_connect_flows, + name, + request_id, + role_arn, + sources, + tags, + type, + vpc, + ): + input_id = uuid4().hex + arn = "arn:aws:medialive:input:{}".format(input_id) + a_input = Input( + arn=arn, + input_id=input_id, + destinations=destinations, + input_devices=input_devices, + input_security_groups=input_security_groups, + media_connect_flows=media_connect_flows, + name=name, + role_arn=role_arn, + sources=sources, + tags=tags, + input_type=type, + state="CREATING", + ) + self._inputs[input_id] = a_input + return a_input + + def describe_input(self, input_id): + a_input = self._inputs[input_id] + a_input._resolve_transient_states() + return a_input.to_dict() + + def list_inputs(self, max_results, next_token): + inputs = list(self._inputs.values()) + if max_results is not None: + inputs = inputs[:max_results] + response_inputs = [i.to_dict() for i in inputs] + return response_inputs, next_token + + def delete_input(self, input_id): + a_input = self._inputs[input_id] + a_input.state = "DELETING" + return a_input.to_dict() + + def update_input( + self, + destinations, + input_devices, + input_id, + input_security_groups, + media_connect_flows, + name, + role_arn, + sources, + ): + a_input = self._inputs[input_id] + a_input.destinations = destinations + a_input.input_devices = input_devices + a_input.security_groups = input_security_groups + a_input.media_connect_flows = media_connect_flows + a_input.name = name + a_input.role_arn = role_arn + a_input.sources = sources + return a_input + + +medialive_backends = {} +for region in Session().get_available_regions("medialive"): + medialive_backends[region] = MediaLiveBackend() +for region in Session().get_available_regions("medialive", partition_name="aws-us-gov"): + medialive_backends[region] = MediaLiveBackend() +for region in Session().get_available_regions("medialive", partition_name="aws-cn"): + medialive_backends[region] = MediaLiveBackend() diff --git a/moto/medialive/responses.py b/moto/medialive/responses.py new file mode 100644 index 000000000..2c8568bb5 --- /dev/null +++ b/moto/medialive/responses.py @@ -0,0 +1,160 @@ +from __future__ import unicode_literals +from moto.core.responses import BaseResponse +from .models import medialive_backends +import json + + +class MediaLiveResponse(BaseResponse): + SERVICE_NAME = "medialive" + + @property + def medialive_backend(self): + return medialive_backends[self.region] + + def create_channel(self): + cdi_input_specification = self._get_param("cdiInputSpecification") + channel_class = self._get_param("channelClass") + destinations = self._get_param("destinations") + encoder_settings = self._get_param("encoderSettings") + input_attachments = self._get_param("inputAttachments") + input_specification = self._get_param("inputSpecification") + log_level = self._get_param("logLevel") + name = self._get_param("name") + request_id = self._get_param("requestId") + reserved = self._get_param("reserved") + role_arn = self._get_param("roleArn") + tags = self._get_param("tags") + channel = self.medialive_backend.create_channel( + cdi_input_specification=cdi_input_specification, + channel_class=channel_class, + destinations=destinations, + encoder_settings=encoder_settings, + input_attachments=input_attachments, + input_specification=input_specification, + log_level=log_level, + name=name, + request_id=request_id, + reserved=reserved, + role_arn=role_arn, + tags=tags, + ) + + return json.dumps( + dict(channel=channel.to_dict(exclude=["pipelinesRunningCount"])) + ) + + def list_channels(self): + max_results = self._get_int_param("maxResults") + next_token = self._get_param("nextToken") + channels, next_token = self.medialive_backend.list_channels( + max_results=max_results, next_token=next_token, + ) + + return json.dumps(dict(channels=channels, nextToken=next_token)) + + def describe_channel(self): + channel_id = self._get_param("channelId") + return json.dumps( + self.medialive_backend.describe_channel(channel_id=channel_id,) + ) + + def delete_channel(self): + channel_id = self._get_param("channelId") + return json.dumps(self.medialive_backend.delete_channel(channel_id=channel_id,)) + + def start_channel(self): + channel_id = self._get_param("channelId") + return json.dumps(self.medialive_backend.start_channel(channel_id=channel_id,)) + + def stop_channel(self): + channel_id = self._get_param("channelId") + return json.dumps(self.medialive_backend.stop_channel(channel_id=channel_id,)) + + def update_channel(self): + channel_id = self._get_param("channelId") + cdi_input_specification = self._get_param("cdiInputSpecification") + destinations = self._get_param("destinations") + encoder_settings = self._get_param("encoderSettings") + input_attachments = self._get_param("inputAttachments") + input_specification = self._get_param("inputSpecification") + log_level = self._get_param("logLevel") + name = self._get_param("name") + role_arn = self._get_param("roleArn") + channel = self.medialive_backend.update_channel( + channel_id=channel_id, + cdi_input_specification=cdi_input_specification, + destinations=destinations, + encoder_settings=encoder_settings, + input_attachments=input_attachments, + input_specification=input_specification, + log_level=log_level, + name=name, + role_arn=role_arn, + ) + return json.dumps(dict(channel=channel.to_dict())) + + def create_input(self): + destinations = self._get_param("destinations") + input_devices = self._get_param("inputDevices") + input_security_groups = self._get_param("inputSecurityGroups") + media_connect_flows = self._get_param("mediaConnectFlows") + name = self._get_param("name") + request_id = self._get_param("requestId") + role_arn = self._get_param("roleArn") + sources = self._get_param("sources") + tags = self._get_param("tags") + type = self._get_param("type") + vpc = self._get_param("vpc") + a_input = self.medialive_backend.create_input( + destinations=destinations, + input_devices=input_devices, + input_security_groups=input_security_groups, + media_connect_flows=media_connect_flows, + name=name, + request_id=request_id, + role_arn=role_arn, + sources=sources, + tags=tags, + type=type, + vpc=vpc, + ) + return json.dumps({"input": a_input.to_dict()}) + + def describe_input(self): + input_id = self._get_param("inputId") + return json.dumps(self.medialive_backend.describe_input(input_id=input_id,)) + + def list_inputs(self): + max_results = self._get_int_param("maxResults") + next_token = self._get_param("nextToken") + inputs, next_token = self.medialive_backend.list_inputs( + max_results=max_results, next_token=next_token, + ) + + return json.dumps(dict(inputs=inputs, nextToken=next_token)) + + def delete_input(self): + input_id = self._get_param("inputId") + self.medialive_backend.delete_input(input_id=input_id,) + return json.dumps({}) + + def update_input(self): + destinations = self._get_param("destinations") + input_devices = self._get_param("inputDevices") + input_id = self._get_param("inputId") + input_security_groups = self._get_param("inputSecurityGroups") + media_connect_flows = self._get_param("mediaConnectFlows") + name = self._get_param("name") + role_arn = self._get_param("roleArn") + sources = self._get_param("sources") + a_input = self.medialive_backend.update_input( + destinations=destinations, + input_devices=input_devices, + input_id=input_id, + input_security_groups=input_security_groups, + media_connect_flows=media_connect_flows, + name=name, + role_arn=role_arn, + sources=sources, + ) + return json.dumps(dict(input=a_input.to_dict())) diff --git a/moto/medialive/urls.py b/moto/medialive/urls.py new file mode 100644 index 000000000..c5183757d --- /dev/null +++ b/moto/medialive/urls.py @@ -0,0 +1,19 @@ +from __future__ import unicode_literals +from .responses import MediaLiveResponse + +url_bases = [ + "https?://medialive.(.+).amazonaws.com", +] + + +response = MediaLiveResponse() + + +url_paths = { + "{0}/prod/channels": response.dispatch, + "{0}/prod/channels/(?P[^/.]+)": response.dispatch, + "{0}/prod/channels/(?P[^/.]+)/start": response.dispatch, + "{0}/prod/channels/(?P[^/.]+)/stop": response.dispatch, + "{0}/prod/inputs": response.dispatch, + "{0}/prod/inputs/(?P[^/.]+)": response.dispatch, +} diff --git a/tests/test_medialive/test_medialive.py b/tests/test_medialive/test_medialive.py new file mode 100644 index 000000000..c61dbf0f9 --- /dev/null +++ b/tests/test_medialive/test_medialive.py @@ -0,0 +1,336 @@ +from __future__ import unicode_literals + +import boto3 +import sure # noqa +from moto import mock_medialive +from uuid import uuid4 + +from moto.core import ACCOUNT_ID + +region = "eu-west-1" + + +def _create_input_config(name, **kwargs): + role_arn = kwargs.get( + "role_arn", + "arn:aws:iam::{}:role/TestMediaLiveInputCreateRole".format(ACCOUNT_ID), + ) + input_type = kwargs.get("type", "RTP_PUSH") + request_id = kwargs.get("request_id", uuid4().hex) + destinations = kwargs.get("destinations", []) + input_devices = kwargs.get("input_devices", [{"Id": "1234-56"}]) + input_security_groups = ["123456"] + media_connect_flows = kwargs.get("media_connect_flows", [{"FlowArn": "flow:1"}]) + sources = kwargs.get( + "sources", + [ + { + "PasswordParam": "pwd431$%!", + "Url": "scheme://url:1234/", + "Username": "userX", + } + ], + ) + tags = kwargs.get("tags", {"Customer": "moto"}) + vpc_config = kwargs.get( + "vpc", {"SubnetIds": ["subnet-1"], "SecurityGroupIds": ["sg-0001"]} + ) + input_config = dict( + Name=name, + Destinations=destinations, + InputDevices=input_devices, + InputSecurityGroups=input_security_groups, + MediaConnectFlows=media_connect_flows, + RoleArn=role_arn, + RequestId=request_id, + Sources=sources, + Type=input_type, + Tags=tags, + Vpc=vpc_config, + ) + return input_config + + +def _create_channel_config(name, **kwargs): + role_arn = kwargs.get( + "role_arn", + "arn:aws:iam::{}:role/TestMediaLiveChannelCreateRole".format(ACCOUNT_ID), + ) + input_id = kwargs.get("input_id", "an-attachment-id") + input_settings = kwargs.get( + "input_settings", + [ + { + "InputId": input_id, + "InputSettings": { + "DenoiseFilter": "DISABLED", + "AudioSelectors": [ + {"Name": "EnglishLanguage", "SelectorSettings": {},} + ], + "InputFilter": "AUTO", + "DeblockFilter": "DISABLED", + "NetworkInputSettings": { + "ServerValidation": "CHECK_CRYPTOGRAPHY_AND_VALIDATE_NAME", + }, + "SourceEndBehavior": "CONTINUE", + "FilterStrength": 1, + }, + } + ], + ) + destinations = kwargs.get( + "destinations", [{"Id": "destination.1"}, {"Id": "destination.2"}] + ) + encoder_settings = kwargs.get( + "encoder_settings", + { + "VideoDescriptions": [], + "AudioDescriptions": [], + "OutputGroups": [], + "TimecodeConfig": {"Source": "a-source",}, + }, + ) + input_specification = kwargs.get("input_specification", {}) + log_level = kwargs.get("log_level", "INFO") + tags = kwargs.get("tags", {"Customer": "moto"}) + channel_config = dict( + Name=name, + RoleArn=role_arn, + InputAttachments=input_settings, + Destinations=destinations, + EncoderSettings=encoder_settings, + InputSpecification=input_specification, + RequestId=name, + LogLevel=log_level, + Tags=tags, + ) + return channel_config + + +@mock_medialive +def test_create_channel_succeeds(): + client = boto3.client("medialive", region_name=region) + channel_config = _create_channel_config("test channel 1") + + response = client.create_channel(**channel_config) + + response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + response["Channel"]["Arn"].should.equal( + "arn:aws:medialive:channel:{}".format(response["Channel"]["Id"]) + ) + response["Channel"]["Destinations"].should.equal(channel_config["Destinations"]) + response["Channel"]["EncoderSettings"].should.equal( + channel_config["EncoderSettings"] + ) + 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 +def test_list_channels_succeeds(): + client = boto3.client("medialive", region_name=region) + channel1_config = _create_channel_config("test channel 1", request_id="request-1") + channel2_config = _create_channel_config("test channel 2", request_id="request-2") + channel2_config["ChannelClass"] = "SINGLE_PIPELINE" + + client.create_channel(**channel1_config) + client.create_channel(**channel2_config) + + response = client.list_channels() + len(response["Channels"]).should.equal(2) + + response["Channels"][0]["Name"].should.equal("test channel 1") + response["Channels"][0]["ChannelClass"].should.equal("STANDARD") + response["Channels"][0]["PipelinesRunningCount"].should.equal(2) + + response["Channels"][1]["Name"].should.equal("test channel 2") + response["Channels"][1]["ChannelClass"].should.equal("SINGLE_PIPELINE") + response["Channels"][1]["PipelinesRunningCount"].should.equal(1) + + +@mock_medialive +def test_delete_channel_moves_channel_in_deleted_state(): + client = boto3.client("medialive", region_name=region) + channel_name = "test channel X" + channel_config = _create_channel_config(channel_name) + + create_response = client.create_channel(**channel_config) + delete_response = client.delete_channel(ChannelId=create_response["Channel"]["Id"]) + + delete_response["Name"].should.equal(channel_name) + delete_response["State"].should.equal("DELETING") + + +@mock_medialive +def test_describe_channel_succeeds(): + client = boto3.client("medialive", region_name=region) + channel_name = "test channel X" + channel_config = _create_channel_config(channel_name) + + create_response = client.create_channel(**channel_config) + describe_response = client.describe_channel( + ChannelId=create_response["Channel"]["Id"] + ) + + describe_response["Arn"].should.equal( + "arn:aws:medialive:channel:{}".format(describe_response["Id"]) + ) + describe_response["Destinations"].should.equal(channel_config["Destinations"]) + describe_response["EncoderSettings"].should.equal(channel_config["EncoderSettings"]) + describe_response["InputAttachments"].should.equal( + channel_config["InputAttachments"] + ) + describe_response["Name"].should.equal(channel_name) + describe_response["State"].should.equal("IDLE") + describe_response["Tags"]["Customer"].should.equal("moto") + + +@mock_medialive +def test_start_channel_succeeds(): + client = boto3.client("medialive", region_name=region) + channel_name = "testchan1" + channel_config = _create_channel_config(channel_name) + + create_response = client.create_channel(**channel_config) + start_response = client.start_channel(ChannelId=create_response["Channel"]["Id"]) + start_response["Name"].should.equal(channel_name) + start_response["State"].should.equal("STARTING") + + describe_response = client.describe_channel( + ChannelId=create_response["Channel"]["Id"] + ) + describe_response["State"].should.equal("RUNNING") + + +@mock_medialive +def test_stop_channel_succeeds(): + client = boto3.client("medialive", region_name=region) + channel_name = "testchan2" + channel_config = _create_channel_config(channel_name) + + create_response = client.create_channel(**channel_config) + channel_id = create_response["Channel"]["Id"] + assert len(channel_id) > 1 + start_response = client.start_channel(ChannelId=channel_id) + stop_response = client.stop_channel(ChannelId=channel_id) + stop_response["Name"].should.equal(channel_name) + stop_response["State"].should.equal("STOPPING") + + describe_response = client.describe_channel( + ChannelId=create_response["Channel"]["Id"] + ) + describe_response["State"].should.equal("IDLE") + + +@mock_medialive +def test_update_channel_succeeds(): + client = boto3.client("medialive", region_name=region) + channel_name = "Original Channel" + channel_config = _create_channel_config(channel_name) + + create_response = client.create_channel(**channel_config) + channel_id = create_response["Channel"]["Id"] + assert len(channel_id) > 1 + + update_response = client.update_channel( + ChannelId=channel_id, Name="Updated Channel", + ) + update_response["Channel"]["State"].should.equal("UPDATING") + update_response["Channel"]["Name"].should.equal("Updated Channel") + + describe_response = client.describe_channel(ChannelId=channel_id,) + describe_response["State"].should.equal("IDLE") + describe_response["Name"].should.equal("Updated Channel") + + +@mock_medialive +def test_create_input_succeeds(): + client = boto3.client("medialive", region_name=region) + input_name = "Input One" + input_config = _create_input_config(input_name) + + create_response = client.create_input(**input_config) + create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + r_input = create_response["Input"] + input_id = r_input["Id"] + assert len(input_id) > 1 + r_input["Arn"].should.equal("arn:aws:medialive:input:{}".format(r_input["Id"])) + r_input["Name"].should.equal(input_name) + r_input["AttachedChannels"].should.equal([]) + r_input["Destinations"].should.equal(input_config["Destinations"]) + r_input["InputClass"].should.equal("STANDARD") + r_input["InputDevices"].should.equal(input_config["InputDevices"]) + r_input["InputSourceType"].should.equal("STATIC") + r_input["MediaConnectFlows"].should.equal(input_config["MediaConnectFlows"]) + r_input["RoleArn"].should.equal(input_config["RoleArn"]) + r_input["SecurityGroups"].should.equal([]) + r_input["Sources"].should.equal(input_config["Sources"]) + r_input["State"].should.equal("CREATING") + r_input["Tags"].should.equal(input_config["Tags"]) + r_input["Type"].should.equal(input_config["Type"]) + + +@mock_medialive +def test_describe_input_succeeds(): + client = boto3.client("medialive", region_name=region) + input_name = "Input Two" + input_config = _create_input_config(input_name) + + create_response = client.create_input(**input_config) + create_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + create_response["Input"]["State"].should.equal("CREATING") + + describe_response = client.describe_input(InputId=create_response["Input"]["Id"]) + describe_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + describe_response["Name"].should.equal(input_name) + describe_response["State"].should.equal("DETACHED") + describe_response["MediaConnectFlows"].should.equal( + input_config["MediaConnectFlows"] + ) + + +@mock_medialive +def test_list_inputs_succeeds(): + client = boto3.client("medialive", region_name=region) + input_config1 = _create_input_config("Input One") + create_response = client.create_input(**input_config1) + input_config2 = _create_input_config("Input Two") + create_response = client.create_input(**input_config2) + + response = client.list_inputs() + len(response["Inputs"]).should.equal(2) + + response["Inputs"][0]["Name"].should.equal("Input One") + response["Inputs"][1]["Name"].should.equal("Input Two") + + +@mock_medialive +def test_delete_input_moves_input_in_deleted_state(): + client = boto3.client("medialive", region_name=region) + input_name = "test input X" + input_config = _create_input_config(input_name) + + create_response = client.create_input(**input_config) + delete_response = client.delete_input(InputId=create_response["Input"]["Id"]) + delete_response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200) + + describe_response = client.describe_input(InputId=create_response["Input"]["Id"]) + describe_response["Name"].should.equal(input_name) + describe_response["State"].should.equal("DELETED") + + +@mock_medialive +def test_update_input_succeeds(): + client = boto3.client("medialive", region_name=region) + input_name = "test input X" + input_config = _create_input_config(input_name) + + create_response = client.create_input(**input_config) + update_response = client.update_input( + InputId=create_response["Input"]["Id"], Name="test input U", + ) + update_response["Input"]["Name"].should.equal("test input U") diff --git a/tests/test_medialive/test_server.py b/tests/test_medialive/test_server.py new file mode 100644 index 000000000..df51acf8e --- /dev/null +++ b/tests/test_medialive/test_server.py @@ -0,0 +1,32 @@ +from __future__ import unicode_literals + +import sure # noqa + +import moto.server as server +from moto import mock_medialive + +""" +Test the different server responses +""" + + +@mock_medialive +def test_medialive_list_channels(): + backend = server.create_backend_app("medialive") + test_client = backend.test_client() + + res = test_client.get("/prod/channels") + + result = res.data.decode("utf-8") + result.should.contain('"channels": []') + + +@mock_medialive +def test_medialive_list_inputs(): + backend = server.create_backend_app("medialive") + test_client = backend.test_client() + + res = test_client.get("/prod/inputs") + + result = res.data.decode("utf-8") + result.should.contain('"inputs": []')