Add kinesisvideo archived media (#3280)

* add get_hls_streaming_session_url

* add get_dash_streaming_session_url

* add get_clip

* add test_server for kinesisvideo archived media

* fix for lint

* fix for lint

* avoid testing kinesisvideoarchivedmedia with TEST_SERVER_MODE=true

Makefile

@@ -3,7 +3,11 @@ SHELL := /bin/bash
 ifeq ($(TEST_SERVER_MODE), true)
 	# exclude test_iot and test_iotdata for now
 	# because authentication of iot is very complicated
-	TEST_EXCLUDE :=  --exclude='test_iot.*'
+
+	# exclude test_kinesisvideoarchivedmedia
+	# because testing with moto_server is difficult with data-endpoint
+
+	TEST_EXCLUDE :=  --exclude='test_iot.*' --exclude="test_kinesisvideoarchivedmedia.*"
 else
 	TEST_EXCLUDE :=
 endif

moto/__init__.py

@@ -114,6 +114,9 @@ XRaySegment = lazy_load(".xray", "XRaySegment")
 mock_xray = lazy_load(".xray", "mock_xray")
 mock_xray_client = lazy_load(".xray", "mock_xray_client")
 mock_kinesisvideo = lazy_load(".kinesisvideo", "mock_kinesisvideo")
+mock_kinesisvideoarchivedmedia = lazy_load(
+    ".kinesisvideoarchivedmedia", "mock_kinesisvideoarchivedmedia"
+)
 
 # import logging
 # logging.getLogger('boto').setLevel(logging.CRITICAL)

moto/backends.py

@@ -70,6 +70,10 @@ BACKENDS = {
     "swf": ("swf", "swf_backends"),
     "xray": ("xray", "xray_backends"),
     "kinesisvideo": ("kinesisvideo", "kinesisvideo_backends"),
+    "kinesis-video-archived-media": (
+        "kinesisvideoarchivedmedia",
+        "kinesisvideoarchivedmedia_backends",
+    ),
 }
 
 

moto/kinesisvideo/responses.py

@@ -63,8 +63,3 @@ class KinesisVideoResponse(BaseResponse):
             stream_name=stream_name, stream_arn=stream_arn, api_name=api_name,
         )
         return json.dumps(dict(DataEndpoint=data_endpoint))
-
-    # add methods from here
-
-
-# add templates from here

moto/kinesisvideoarchivedmedia/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import unicode_literals
+from .models import kinesisvideoarchivedmedia_backends
+from ..core.models import base_decorator
+
+kinesisvideoarchivedmedia_backend = kinesisvideoarchivedmedia_backends["us-east-1"]
+mock_kinesisvideoarchivedmedia = base_decorator(kinesisvideoarchivedmedia_backends)

moto/kinesisvideoarchivedmedia/exceptions.py

@@ -0,0 +1,3 @@
+from __future__ import unicode_literals
+
+# Not implemented exceptions for now

moto/kinesisvideoarchivedmedia/models.py

@@ -0,0 +1,88 @@
+from __future__ import unicode_literals
+from boto3 import Session
+from moto.core import BaseBackend
+from moto.kinesisvideo import kinesisvideo_backends
+from moto.sts.utils import random_session_token
+
+
+class KinesisVideoArchivedMediaBackend(BaseBackend):
+    def __init__(self, region_name=None):
+        super(KinesisVideoArchivedMediaBackend, self).__init__()
+        self.region_name = region_name
+
+    def reset(self):
+        region_name = self.region_name
+        self.__dict__ = {}
+        self.__init__(region_name)
+
+    def _get_streaming_url(self, stream_name, stream_arn, api_name):
+        stream = kinesisvideo_backends[self.region_name]._get_stream(
+            stream_name, stream_arn
+        )
+        data_endpoint = stream.get_data_endpoint(api_name)
+        session_token = random_session_token()
+        api_to_relative_path = {
+            "GET_HLS_STREAMING_SESSION_URL": "/hls/v1/getHLSMasterPlaylist.m3u8",
+            "GET_DASH_STREAMING_SESSION_URL": "/dash/v1/getDASHManifest.mpd",
+        }
+        relative_path = api_to_relative_path[api_name]
+        url = "{}{}?SessionToken={}".format(data_endpoint, relative_path, session_token)
+        return url
+
+    def get_hls_streaming_session_url(
+        self,
+        stream_name,
+        stream_arn,
+        playback_mode,
+        hls_fragment_selector,
+        container_format,
+        discontinuity_mode,
+        display_fragment_timestamp,
+        expires,
+        max_media_playlist_fragment_results,
+    ):
+        # Ignore option paramters as the format of hls_url does't depends on them
+        api_name = "GET_HLS_STREAMING_SESSION_URL"
+        url = self._get_streaming_url(stream_name, stream_arn, api_name)
+        return url
+
+    def get_dash_streaming_session_url(
+        self,
+        stream_name,
+        stream_arn,
+        playback_mode,
+        display_fragment_timestamp,
+        display_fragment_number,
+        dash_fragment_selector,
+        expires,
+        max_manifest_fragment_results,
+    ):
+        # Ignore option paramters as the format of hls_url does't depends on them
+        api_name = "GET_DASH_STREAMING_SESSION_URL"
+        url = self._get_streaming_url(stream_name, stream_arn, api_name)
+        return url
+
+    def get_clip(self, stream_name, stream_arn, clip_fragment_selector):
+        kinesisvideo_backends[self.region_name]._get_stream(stream_name, stream_arn)
+        content_type = "video/mp4"  # Fixed content_type as it depends on input stream
+        payload = b"sample-mp4-video"
+        return content_type, payload
+
+
+kinesisvideoarchivedmedia_backends = {}
+for region in Session().get_available_regions("kinesis-video-archived-media"):
+    kinesisvideoarchivedmedia_backends[region] = KinesisVideoArchivedMediaBackend(
+        region
+    )
+for region in Session().get_available_regions(
+    "kinesis-video-archived-media", partition_name="aws-us-gov"
+):
+    kinesisvideoarchivedmedia_backends[region] = KinesisVideoArchivedMediaBackend(
+        region
+    )
+for region in Session().get_available_regions(
+    "kinesis-video-archived-media", partition_name="aws-cn"
+):
+    kinesisvideoarchivedmedia_backends[region] = KinesisVideoArchivedMediaBackend(
+        region
+    )

moto/kinesisvideoarchivedmedia/responses.py

@@ -0,0 +1,70 @@
+from __future__ import unicode_literals
+from moto.core.responses import BaseResponse
+from .models import kinesisvideoarchivedmedia_backends
+import json
+
+
+class KinesisVideoArchivedMediaResponse(BaseResponse):
+    SERVICE_NAME = "kinesis-video-archived-media"
+
+    @property
+    def kinesisvideoarchivedmedia_backend(self):
+        return kinesisvideoarchivedmedia_backends[self.region]
+
+    def get_hls_streaming_session_url(self):
+        stream_name = self._get_param("StreamName")
+        stream_arn = self._get_param("StreamARN")
+        playback_mode = self._get_param("PlaybackMode")
+        hls_fragment_selector = self._get_param("HLSFragmentSelector")
+        container_format = self._get_param("ContainerFormat")
+        discontinuity_mode = self._get_param("DiscontinuityMode")
+        display_fragment_timestamp = self._get_param("DisplayFragmentTimestamp")
+        expires = self._get_int_param("Expires")
+        max_media_playlist_fragment_results = self._get_param(
+            "MaxMediaPlaylistFragmentResults"
+        )
+        hls_streaming_session_url = self.kinesisvideoarchivedmedia_backend.get_hls_streaming_session_url(
+            stream_name=stream_name,
+            stream_arn=stream_arn,
+            playback_mode=playback_mode,
+            hls_fragment_selector=hls_fragment_selector,
+            container_format=container_format,
+            discontinuity_mode=discontinuity_mode,
+            display_fragment_timestamp=display_fragment_timestamp,
+            expires=expires,
+            max_media_playlist_fragment_results=max_media_playlist_fragment_results,
+        )
+        return json.dumps(dict(HLSStreamingSessionURL=hls_streaming_session_url))
+
+    def get_dash_streaming_session_url(self):
+        stream_name = self._get_param("StreamName")
+        stream_arn = self._get_param("StreamARN")
+        playback_mode = self._get_param("PlaybackMode")
+        display_fragment_timestamp = self._get_param("DisplayFragmentTimestamp")
+        display_fragment_number = self._get_param("DisplayFragmentNumber")
+        dash_fragment_selector = self._get_param("DASHFragmentSelector")
+        expires = self._get_int_param("Expires")
+        max_manifest_fragment_results = self._get_param("MaxManifestFragmentResults")
+        dash_streaming_session_url = self.kinesisvideoarchivedmedia_backend.get_dash_streaming_session_url(
+            stream_name=stream_name,
+            stream_arn=stream_arn,
+            playback_mode=playback_mode,
+            display_fragment_timestamp=display_fragment_timestamp,
+            display_fragment_number=display_fragment_number,
+            dash_fragment_selector=dash_fragment_selector,
+            expires=expires,
+            max_manifest_fragment_results=max_manifest_fragment_results,
+        )
+        return json.dumps(dict(DASHStreamingSessionURL=dash_streaming_session_url))
+
+    def get_clip(self):
+        stream_name = self._get_param("StreamName")
+        stream_arn = self._get_param("StreamARN")
+        clip_fragment_selector = self._get_param("ClipFragmentSelector")
+        content_type, payload = self.kinesisvideoarchivedmedia_backend.get_clip(
+            stream_name=stream_name,
+            stream_arn=stream_arn,
+            clip_fragment_selector=clip_fragment_selector,
+        )
+        new_headers = {"Content-Type": content_type}
+        return payload, new_headers

moto/kinesisvideoarchivedmedia/urls.py

@@ -0,0 +1,14 @@
+from __future__ import unicode_literals
+from .responses import KinesisVideoArchivedMediaResponse
+
+url_bases = [
+    r"https?://.*\.kinesisvideo.(.+).amazonaws.com",
+]
+
+
+response = KinesisVideoArchivedMediaResponse()
+
+
+url_paths = {
+    "{0}/.*$": response.dispatch,
+}

tests/test_kinesisvideoarchivedmedia/test_kinesisvideoarchivedmedia.py

@@ -0,0 +1,86 @@
+from __future__ import unicode_literals
+
+import boto3
+import sure  # noqa
+from moto import mock_kinesisvideoarchivedmedia
+from moto import mock_kinesisvideo
+from datetime import datetime, timedelta
+
+
+@mock_kinesisvideo
+@mock_kinesisvideoarchivedmedia
+def test_get_hls_streaming_session_url():
+    region_name = "ap-northeast-1"
+    kvs_client = boto3.client("kinesisvideo", region_name=region_name)
+    stream_name = "my-stream"
+    kvs_client.create_stream(StreamName=stream_name)
+
+    api_name = "GET_HLS_STREAMING_SESSION_URL"
+    res = kvs_client.get_data_endpoint(StreamName=stream_name, APIName=api_name)
+    data_endpoint = res["DataEndpoint"]
+
+    client = boto3.client(
+        "kinesis-video-archived-media",
+        region_name=region_name,
+        endpoint_url=data_endpoint,
+    )
+    res = client.get_hls_streaming_session_url(StreamName=stream_name,)
+    reg_exp = "^{}/hls/v1/getHLSMasterPlaylist.m3u8\?SessionToken\=.+$".format(
+        data_endpoint
+    )
+    res.should.have.key("HLSStreamingSessionURL").which.should.match(reg_exp)
+
+
+@mock_kinesisvideo
+@mock_kinesisvideoarchivedmedia
+def test_get_dash_streaming_session_url():
+    region_name = "ap-northeast-1"
+    kvs_client = boto3.client("kinesisvideo", region_name=region_name)
+    stream_name = "my-stream"
+    kvs_client.create_stream(StreamName=stream_name)
+
+    api_name = "GET_DASH_STREAMING_SESSION_URL"
+    res = kvs_client.get_data_endpoint(StreamName=stream_name, APIName=api_name)
+    data_endpoint = res["DataEndpoint"]
+
+    client = boto3.client(
+        "kinesis-video-archived-media",
+        region_name=region_name,
+        endpoint_url=data_endpoint,
+    )
+    res = client.get_dash_streaming_session_url(StreamName=stream_name,)
+    reg_exp = "^{}/dash/v1/getDASHManifest.mpd\?SessionToken\=.+$".format(data_endpoint)
+    res.should.have.key("DASHStreamingSessionURL").which.should.match(reg_exp)
+
+
+@mock_kinesisvideo
+@mock_kinesisvideoarchivedmedia
+def test_get_clip():
+    region_name = "ap-northeast-1"
+    kvs_client = boto3.client("kinesisvideo", region_name=region_name)
+    stream_name = "my-stream"
+    kvs_client.create_stream(StreamName=stream_name)
+
+    api_name = "GET_DASH_STREAMING_SESSION_URL"
+    res = kvs_client.get_data_endpoint(StreamName=stream_name, APIName=api_name)
+    data_endpoint = res["DataEndpoint"]
+
+    client = boto3.client(
+        "kinesis-video-archived-media",
+        region_name=region_name,
+        endpoint_url=data_endpoint,
+    )
+    end_timestamp = datetime.utcnow() - timedelta(hours=1)
+    start_timestamp = end_timestamp - timedelta(minutes=5)
+    res = client.get_clip(
+        StreamName=stream_name,
+        ClipFragmentSelector={
+            "FragmentSelectorType": "PRODUCER_TIMESTAMP",
+            "TimestampRange": {
+                "StartTimestamp": start_timestamp,
+                "EndTimestamp": end_timestamp,
+            },
+        },
+    )
+    res.should.have.key("ContentType").which.should.match("video/mp4")
+    res.should.have.key("Payload")

tests/test_kinesisvideoarchivedmedia/test_server.py

@@ -0,0 +1,19 @@
+from __future__ import unicode_literals
+
+import sure  # noqa
+
+import moto.server as server
+from moto import mock_kinesisvideoarchivedmedia
+
+"""
+Test the different server responses
+"""
+
+
+@mock_kinesisvideoarchivedmedia
+def test_kinesisvideoarchivedmedia_server_is_up():
+    backend = server.create_backend_app("kinesis-video-archived-media")
+    test_client = backend.test_client()
+    res = test_client.post("/getHLSStreamingSessionURL")
+    # Just checking server is up
+    res.status_code.should.equal(404)