Send limit parameter in execute_list server requests

airflow-ctl/src/airflowctl/api/operations.py

@@ -164,7 +164,7 @@ def execute_list(
         limit: int = 50,
         params: dict | None = None,
     ) -> T | ServerResponseError:
-        shared_params = {**(params or {})}
+        shared_params = {"limit": limit, **(params or {})}
         self.response = self.client.get(path, params=shared_params)
         first_pass = data_model.model_validate_json(self.response.content)
         total_entries = first_pass.total_entries  # type: ignore[attr-defined]

airflow-ctl/tests/airflow_ctl/api/test_operations.py

@@ -194,6 +194,38 @@ def test_execute_list(self, total_entries, limit, expected_response):
 
         assert expected_response == response
 
+    def test_execute_list_sends_limit_to_server(self):
+        """``limit`` must be included in request params so the server returns
+        the expected page size.  Without it the server uses its own default
+        (e.g. 100) which causes duplicate entries when ``limit`` differs."""
+        mock_client = Mock()
+        mock_client.get.return_value = Mock(
+            content=json.dumps({"hellos": [{"name": "hello"}] * 3, "total_entries": 3})
+        )
+        base_operation = BaseOperations(client=mock_client)
+
+        base_operation.execute_list(path="hello", data_model=HelloCollectionResponse, limit=50)
+
+        call_params = mock_client.get.call_args_list[0]
+        assert call_params.kwargs["params"]["limit"] == 50
+
+    def test_execute_list_sends_limit_on_subsequent_pages(self):
+        """Every paginated request must include ``limit`` so that offset
+        arithmetic stays consistent with the actual page size returned."""
+        mock_client = Mock()
+        mock_client.get.side_effect = [
+            Mock(content=json.dumps({"hellos": [{"name": "a"}, {"name": "b"}], "total_entries": 3})),
+            Mock(content=json.dumps({"hellos": [{"name": "c"}], "total_entries": 3})),
+        ]
+        base_operation = BaseOperations(client=mock_client)
+
+        response = base_operation.execute_list(path="hello", data_model=HelloCollectionResponse, limit=2)
+
+        assert len(response.hellos) == 3
+        # Verify limit is sent on both the first and second request
+        for call in mock_client.get.call_args_list:
+            assert call.kwargs["params"]["limit"] == 2
+
 
 class TestAssetsOperations:
     asset_id: int = 1