IOT list_execs_for_thing: fix pagination (#4537)

moto/iot/models.py

@@ -7,10 +7,12 @@ import uuid
 from collections import OrderedDict
 from datetime import datetime
 
-from boto3 import Session
+from .utils import PAGINATION_MODEL
 
+from boto3 import Session
 from moto.core import BaseBackend, BaseModel
 from moto.utilities.utils import random_string
+from moto.utilities.paginator import paginate
 from .exceptions import (
     CertificateStateException,
     DeleteConflictException,
@@ -1338,9 +1340,8 @@ class IoTBackend(BaseBackend):
 
         return job_executions, next_token
 
-    def list_job_executions_for_thing(
-        self, thing_name, status, max_results, next_token
-    ):
+    @paginate(PAGINATION_MODEL)
+    def list_job_executions_for_thing(self, thing_name, status):
         job_executions = [
             self.job_executions[je].to_dict()
             for je in self.job_executions
@@ -1355,20 +1356,7 @@ class IoTBackend(BaseBackend):
                 )
             )
 
-        token = next_token
-        if token is None:
-            job_executions = job_executions[0:max_results]
-            next_token = str(max_results) if len(job_executions) > max_results else None
-        else:
-            token = int(token)
-            job_executions = job_executions[token : token + max_results]
-            next_token = (
-                str(token + max_results)
-                if len(job_executions) > token + max_results
-                else None
-            )
-
-        return job_executions, next_token
+        return job_executions
 
     def list_topic_rules(self):
         return [r.to_dict() for r in self.rules.values()]

moto/iot/utils.py

@@ -0,0 +1,8 @@
+PAGINATION_MODEL = {
+    "list_job_executions_for_thing": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "page_ending_range_keys": ["jobId"],
+    }
+}

tests/test_iot/test_iot.py

@@ -2061,6 +2061,39 @@ def test_list_job_executions_for_thing():
     )
 
 
+@mock_iot
+def test_list_job_executions_for_thing_paginated():
+    client = boto3.client("iot", region_name="eu-west-1")
+    name = "my-thing"
+    thing = client.create_thing(thingName=name)
+
+    for idx in range(0, 10):
+        client.create_job(
+            jobId=f"TestJob_{idx}",
+            targets=[thing["thingArn"]],
+            document=json.dumps({"field": "value"}),
+        )
+
+    res = client.list_job_executions_for_thing(thingName=name, maxResults=2)
+    executions = res["executionSummaries"]
+    executions.should.have.length_of(2)
+    res.should.have.key("nextToken")
+
+    res = client.list_job_executions_for_thing(
+        thingName=name, maxResults=1, nextToken=res["nextToken"]
+    )
+    executions = res["executionSummaries"]
+    executions.should.have.length_of(1)
+    res.should.have.key("nextToken")
+
+    res = client.list_job_executions_for_thing(
+        thingName=name, nextToken=res["nextToken"]
+    )
+    executions = res["executionSummaries"]
+    executions.should.have.length_of(7)
+    res.shouldnt.have.key("nextToken")
+
+
 class TestTopicRules:
     name = "my-rule"
     payload = {