MediaLive: list_channels()/list_inputs() now support pagination

docs/docs/services/medialive.rst

@@ -86,10 +86,6 @@ medialive
 - [ ] get_signal_map
 - [ ] list_channel_placement_groups
 - [X] list_channels
-
-        Pagination is not yet implemented
-
-
 - [ ] list_cloud_watch_alarm_template_groups
 - [ ] list_cloud_watch_alarm_templates
 - [ ] list_clusters
@@ -99,10 +95,6 @@ medialive
 - [ ] list_input_devices
 - [ ] list_input_security_groups
 - [X] list_inputs
-
-        Pagination is not yet implemented
-
-
 - [ ] list_multiplex_programs
 - [ ] list_multiplexes
 - [ ] list_networks

moto/medialive/models.py

@@ -4,8 +4,24 @@
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.moto_api._internal import mock_random
+from moto.utilities.paginator import paginate
 from moto.utilities.utils import get_partition
 
+PAGINATION_MODEL = {
+    "list_channels": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "arn",
+    },
+    "list_inputs": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "arn",
+    },
+}
+
 
 class Input(BaseModel):
     def __init__(self, **kwargs: Any):
@@ -157,13 +173,9 @@ def create_channel(
         self._channels[channel_id] = channel
         return channel
 
-    def list_channels(self, max_results: Optional[int]) -> List[Dict[str, Any]]:
-        """
-        Pagination is not yet implemented
-        """
+    @paginate(pagination_model=PAGINATION_MODEL)  # type: ignore[misc]
+    def list_channels(self) -> List[Dict[str, Any]]:
         channels = list(self._channels.values())
-        if max_results is not None:
-            channels = channels[:max_results]
         return [
             c.to_dict(exclude=["encoderSettings", "pipelineDetails"]) for c in channels
         ]
@@ -255,13 +267,9 @@ def describe_input(self, input_id: str) -> Input:
         a_input._resolve_transient_states()
         return a_input
 
-    def list_inputs(self, max_results: Optional[int]) -> List[Dict[str, Any]]:
-        """
-        Pagination is not yet implemented
-        """
+    @paginate(PAGINATION_MODEL)  # type: ignore[misc]
+    def list_inputs(self) -> List[Dict[str, Any]]:
         inputs = list(self._inputs.values())
-        if max_results is not None:
-            inputs = inputs[:max_results]
         return [i.to_dict() for i in inputs]
 
     def delete_input(self, input_id: str) -> None:

moto/medialive/responses.py

@@ -43,9 +43,12 @@ def create_channel(self) -> str:
 
     def list_channels(self) -> str:
         max_results = self._get_int_param("maxResults")
-        channels = self.medialive_backend.list_channels(max_results=max_results)
+        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=None))
+        return json.dumps(dict(channels=channels, nextToken=next_token))
 
     def describe_channel(self) -> str:
         channel_id = self._get_param("channelId")
@@ -120,9 +123,12 @@ def describe_input(self) -> str:
 
     def list_inputs(self) -> str:
         max_results = self._get_int_param("maxResults")
-        inputs = self.medialive_backend.list_inputs(max_results=max_results)
+        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=None))
+        return json.dumps(dict(inputs=inputs, nextToken=next_token))
 
     def delete_input(self) -> str:
         input_id = self._get_param("inputId")

tests/test_medialive/test_medialive.py

@@ -126,24 +126,41 @@ def test_create_channel_succeeds():
 @mock_aws
 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"
+    channel_configs = [
+        _create_channel_config(f"test {idx}", request_id=f"req-{idx}")
+        for idx in range(10)
+    ]
+    channel_configs[1]["ChannelClass"] = "SINGLE_PIPELINE"
 
-    client.create_channel(**channel1_config)
-    client.create_channel(**channel2_config)
+    for config in channel_configs:
+        client.create_channel(**config)
 
     response = client.list_channels()
-    assert len(response["Channels"]) == 2
+    assert len(response["Channels"]) == 10
 
-    assert response["Channels"][0]["Name"] == "test channel 1"
+    assert response["Channels"][0]["Name"] == "test 0"
     assert response["Channels"][0]["ChannelClass"] == "STANDARD"
     assert response["Channels"][0]["PipelinesRunningCount"] == 2
 
-    assert response["Channels"][1]["Name"] == "test channel 2"
+    assert response["Channels"][1]["Name"] == "test 1"
     assert response["Channels"][1]["ChannelClass"] == "SINGLE_PIPELINE"
     assert response["Channels"][1]["PipelinesRunningCount"] == 1
 
+    page1 = client.list_channels(MaxResults=2)
+    assert len(page1["Channels"]) == 2
+    channel_names = [c["Name"] for c in page1["Channels"]]
+    assert ["test 0", "test 1"] == sorted(channel_names)
+
+    page2 = client.list_channels(MaxResults=5, NextToken=page1["NextToken"])
+    assert len(page2["Channels"]) == 5
+    channel_names = [c["Name"] for c in page2["Channels"]]
+    assert ["test 2", "test 3", "test 4", "test 5", "test 6"] == sorted(channel_names)
+
+    page3 = client.list_channels(NextToken=page2["NextToken"])
+    assert len(page3["Channels"]) == 3
+    channel_names = [c["Name"] for c in page3["Channels"]]
+    assert ["test 7", "test 8", "test 9"] == sorted(channel_names)
+
 
 @mock_aws
 def test_delete_channel_moves_channel_in_deleted_state():
@@ -272,16 +289,27 @@ def test_describe_input_succeeds():
 @mock_aws
 def test_list_inputs_succeeds():
     client = boto3.client("medialive", region_name=region)
-    input_config1 = _create_input_config("Input One")
-    client.create_input(**input_config1)
-    input_config2 = _create_input_config("Input Two")
-    client.create_input(**input_config2)
+    configs = [_create_input_config(f"Input {idx}") for idx in range(10)]
+    for config in configs:
+        client.create_input(**config)
 
     inputs = client.list_inputs()["Inputs"]
-    assert len(inputs) == 2
-
-    assert inputs[0]["Name"] == "Input One"
-    assert inputs[1]["Name"] == "Input Two"
+    assert len(inputs) == 10
+
+    page1 = client.list_inputs(MaxResults=2)
+    assert len(page1["Inputs"]) == 2
+    names = [i["Name"] for i in page1["Inputs"]]
+    assert sorted(names) == ["Input 0", "Input 1"]
+
+    page2 = client.list_inputs(MaxResults=5, NextToken=page1["NextToken"])
+    assert len(page2["Inputs"]) == 5
+    names = [i["Name"] for i in page2["Inputs"]]
+    assert sorted(names) == ["Input 2", "Input 3", "Input 4", "Input 5", "Input 6"]
+
+    page3 = client.list_inputs(NextToken=page2["NextToken"])
+    assert len(page3["Inputs"]) == 3
+    names = [i["Name"] for i in page3["Inputs"]]
+    assert sorted(names) == ["Input 7", "Input 8", "Input 9"]
 
 
 @mock_aws