fix autopagination loop on non paginated api endpoint

Author: iam404

src/pytfe/resources/_base.py

@@ -7,6 +7,7 @@
 from typing import Any
 
 from .._http import HTTPTransport
+from .._logging import logger
 
 
 def _to_int(value: Any) -> int | None:
@@ -26,9 +27,26 @@ def __init__(self, t: HTTPTransport) -> None:
         self.t = t
 
     def _list(
-        self, path: str, *, params: dict | None = None
+        self, path: str, *, params: dict | None = None, paginated: bool = True
     ) -> Iterator[dict[str, Any]]:
         base_params = dict(params or {})
+
+        # Some TFE endpoints are not paginated: they return the full collection
+        # in a single response, ignore page[number]/page[size], and emit no
+        # meta.pagination block (e.g. workspace /vars and /all-vars). Callers
+        # opt those out so we issue exactly one request and never loop trying to
+        # fetch a "next" page that re-returns the same set
+        # (see hashicorp/python-tfe#181).
+        if not paginated:
+            r = self.t.request("GET", path, params=base_params)
+            json_response = r.json()
+            if not isinstance(json_response, dict):
+                json_response = {}
+            data = json_response.get("data", [])
+            if isinstance(data, list):
+                yield from data
+            return
+
         page = int(base_params.get("page[number]", 1))
         while True:
             p = dict(base_params)
@@ -81,8 +99,18 @@ def _list(
                 # Metadata present and indicates no next page.
                 break
 
-            # Fallback for endpoints that do not return pagination metadata.
-            page_size = int(p["page[size]"])
-            if len(data) < page_size:
-                break
-            page += 1
+            # No pagination metadata. Genuine TFE list endpoints always include
+            # meta.pagination, so a response without it is a non-paginated
+            # collection returned in full. Treat it as a single complete page
+            # and stop. Re-requesting would loop forever when the endpoint
+            # ignores page[number]/page[size] and re-returns the same full set
+            # (the root cause of hashicorp/python-tfe#181).
+            if len(data) >= int(p["page[size]"]):
+                logger.debug(
+                    "List endpoint %s returned a full page (%d rows) without "
+                    "pagination metadata; treating it as a single, "
+                    "non-paginated response.",
+                    path,
+                    len(data),
+                )
+            break

src/pytfe/resources/variable.py

@@ -30,13 +30,16 @@ def list(
         if not valid_string_id(workspace_id):
             raise ValueError(ERR_INVALID_WORKSPACE_ID)
 
+        # The /vars endpoint is not paginated: it returns every variable in a
+        # single response and ignores page[number]/page[size]. Opt out of the
+        # pagination loop so we issue exactly one request (python-tfe#181).
         path = f"/api/v2/workspaces/{workspace_id}/vars"
         params: dict[str, Any] = {}
         if options:
             # Add any options if needed in the future
             pass
 
-        for item in self._list(path, params=params):
+        for item in self._list(path, params=params, paginated=False):
             attr = item.get("attributes", {}) or {}
             var_id = item.get("id", "")
             variable_data = dict(attr)
@@ -50,13 +53,14 @@ def list_all(
         if not valid_string_id(workspace_id):
             raise ValueError(ERR_INVALID_WORKSPACE_ID)
 
+        # Like /vars, the /all-vars endpoint is not paginated; request once.
         path = f"/api/v2/workspaces/{workspace_id}/all-vars"
         params: dict[str, Any] = {}
         if options:
             # Add any options if needed in the future
             pass
 
-        for item in self._list(path, params=params):
+        for item in self._list(path, params=params, paginated=False):
             attr = item.get("attributes", {}) or {}
             var_id = item.get("id", "")
             variable_data = dict(attr)

tests/units/test_variable.py

@@ -0,0 +1,132 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+"""Unit tests for the workspace Variables resource.
+
+The headline cases here are regression tests for hashicorp/python-tfe#181:
+``variables.list()`` infinite-looping on workspaces with >= 100 variables
+because the ``/vars`` (and ``/all-vars``) endpoints are not paginated.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+
+from pytfe._http import HTTPTransport
+from pytfe.errors import ERR_INVALID_WORKSPACE_ID
+from pytfe.models.variable import Variable
+from pytfe.resources._base import _Service
+from pytfe.resources.variable import Variables
+
+
+def _vars_payload(count: int) -> dict:
+    """A /vars-style response: full set in one page, no meta.pagination."""
+    return {
+        "data": [
+            {
+                "id": f"var-{i}",
+                "type": "vars",
+                "attributes": {"key": f"key-{i}", "value": f"value-{i}"},
+            }
+            for i in range(count)
+        ]
+    }
+
+
+class TestVariablesList:
+    """Tests for Variables.list / list_all."""
+
+    def setup_method(self):
+        self.mock_transport = Mock(spec=HTTPTransport)
+        self.variables = Variables(self.mock_transport)
+        self.workspace_id = "ws-test123"
+
+    def test_list_validations(self):
+        with pytest.raises(ValueError, match=ERR_INVALID_WORKSPACE_ID):
+            list(self.variables.list(""))
+        with pytest.raises(ValueError, match=ERR_INVALID_WORKSPACE_ID):
+            list(self.variables.list(None))
+
+    def test_list_all_validations(self):
+        with pytest.raises(ValueError, match=ERR_INVALID_WORKSPACE_ID):
+            list(self.variables.list_all(""))
+        with pytest.raises(ValueError, match=ERR_INVALID_WORKSPACE_ID):
+            list(self.variables.list_all(None))
+
+    def test_list_does_not_paginate_with_100_plus_variables(self):
+        """Regression for #181: a workspace with >= 100 vars must not loop.
+
+        The endpoint ignores page params and re-returns the full set, so the
+        old pagination heuristic looped forever. We now issue exactly one
+        request and return each variable once.
+        """
+        response = Mock()
+        response.json.return_value = _vars_payload(150)
+        self.mock_transport.request.return_value = response
+
+        result = list(self.variables.list(self.workspace_id))
+
+        # Exactly one request — no follow-up page fetches.
+        self.mock_transport.request.assert_called_once_with(
+            "GET",
+            f"/api/v2/workspaces/{self.workspace_id}/vars",
+            params={},
+        )
+        # All 150 variables, no duplication.
+        assert len(result) == 150
+        assert all(isinstance(v, Variable) for v in result)
+        assert [v.id for v in result] == [f"var-{i}" for i in range(150)]
+
+    def test_list_all_does_not_paginate_with_100_plus_variables(self):
+        response = Mock()
+        response.json.return_value = _vars_payload(120)
+        self.mock_transport.request.return_value = response
+
+        result = list(self.variables.list_all(self.workspace_id))
+
+        self.mock_transport.request.assert_called_once_with(
+            "GET",
+            f"/api/v2/workspaces/{self.workspace_id}/all-vars",
+            params={},
+        )
+        assert len(result) == 120
+        assert [v.id for v in result] == [f"var-{i}" for i in range(120)]
+
+    def test_list_exactly_100_variables(self):
+        """The exactly-page-size boundary also looped under the old logic."""
+        response = Mock()
+        response.json.return_value = _vars_payload(100)
+        self.mock_transport.request.return_value = response
+
+        result = list(self.variables.list(self.workspace_id))
+
+        self.mock_transport.request.assert_called_once()
+        assert len(result) == 100
+
+    def test_list_empty(self):
+        response = Mock()
+        response.json.return_value = {"data": []}
+        self.mock_transport.request.return_value = response
+
+        result = list(self.variables.list(self.workspace_id))
+
+        self.mock_transport.request.assert_called_once()
+        assert result == []
+
+
+class TestListSafetyNet:
+    """The generic _list safety net protects any non-paginated endpoint."""
+
+    def test_full_page_without_metadata_is_treated_as_single_page(self):
+        """A paginated (paginated=True) call that gets a full page with no
+        meta.pagination must stop after one request rather than loop."""
+        transport = Mock(spec=HTTPTransport)
+        response = Mock()
+        response.json.return_value = _vars_payload(100)  # full page, no meta
+        transport.request.return_value = response
+
+        service = _Service(transport)
+        result = list(service._list("/api/v2/some/unpaginated", params={}))
+
+        transport.request.assert_called_once()
+        assert len(result) == 100