From 830cbf724e11fda11ae61f35cc46eba33e57d010 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 5 Mar 2026 17:03:17 +0530 Subject: [PATCH 1/4] refactor(policy-check): Iterator conversion and unit tests added to the policy-check --- examples/policy_check.py | 11 +- src/pytfe/models/__init__.py | 3 - src/pytfe/models/policy_check.py | 1 - src/pytfe/resources/policy_check.py | 36 +--- tests/units/test_policy_check.py | 284 ++++++++++++++++++++++++++++ 5 files changed, 298 insertions(+), 37 deletions(-) create mode 100644 tests/units/test_policy_check.py diff --git a/examples/policy_check.py b/examples/policy_check.py index 67f20813..f23a7396 100644 --- a/examples/policy_check.py +++ b/examples/policy_check.py @@ -35,7 +35,6 @@ def main(): action="store_true", help="Get logs for the specified policy check", ) - parser.add_argument("--page", type=int, default=1) parser.add_argument("--page-size", type=int, default=20) args = parser.parse_args() @@ -50,21 +49,19 @@ def main(): _print_header(f"Listing policy checks for run: {args.run_id}") options = PolicyCheckListOptions( - page_number=args.page, page_size=args.page_size, ) try: - pc_list = client.policy_checks.list(args.run_id, options) + pc_list = list(client.policy_checks.list(args.run_id, options)) - print(f"Total policy checks: {pc_list.total_count}") - print(f"Page {pc_list.current_page} of {pc_list.total_pages}") + print(f"Total policy checks fetched: {len(pc_list)}") print() - if not pc_list.items: + if not pc_list: print("No policy checks found for this run.") else: - for pc in pc_list.items: + for pc in pc_list: print(f"- ID: {pc.id}") print(f"Status: {pc.status}") print(f"Scope: {pc.scope}") diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 71faee66..8efe5e75 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -106,7 +106,6 @@ PolicyActions, PolicyCheck, PolicyCheckIncludeOpt, - PolicyCheckList, PolicyCheckListOptions, PolicyPermissions, PolicyResult, @@ -611,7 +610,6 @@ "PolicyResult", "PolicyStatusTimestamps", "PolicyCheckListOptions", - "PolicyCheckList", # Policy Evaluation "PolicyAttachable", "PolicyEvaluation", @@ -680,4 +678,3 @@ # Rebuild models with forward references after all models are loaded PolicyCheck.model_rebuild() -PolicyCheckList.model_rebuild() diff --git a/src/pytfe/models/policy_check.py b/src/pytfe/models/policy_check.py index 7b97274e..c0bf3a71 100644 --- a/src/pytfe/models/policy_check.py +++ b/src/pytfe/models/policy_check.py @@ -106,7 +106,6 @@ class PolicyCheckListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) include: list[PolicyCheckIncludeOpt] | None = Field(None, alias="include") - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/resources/policy_check.py b/src/pytfe/resources/policy_check.py index affae777..4f05811d 100644 --- a/src/pytfe/resources/policy_check.py +++ b/src/pytfe/resources/policy_check.py @@ -1,6 +1,7 @@ from __future__ import annotations import time +from collections.abc import Iterator from ..errors import ( InvalidPolicyCheckIDError, @@ -8,7 +9,6 @@ ) from ..models.policy_check import ( PolicyCheck, - PolicyCheckList, PolicyCheckListOptions, PolicyStatus, ) @@ -24,35 +24,19 @@ class PolicyChecks(_Service): def list( self, run_id: str, options: PolicyCheckListOptions | None = None - ) -> PolicyCheckList: + ) -> Iterator[PolicyCheck]: """List all policy checks of the given run.""" if not valid_string_id(run_id): raise InvalidRunIDError() params = ( options.model_dump(by_alias=True, exclude_none=True) if options else None ) - r = self.t.request( - "GET", - f"/api/v2/runs/{run_id}/policy-checks", - params=params, - ) - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - for d in jd.get("data", []): - attrs = d.get("attributes", {}) - attrs["id"] = d.get("id") - attrs["run"] = d.get("relationships", {}).get("run", {}) - items.append(PolicyCheck.model_validate(attrs)) - return PolicyCheckList( - items=items, - current_page=pagination.get("current-page"), - total_pages=pagination.get("total-pages"), - prev_page=pagination.get("prev-page"), - next_page=pagination.get("next-page"), - total_count=pagination.get("total-count"), - ) + path = f"/api/v2/runs/{run_id}/policy-checks" + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + attrs["run"] = item.get("relationships", {}).get("run", {}).get("data") + yield PolicyCheck.model_validate(attrs) def read(self, policy_check_id: str) -> PolicyCheck: """Read a policy check by its ID.""" @@ -66,7 +50,7 @@ def read(self, policy_check_id: str) -> PolicyCheck: d = jd.get("data", {}) attrs = d.get("attributes", {}) attrs["id"] = d.get("id") - attrs["run"] = d.get("relationships", {}).get("run", {}) + attrs["run"] = d.get("relationships", {}).get("run", {}).get("data") return PolicyCheck.model_validate(attrs) def override(self, policy_check_id: str) -> PolicyCheck: @@ -81,7 +65,7 @@ def override(self, policy_check_id: str) -> PolicyCheck: d = jd.get("data", {}) attrs = d.get("attributes", {}) attrs["id"] = d.get("id") - attrs["run"] = d.get("relationships", {}).get("run", {}) + attrs["run"] = d.get("relationships", {}).get("run", {}).get("data") return PolicyCheck.model_validate(attrs) def logs(self, policy_check_id: str) -> str: diff --git a/tests/units/test_policy_check.py b/tests/units/test_policy_check.py new file mode 100644 index 00000000..07642cb5 --- /dev/null +++ b/tests/units/test_policy_check.py @@ -0,0 +1,284 @@ +"""Unit tests for the policy_check module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidPolicyCheckIDError, InvalidRunIDError +from pytfe.models.policy_check import ( + PolicyCheck, + PolicyCheckIncludeOpt, + PolicyCheckListOptions, + PolicyStatus, +) +from pytfe.resources.policy_check import PolicyChecks + + +class TestPolicyChecks: + """Test the PolicyChecks service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def policy_checks_service(self, mock_transport): + """Create a PolicyChecks service with mocked transport.""" + return PolicyChecks(mock_transport) + + def test_list_policy_checks_validation(self, policy_checks_service): + """Test list() with invalid run ID.""" + with pytest.raises(InvalidRunIDError): + list(policy_checks_service.list("")) + + def test_list_policy_checks_iterator(self, policy_checks_service): + """Test list() returns an iterator of PolicyCheck models.""" + # Mock items match the raw data objects yielded by _list() as per the + # List Policy Checks API response: + # https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks#list-policy-checks + mock_items = [ + { + "id": "polchk-9VYRc9bpfJEsnwum", + "type": "policy-checks", + "attributes": { + "result": { + "result": False, + "passed": 0, + "total-failed": 1, + "hard-failed": 0, + "soft-failed": 1, + "advisory-failed": 0, + "duration-ms": 0, + "sentinel": None, + }, + "scope": "organization", + "status": "soft_failed", + "status-timestamps": { + "queued-at": "2017-11-29T20:02:17+00:00", + "soft-failed-at": "2017-11-29T20:02:20+00:00", + }, + "actions": {"is-overridable": True}, + "permissions": {"can-override": False}, + }, + "relationships": { + "run": {"data": {"id": "run-veDoQbv6xh6TbnJD", "type": "runs"}} + }, + "links": { + "output": "/api/v2/policy-checks/polchk-9VYRc9bpfJEsnwum/output" + }, + }, + { + "id": "polchk-passed456", + "type": "policy-checks", + "attributes": { + "result": { + "result": True, + "passed": 3, + "total-failed": 0, + "hard-failed": 0, + "soft-failed": 0, + "advisory-failed": 0, + "duration-ms": 120, + "sentinel": None, + }, + "scope": "workspace", + "status": "passed", + "status-timestamps": { + "queued-at": "2017-11-29T20:02:17+00:00", + "passed-at": "2017-11-29T20:02:19+00:00", + }, + "actions": {"is-overridable": False}, + "permissions": {"can-override": False}, + }, + "relationships": { + "run": {"data": {"id": "run-veDoQbv6xh6TbnJD", "type": "runs"}} + }, + "links": {"output": "/api/v2/policy-checks/polchk-passed456/output"}, + }, + ] + + with patch.object(policy_checks_service, "_list") as mock_list: + mock_list.return_value = mock_items + + options = PolicyCheckListOptions( + page_size=25, + include=[PolicyCheckIncludeOpt.POLICY_CHECK_RUN], + ) + result = list(policy_checks_service.list("run-1", options)) + + mock_list.assert_called_once() + call_args = mock_list.call_args + assert call_args[0][0] == "/api/v2/runs/run-1/policy-checks" + params = call_args[1]["params"] + assert params["page[size]"] == 25 + + assert len(result) == 2 + assert all(isinstance(item, PolicyCheck) for item in result) + + pc0 = result[0] + assert pc0.id == "polchk-9VYRc9bpfJEsnwum" + assert pc0.status == PolicyStatus.POLICY_SOFT_FAILED + assert pc0.scope.value == "organization" + assert pc0.result is not None + assert pc0.result.soft_failed == 1 + assert pc0.result.passed == 0 + assert pc0.result.total_failed == 1 + assert pc0.actions.is_overridable is True + assert pc0.permissions.can_override is False + assert pc0.status_timestamps.queued_at is not None + assert pc0.status_timestamps.soft_failed_at is not None + + pc1 = result[1] + assert pc1.id == "polchk-passed456" + assert pc1.status == PolicyStatus.POLICY_PASSES + assert pc1.scope.value == "workspace" + assert pc1.result.passed == 3 + assert pc1.result.total_failed == 0 + + def test_read_policy_check(self, policy_checks_service, mock_transport): + """Test read() for a policy check. + + Mock matches the Show Policy Check API response: + https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks#show-policy-check + """ + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "polchk-9VYRc9bpfJEsnwum", + "type": "policy-checks", + "attributes": { + "result": { + "result": False, + "passed": 0, + "total-failed": 1, + "hard-failed": 0, + "soft-failed": 1, + "advisory-failed": 0, + "duration-ms": 0, + "sentinel": None, + }, + "scope": "organization", + "status": "soft_failed", + "status-timestamps": { + "queued-at": "2017-11-29T20:02:17+00:00", + "soft-failed-at": "2017-11-29T20:02:20+00:00", + }, + "actions": {"is-overridable": True}, + "permissions": {"can-override": False}, + }, + "relationships": { + "run": {"data": {"id": "run-veDoQbv6xh6TbnJD", "type": "runs"}} + }, + "links": { + "output": "/api/v2/policy-checks/polchk-9VYRc9bpfJEsnwum/output" + }, + } + } + mock_transport.request.return_value = mock_response + + result = policy_checks_service.read("polchk-9VYRc9bpfJEsnwum") + + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/policy-checks/polchk-9VYRc9bpfJEsnwum" + ) + assert result.id == "polchk-9VYRc9bpfJEsnwum" + assert result.status == PolicyStatus.POLICY_SOFT_FAILED + assert result.scope.value == "organization" + assert result.result is not None + assert result.result.soft_failed == 1 + assert result.result.total_failed == 1 + assert result.result.passed == 0 + assert result.result.result is False + assert result.actions.is_overridable is True + assert result.permissions.can_override is False + assert result.status_timestamps.queued_at is not None + assert result.status_timestamps.soft_failed_at is not None + + def test_override_policy_check(self, policy_checks_service, mock_transport): + """Test override() for a policy check. + + Mock matches the Override Policy API response: + https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks#override-policy + """ + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "polchk-EasPB4Srx5NAiWAU", + "type": "policy-checks", + "attributes": { + "result": { + "result": False, + "passed": 0, + "total-failed": 1, + "hard-failed": 0, + "soft-failed": 1, + "advisory-failed": 0, + "duration-ms": 0, + "sentinel": None, + }, + "scope": "organization", + "status": "overridden", + "status-timestamps": { + "queued-at": "2017-11-29T20:13:37+00:00", + "soft-failed-at": "2017-11-29T20:13:40+00:00", + "overridden-at": "2017-11-29T20:14:11+00:00", + }, + "actions": {"is-overridable": True}, + "permissions": {"can-override": False}, + }, + "links": { + "output": "/api/v2/policy-checks/polchk-EasPB4Srx5NAiWAU/output" + }, + } + } + mock_transport.request.return_value = mock_response + + result = policy_checks_service.override("polchk-EasPB4Srx5NAiWAU") + + mock_transport.request.assert_called_once_with( + "POST", "/api/v2/policy-checks/polchk-EasPB4Srx5NAiWAU/actions/override" + ) + assert result.id == "polchk-EasPB4Srx5NAiWAU" + assert result.status == PolicyStatus.POLICY_OVERRIDDEN + assert result.scope.value == "organization" + assert result.result.soft_failed == 1 + assert result.actions.is_overridable is True + assert result.status_timestamps.soft_failed_at is not None + assert result.status_timestamps.queued_at is not None + + def test_logs_invalid_id(self, policy_checks_service): + """Test logs() with invalid policy check ID.""" + with pytest.raises(InvalidPolicyCheckIDError): + policy_checks_service.logs("") + + def test_logs_waits_until_ready(self, policy_checks_service, mock_transport): + """Test logs() polling until policy status is no longer pending/queued.""" + pending_pc = PolicyCheck(id="pc-1", status=PolicyStatus.POLICY_PENDING) + passed_pc = PolicyCheck(id="pc-1", status=PolicyStatus.POLICY_PASSES) + + with ( + patch.object(policy_checks_service, "read") as mock_read, + patch("pytfe.resources.policy_check.time.sleep") as mock_sleep, + ): + mock_read.side_effect = [pending_pc, passed_pc] + mock_response = Mock() + mock_response.text = "policy output" + mock_transport.request.return_value = mock_response + + logs = policy_checks_service.logs("pc-1") + + assert logs == "policy output" + assert mock_read.call_count == 2 + mock_sleep.assert_called_once_with(0.5) + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/policy-checks/pc-1/output" + ) + + +def test_policy_check_list_options_has_no_page_number(): + """Ensure iterator-style list options no longer expose page_number.""" + options = PolicyCheckListOptions(page_size=10) + dumped = options.model_dump(by_alias=True, exclude_none=True) + assert dumped == {"page[size]": 10} From e801ed0370acbc0d938242eec55d42e527570b9e Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 12 Mar 2026 12:05:35 +0530 Subject: [PATCH 2/4] refactor(state-version): Iterator conversion and unit tests added to the state-version --- examples/state_versions.py | 12 +- src/pytfe/models/__init__.py | 2 - src/pytfe/models/state_version.py | 1 - src/pytfe/resources/state_versions.py | 21 +- tests/units/test_state_version.py | 317 ++++++++++++++++++++++++++ 5 files changed, 327 insertions(+), 26 deletions(-) create mode 100644 tests/units/test_state_version.py diff --git a/examples/state_versions.py b/examples/state_versions.py index 6d3f8b82..6d6f83ad 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -33,7 +33,6 @@ def main(): parser.add_argument("--workspace-id", required=True, help="Workspace ID") parser.add_argument("--download", help="Path to save downloaded current state") parser.add_argument("--upload", help="Path to a .tfstate (or JSON state) to upload") - parser.add_argument("--page", type=int, default=1) parser.add_argument("--page-size", type=int, default=10) args = parser.parse_args() @@ -41,19 +40,16 @@ def main(): client = TFEClient(cfg) options = StateVersionListOptions( - page_number=args.page, page_size=args.page_size, organization=args.org, workspace=args.workspace, ) - sv_list = client.state_versions.list(options) - - print(f"Total state versions: {sv_list.total_count}") - print(f"Page {sv_list.current_page} of {sv_list.total_pages}") + sv_list = list(client.state_versions.list(options)) + print(f"Total state versions: {len(sv_list)}") print() - for sv in sv_list.items: + for sv in sv_list: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 1) List all state versions across org and workspace filters @@ -63,7 +59,7 @@ def main(): organization=args.org, workspace=args.workspace, page_size=args.page_size ) ) - for sv in all_sv.items: + for sv in all_sv: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 2) Read the current state version (with outputs included if you want) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 8efe5e75..79e41e80 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -294,7 +294,6 @@ StateVersion, StateVersionCreateOptions, StateVersionCurrentOptions, - StateVersionList, StateVersionListOptions, StateVersionReadOptions, ) @@ -667,7 +666,6 @@ "StateVersion", "StateVersionCreateOptions", "StateVersionCurrentOptions", - "StateVersionList", "StateVersionListOptions", "StateVersionReadOptions", # State Version Outputs diff --git a/src/pytfe/models/state_version.py b/src/pytfe/models/state_version.py index 018620c2..1fcfb5aa 100644 --- a/src/pytfe/models/state_version.py +++ b/src/pytfe/models/state_version.py @@ -78,7 +78,6 @@ class StateVersionListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) # Standard pagination + filters - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") organization: str | None = Field(None, alias="filter[organization][name]") workspace: str | None = Field(None, alias="filter[workspace][name]") diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index 29e1f5c8..026fa75d 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from urllib.parse import urlencode @@ -10,7 +11,6 @@ StateVersion, StateVersionCreateOptions, StateVersionCurrentOptions, - StateVersionList, StateVersionListOptions, StateVersionReadOptions, ) @@ -69,28 +69,19 @@ def _encode_query(params: dict[str, Any]) -> str: return "" return "?" + urlencode(clean, doseq=True) - def list(self, options: StateVersionListOptions | None = None) -> StateVersionList: + def list( + self, options: StateVersionListOptions | None = None + ) -> Iterator[StateVersion]: """ GET /state-versions Accepts filters for organization and workspace and standard pagination. """ params = options.model_dump(by_alias=True, exclude_none=True) if options else {} path = f"/api/v2/state-versions{self._encode_query(params)}" - r = self.t.request("GET", path) - jd = r.json() - # Expecting JSON:API list. Normalize to models. - items = [] - meta = jd.get("meta", {}) - for d in jd.get("data", []): + for d in self._list(path, params=params): attrs = d.get("attributes", {}) attrs["id"] = d.get("id") - items.append(StateVersion.model_validate(attrs)) - return StateVersionList( - items=items, - current_page=meta.get("pagination", {}).get("current-page"), - total_pages=meta.get("pagination", {}).get("total-pages"), - total_count=meta.get("pagination", {}).get("total-count"), - ) + yield StateVersion.model_validate(attrs) def read(self, state_version_id: str) -> StateVersion: """Read a state version by ID.""" diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py new file mode 100644 index 00000000..922082c5 --- /dev/null +++ b/tests/units/test_state_version.py @@ -0,0 +1,317 @@ +"""Unit tests for the state_versions module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import NotFound +from pytfe.models.state_version import ( + StateVersion, + StateVersionCreateOptions, + StateVersionCurrentOptions, + StateVersionIncludeOpt, + StateVersionListOptions, + StateVersionReadOptions, + StateVersionStatus, +) +from pytfe.models.state_version_output import StateVersionOutputsListOptions +from pytfe.resources.state_versions import StateVersions + + +class TestStateVersions: + """Test the StateVersions service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def state_versions_service(self, mock_transport): + """Create a StateVersions service with mocked transport.""" + return StateVersions(mock_transport) + + def test_list_state_versions_iterator(self, state_versions_service): + """Test list() returns an iterator of StateVersion models.""" + mock_items = [ + { + "id": "sv-1", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-01T00:00:00Z", + "serial": 9, + "state-version": 4, + "status": "finalized", + "hosted-state-download-url": "https://example.com/download-1", + "hosted-json-state-download-url": "https://example.com/json-download-1", + "resources-processed": True, + "terraform-version": "1.7.5", + }, + "relationships": { + "workspace": {"data": {"id": "ws-123", "type": "workspaces"}}, + "run": {"data": {"id": "run-123", "type": "runs"}}, + }, + "links": {"self": "/api/v2/state-versions/sv-1"}, + }, + { + "id": "sv-2", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-02T00:00:00Z", + "serial": 10, + "state-version": 4, + "status": "pending", + }, + "relationships": { + "workspace": {"data": {"id": "ws-123", "type": "workspaces"}}, + }, + "links": {"self": "/api/v2/state-versions/sv-2"}, + }, + ] + + with patch.object(state_versions_service, "_list") as mock_list: + mock_list.return_value = mock_items + + options = StateVersionListOptions( + page_size=2, + organization="demo-org", + workspace="demo-ws", + ) + result = list(state_versions_service.list(options)) + + mock_list.assert_called_once() + call_args = mock_list.call_args + assert call_args[0][0].startswith("/api/v2/state-versions?") + params = call_args[1]["params"] + assert params["page[size]"] == 2 + assert params["filter[organization][name]"] == "demo-org" + assert params["filter[workspace][name]"] == "demo-ws" + + assert len(result) == 2 + assert all(isinstance(item, StateVersion) for item in result) + assert result[0].id == "sv-1" + assert result[0].status == StateVersionStatus.FINALIZED + assert result[1].id == "sv-2" + assert result[1].status == StateVersionStatus.PENDING + + def test_read_state_version_invalid_id(self, state_versions_service): + """Test read() with invalid state version id.""" + with pytest.raises(ValueError, match="invalid state version id"): + state_versions_service.read("") + + def test_read_state_version_success(self, state_versions_service, mock_transport): + """Test successful read() operation.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "sv-read-1", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-01T00:00:00Z", + "serial": 9, + "state-version": 4, + "status": "finalized", + "hosted-state-download-url": "https://example.com/download", + "resources-processed": True, + }, + "relationships": { + "workspace": {"data": {"id": "ws-123", "type": "workspaces"}}, + "run": {"data": {"id": "run-123", "type": "runs"}}, + }, + "links": {"self": "/api/v2/state-versions/sv-read-1"}, + } + } + mock_transport.request.return_value = mock_response + + result = state_versions_service.read("sv-read-1") + + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/state-versions/sv-read-1" + ) + assert result.id == "sv-read-1" + assert result.status == StateVersionStatus.FINALIZED + assert result.hosted_state_download_url == "https://example.com/download" + + def test_read_with_options_success(self, state_versions_service, mock_transport): + """Test successful read_with_options() operation.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "sv-read-2", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-01T00:00:00Z", + "status": "pending", + }, + "relationships": { + "outputs": { + "data": [{"id": "wsout-1", "type": "state-version-outputs"}] + } + }, + } + } + mock_transport.request.return_value = mock_response + + options = StateVersionReadOptions( + include=[StateVersionIncludeOpt.OUTPUTS, StateVersionIncludeOpt.RUN] + ) + result = state_versions_service.read_with_options("sv-read-2", options) + + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/state-versions/sv-read-2", + params={"include": "outputs,run"}, + ) + assert result.id == "sv-read-2" + assert result.status == StateVersionStatus.PENDING + + def test_read_current_with_options_success( + self, state_versions_service, mock_transport + ): + """Test successful read_current_with_options() operation. + + Mock payload shape follows docs sample for: + GET /api/v2/workspaces/:workspace_id/current-state-version + """ + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "sv-current-1", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-01T00:00:00Z", + "serial": 9, + "status": "finalized", + "resources-processed": True, + }, + "relationships": { + "workspace": {"data": {"id": "ws-123", "type": "workspaces"}}, + "created-by": {"data": {"id": "user-123", "type": "users"}}, + }, + } + } + mock_transport.request.return_value = mock_response + + options = StateVersionCurrentOptions( + include=[StateVersionIncludeOpt.CREATED_BY] + ) + result = state_versions_service.read_current_with_options("ws-123", options) + + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/workspaces/ws-123/current-state-version", + params={"include": "created_by"}, + ) + assert result.id == "sv-current-1" + + def test_create_state_version_success(self, state_versions_service, mock_transport): + """Test successful create() operation.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "sv-new-1", + "type": "state-versions", + "attributes": { + "created-at": "2024-01-01T00:00:00Z", + "status": "pending", + "hosted-state-upload-url": "https://example.com/upload", + "hosted-json-state-upload-url": "https://example.com/json-upload", + "serial": 10, + }, + "links": {"self": "/api/v2/state-versions/sv-new-1"}, + } + } + mock_transport.request.return_value = mock_response + + options = StateVersionCreateOptions(serial=10, md5="abc123") + + with patch.object( + state_versions_service, "_resolve_workspace_id", return_value="ws-123" + ): + result = state_versions_service.create("my-workspace", options) + + mock_transport.request.assert_called_once_with( + "POST", + "/api/v2/workspaces/ws-123/state-versions", + json_body={ + "data": { + "type": "state-versions", + "attributes": { + "serial": 10, + "md5": "abc123", + }, + } + }, + ) + assert result.id == "sv-new-1" + assert result.status == StateVersionStatus.PENDING + + def test_download_state_version_not_found_when_url_missing( + self, state_versions_service + ): + """Test download() raises NotFound if signed download URL is missing.""" + sv = StateVersion( + id="sv-404", + created_at="2024-01-01T00:00:00Z", + status=StateVersionStatus.FINALIZED, + hosted_state_download_url=None, + ) + + with patch.object(state_versions_service, "read", return_value=sv): + with pytest.raises(NotFound, match="download url not available"): + state_versions_service.download("sv-404") + + def test_download_state_version_success( + self, state_versions_service, mock_transport + ): + """Test successful download() operation using signed URL.""" + sv = StateVersion( + id="sv-dl-1", + created_at="2024-01-01T00:00:00Z", + status=StateVersionStatus.FINALIZED, + hosted_state_download_url="https://example.com/signed-download", + ) + mock_response = Mock() + mock_response.content = b"{}" + mock_transport.request.return_value = mock_response + + with patch.object(state_versions_service, "read", return_value=sv): + result = state_versions_service.download("sv-dl-1") + + mock_transport.request.assert_called_once_with( + "GET", + "https://example.com/signed-download", + allow_redirects=True, + headers={"Accept": "application/json"}, + ) + assert result == b"{}" + + def test_list_outputs_success(self, state_versions_service, mock_transport): + """Test successful list_outputs() operation.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": [ + { + "id": "wsout-1", + "attributes": { + "name": "vpc_id", + "sensitive": False, + "type": "string", + "value": "vpc-123", + }, + } + ], + } + mock_transport.request.return_value = mock_response + + options = StateVersionOutputsListOptions(page_number=1, page_size=5) + result = state_versions_service.list_outputs("sv-outputs-1", options) + + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/state-versions/sv-outputs-1/outputs", + params={"page[number]": 1, "page[size]": 5}, + ) + assert len(result.items) == 1 + assert result.items[0].id == "wsout-1" From 0d3f6ee855481e875bb184d8715025386a408c4a Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 12 Mar 2026 12:48:35 +0530 Subject: [PATCH 3/4] refactor(state-version-output): Iterator conversion and example added to the state-version-output --- examples/state_versions.py | 23 +++++++-- src/pytfe/models/__init__.py | 2 - src/pytfe/models/state_version_output.py | 1 - src/pytfe/resources/state_version_outputs.py | 35 ++++---------- src/pytfe/resources/state_versions.py | 33 ++++--------- tests/units/test_state_version.py | 50 ++++++++++---------- 6 files changed, 59 insertions(+), 85 deletions(-) diff --git a/examples/state_versions.py b/examples/state_versions.py index 6d6f83ad..3906df01 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -80,15 +80,30 @@ def main(): # 4) List outputs for the current state version (paged) _print_header("Listing outputs (current state version)") - outs = client.state_versions.list_outputs( - current.id, options=StateVersionOutputsListOptions(page_size=50) + outs = list( + client.state_versions.list_outputs( + current.id, options=StateVersionOutputsListOptions(page_size=50) + ) ) - if not outs.items: + if not outs: print("No outputs found.") - for o in outs.items: + for o in outs: # Sensitive outputs will have value = None print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") + if args.workspace_id: + # 4b) List outputs for the current state version via workspace endpoint + _print_header("Listing outputs via workspace endpoint") + outs2 = list( + client.state_version_outputs.read_current( + args.workspace_id, options=StateVersionOutputsListOptions(page_size=50) + ) + ) + if not outs2: + print("No outputs found.") + for o in outs2: + print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") + # 5) (Optional) Upload a new state file if args.upload: _print_header(f"Uploading new state from: {args.upload}") diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 79e41e80..3b768014 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -299,7 +299,6 @@ ) from .state_version_output import ( StateVersionOutput, - StateVersionOutputsList, StateVersionOutputsListOptions, ) from .team import ( @@ -670,7 +669,6 @@ "StateVersionReadOptions", # State Version Outputs "StateVersionOutput", - "StateVersionOutputsList", "StateVersionOutputsListOptions", ] diff --git a/src/pytfe/models/state_version_output.py b/src/pytfe/models/state_version_output.py index 1d69b271..b7c1b025 100644 --- a/src/pytfe/models/state_version_output.py +++ b/src/pytfe/models/state_version_output.py @@ -19,7 +19,6 @@ class StateVersionOutput(BaseModel): class StateVersionOutputsListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/resources/state_version_outputs.py b/src/pytfe/resources/state_version_outputs.py index 205073c4..25be7f74 100644 --- a/src/pytfe/resources/state_version_outputs.py +++ b/src/pytfe/resources/state_version_outputs.py @@ -1,10 +1,10 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from ..models.state_version_output import ( StateVersionOutput, - StateVersionOutputsList, StateVersionOutputsListOptions, ) from ..utils import valid_string_id @@ -42,7 +42,7 @@ def read_current( self, workspace_id: str, options: StateVersionOutputsListOptions | None = None, - ) -> StateVersionOutputsList: + ) -> Iterator[StateVersionOutput]: """ Read outputs for the workspace's current state version. Note: sensitive outputs are returned with null values by the API. @@ -52,32 +52,13 @@ def read_current( params: dict[str, Any] = {} if options: - if options.page_number is not None: - params["page[number]"] = options.page_number if options.page_size is not None: params["page[size]"] = options.page_size + path = f"/api/v2/workspaces/{workspace_id}/current-state-version-outputs" - r = self.t.request( - "GET", - f"/api/v2/workspaces/{workspace_id}/current-state-version-outputs", - params=params, - ) - data = r.json() - - items: list[StateVersionOutput] = [] - for item in data.get("data", []): - attr = item.get("attributes", {}) or {} - items.append( - StateVersionOutput( - id=_safe_str(item.get("id")), - **{k.replace("-", "_"): v for k, v in attr.items()}, - ) + for d in self._list(path, params=params): + attr = d.get("attributes", {}) or {} + yield StateVersionOutput( + id=_safe_str(d.get("id")), + **{k.replace("-", "_"): v for k, v in attr.items()}, ) - - meta = data.get("meta", {}).get("pagination", {}) or {} - return StateVersionOutputsList( - items=items, - current_page=meta.get("current-page"), - total_pages=meta.get("total-pages"), - total_count=meta.get("total-count"), - ) diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index 026fa75d..7f8a81cb 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -16,7 +16,6 @@ ) from ..models.state_version_output import ( StateVersionOutput, - StateVersionOutputsList, StateVersionOutputsListOptions, ) from ..utils import looks_like_workspace_id, valid_string_id @@ -254,40 +253,24 @@ def list_outputs( self, state_version_id: str, options: StateVersionOutputsListOptions | None = None, - ) -> StateVersionOutputsList: + ) -> Iterator[StateVersionOutput]: """List outputs for a given state version (paged).""" if not valid_string_id(state_version_id): raise ValueError("invalid state version id") params: dict[str, Any] = {} if options: - if options.page_number is not None: - params["page[number]"] = options.page_number if options.page_size is not None: params["page[size]"] = options.page_size - r = self.t.request( - "GET", f"/api/v2/state-versions/{state_version_id}/outputs", params=params - ) - data = r.json() - - items: list[StateVersionOutput] = [] - for item in data.get("data", []): - attr = item.get("attributes", {}) or {} - items.append( - StateVersionOutput( - id=_safe_str(item.get("id")), - **{k.replace("-", "_"): v for k, v in attr.items()}, - ) - ) + path = f"/api/v2/state-versions/{state_version_id}/outputs" - meta = data.get("meta", {}).get("pagination", {}) or {} - return StateVersionOutputsList( - items=items, - current_page=meta.get("current-page"), - total_pages=meta.get("total-pages"), - total_count=meta.get("total-count"), - ) + for d in self._list(path, params=params): + attr = d.get("attributes", {}) or {} + yield StateVersionOutput( + id=_safe_str(d.get("id")), + **{k.replace("-", "_"): v for k, v in attr.items()}, + ) # ---------------------------- # TFE-only backing data actions diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py index 922082c5..11f67c8f 100644 --- a/tests/units/test_state_version.py +++ b/tests/units/test_state_version.py @@ -287,31 +287,29 @@ def test_download_state_version_success( ) assert result == b"{}" - def test_list_outputs_success(self, state_versions_service, mock_transport): - """Test successful list_outputs() operation.""" - mock_response = Mock() - mock_response.json.return_value = { - "data": [ - { - "id": "wsout-1", - "attributes": { - "name": "vpc_id", - "sensitive": False, - "type": "string", - "value": "vpc-123", - }, - } - ], - } - mock_transport.request.return_value = mock_response + def test_list_outputs_success(self, state_versions_service): + """Test successful list_outputs() iterator operation.""" + mock_items = [ + { + "id": "wsout-1", + "attributes": { + "name": "vpc_id", + "sensitive": False, + "type": "string", + "value": "vpc-123", + }, + } + ] + + with patch.object(state_versions_service, "_list") as mock_list: + mock_list.return_value = mock_items - options = StateVersionOutputsListOptions(page_number=1, page_size=5) - result = state_versions_service.list_outputs("sv-outputs-1", options) + options = StateVersionOutputsListOptions(page_size=5) + result = list(state_versions_service.list_outputs("sv-outputs-1", options)) - mock_transport.request.assert_called_once_with( - "GET", - "/api/v2/state-versions/sv-outputs-1/outputs", - params={"page[number]": 1, "page[size]": 5}, - ) - assert len(result.items) == 1 - assert result.items[0].id == "wsout-1" + mock_list.assert_called_once_with( + "/api/v2/state-versions/sv-outputs-1/outputs", + params={"page[size]": 5}, + ) + assert len(result) == 1 + assert result[0].id == "wsout-1" From be6968d8f08133f2b747fe3fc7dd2d044e360f5a Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 12 Mar 2026 13:35:19 +0530 Subject: [PATCH 4/4] refactor(variable-set): Iterator pattern and test cases were updated for variable-set and variable-set-variables --- examples/variable_sets.py | 41 +++++++++++++++------ src/pytfe/models/variable_set.py | 1 - src/pytfe/resources/variable_sets.py | 54 +++++++++------------------- tests/units/test_variable_sets.py | 37 ++++++++++--------- 4 files changed, 69 insertions(+), 64 deletions(-) diff --git a/examples/variable_sets.py b/examples/variable_sets.py index 0b410845..042d89ae 100644 --- a/examples/variable_sets.py +++ b/examples/variable_sets.py @@ -64,7 +64,7 @@ def variable_set_example(): list_options = VariableSetListOptions( page_size=10, include=[VariableSetIncludeOpt.WORKSPACES] ) - variable_sets = client.variable_sets.list(org_name, list_options) + variable_sets = list(client.variable_sets.list(org_name, list_options)) print(f"Found {len(variable_sets)} existing variable sets") for vs in variable_sets[:3]: # Show first 3 @@ -92,6 +92,15 @@ def variable_set_example(): print(f"Priority: {new_variable_set.priority}") print() + print("Listing existing variable sets...") + list_options = VariableSetListOptions(page_size=10) + variable_sets = list(client.variable_sets.list(org_name, list_options)) + print(f"Found {len(variable_sets)} existing variable sets") + + for vs in variable_sets: # Show first 3 + print(f"- {vs.name} (ID: {vs.id}, Global: {vs.global_})") + print() + # 3. Create variables in the variable set print("3. Creating variables in the variable set...") @@ -147,12 +156,16 @@ def variable_set_example(): # 4. List variables in the variable set print("4. Listing variables in the variable set...") var_list_options = VariableSetVariableListOptions(page_size=50) - variables = client.variable_set_variables.list( - created_variable_set_id, var_list_options + variables = list( + client.variable_set_variables.list( + created_variable_set_id, var_list_options + ) ) print(f"Found {len(variables)} variables in the set:") - for var in variables: + for var in client.variable_set_variables.list( + created_variable_set_id, var_list_options + ): sensitive_note = " (sensitive)" if var.sensitive else "" hcl_note = " (HCL)" if var.hcl else "" print(f"- {var.key}: {var.category.value}{sensitive_note}{hcl_note}") @@ -212,10 +225,14 @@ def variable_set_example(): print("Successfully applied to workspace") # List variable sets for this workspace - workspace_varsets = client.variable_sets.list_for_workspace( + print(f"Listing variable sets for workspace: {first_workspace.name}") + workspace_varsets = 0 + for ws_varset in client.variable_sets.list_for_workspace( first_workspace.id - ) - print(f"Workspace now has {len(workspace_varsets)} variable sets") + ): + print(f"- {ws_varset.name} (ID: {ws_varset.id})") + workspace_varsets += 1 + print(f"Workspace now has {workspace_varsets} variable sets") # Remove from workspace remove_ws_options = VariableSetRemoveFromWorkspacesOptions( @@ -250,10 +267,14 @@ def variable_set_example(): print("Successfully applied to project") # List variable sets for this project - project_varsets = client.variable_sets.list_for_project( + print(f"Listing variable sets for project: {first_project.name}") + project_varsets = 0 + for proj_varset in client.variable_sets.list_for_project( first_project.id - ) - print(f"Project now has {len(project_varsets)} variable sets") + ): + print(f"- {proj_varset.name} (ID: {proj_varset.id})") + project_varsets += 1 + print(f"Project now has {project_varsets} variable sets") # Remove from project remove_proj_options = VariableSetRemoveFromProjectsOptions( diff --git a/src/pytfe/models/variable_set.py b/src/pytfe/models/variable_set.py index 0daee068..583f0cf2 100644 --- a/src/pytfe/models/variable_set.py +++ b/src/pytfe/models/variable_set.py @@ -69,7 +69,6 @@ class VariableSetListOptions(BaseModel): """Options for listing variable sets.""" # Pagination options - page_number: int | None = None page_size: int | None = None include: list[VariableSetIncludeOpt] | None = None query: str | None = None # Filter by name diff --git a/src/pytfe/resources/variable_sets.py b/src/pytfe/resources/variable_sets.py index 4bf4353c..18fb4b61 100644 --- a/src/pytfe/resources/variable_sets.py +++ b/src/pytfe/resources/variable_sets.py @@ -1,6 +1,7 @@ """Variable Set resource implementation for the Python TFE SDK.""" import builtins +from collections.abc import Iterator from typing import Any from .._http import HTTPTransport @@ -48,7 +49,7 @@ def list( self, organization: str, options: VariableSetListOptions | None = None, - ) -> list[VariableSet]: + ) -> Iterator[VariableSet]: """List all variable sets within an organization. Args: @@ -56,8 +57,7 @@ def list( options: Optional parameters for filtering and pagination Returns: - List of VariableSet objects - + Iterator of VariableSet objects within the organization Raises: ValueError: If organization name is invalid TFEError: If API request fails @@ -69,8 +69,6 @@ def list( params: dict[str, str] = {} if options: - if options.page_number: - params["page[number]"] = str(options.page_number) if options.page_size: params["page[size]"] = str(options.page_size) if options.query: @@ -78,16 +76,14 @@ def list( if options.include: params["include"] = ",".join([opt.value for opt in options.include]) - response = self.t.request("GET", path, params=params) - data = response.json() - - return self._parse_variable_sets_response(data) + for item in self._list(path, params=params): + yield self._parse_variable_set(item) def list_for_workspace( self, workspace_id: str, options: VariableSetListOptions | None = None, - ) -> builtins.list[VariableSet]: + ) -> Iterator[VariableSet]: """List variable sets associated with a workspace. Args: @@ -95,7 +91,7 @@ def list_for_workspace( options: Optional parameters for filtering and pagination Returns: - List of VariableSet objects associated with the workspace + Iterator of VariableSet objects associated with the workspace Raises: ValueError: If workspace_id is invalid @@ -108,8 +104,6 @@ def list_for_workspace( params: dict[str, str] = {} if options: - if options.page_number: - params["page[number]"] = str(options.page_number) if options.page_size: params["page[size]"] = str(options.page_size) if options.query: @@ -117,16 +111,14 @@ def list_for_workspace( if options.include: params["include"] = ",".join([opt.value for opt in options.include]) - response = self.t.request("GET", path, params=params) - data = response.json() - - return self._parse_variable_sets_response(data) + for item in self._list(path, params=params): + yield self._parse_variable_set(item) def list_for_project( self, project_id: str, options: VariableSetListOptions | None = None, - ) -> builtins.list[VariableSet]: + ) -> Iterator[VariableSet]: """List variable sets associated with a project. Args: @@ -134,7 +126,7 @@ def list_for_project( options: Optional parameters for filtering and pagination Returns: - List of VariableSet objects associated with the project + Iterator of VariableSet objects associated with the project Raises: ValueError: If project_id is invalid @@ -147,8 +139,6 @@ def list_for_project( params: dict[str, str] = {} if options: - if options.page_number: - params["page[number]"] = str(options.page_number) if options.page_size: params["page[size]"] = str(options.page_size) if options.query: @@ -156,10 +146,8 @@ def list_for_project( if options.include: params["include"] = ",".join([opt.value for opt in options.include]) - response = self.t.request("GET", path, params=params) - data = response.json() - - return self._parse_variable_sets_response(data) + for item in self._list(path, params=params): + yield self._parse_variable_set(item) def create( self, @@ -714,7 +702,7 @@ def list( self, variable_set_id: str, options: VariableSetVariableListOptions | None = None, - ) -> list[VariableSetVariable]: + ) -> Iterator[VariableSetVariable]: """List all variables in a variable set. Args: @@ -722,7 +710,7 @@ def list( options: Optional parameters for pagination Returns: - List of VariableSetVariable objects + Iterator of VariableSetVariable objects Raises: ValueError: If variable_set_id is invalid @@ -735,19 +723,11 @@ def list( params: dict[str, str] = {} if options: - if options.page_number: - params["page[number]"] = str(options.page_number) if options.page_size: params["page[size]"] = str(options.page_size) - response = self.t.request("GET", path, params=params) - data = response.json() - - variables = [] - for item in data.get("data", []): - variables.append(self._parse_variable_set_variable(item)) - - return variables + for item in self._list(path, params=params): + yield self._parse_variable_set_variable(item) def create( self, diff --git a/tests/units/test_variable_sets.py b/tests/units/test_variable_sets.py index 01ad8689..f43f5c5c 100644 --- a/tests/units/test_variable_sets.py +++ b/tests/units/test_variable_sets.py @@ -72,7 +72,7 @@ def test_list_variable_sets_success(self): self.mock_transport.request.return_value = mock_response # Call the method - result = self.variable_sets_service.list(self.org_name) + result = list(self.variable_sets_service.list(self.org_name)) # Assertions assert len(result) == 1 @@ -85,7 +85,9 @@ def test_list_variable_sets_success(self): # Verify API call self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/organizations/{self.org_name}/varsets", params={} + "GET", + f"/api/v2/organizations/{self.org_name}/varsets", + params={"page[number]": 1, "page[size]": 100}, ) def test_list_variable_sets_with_options(self): @@ -97,21 +99,20 @@ def test_list_variable_sets_with_options(self): # Create options options = VariableSetListOptions( - page_number=2, page_size=50, query="test", include=[VariableSetIncludeOpt.WORKSPACES, VariableSetIncludeOpt.PROJECTS], ) # Call the method - result = self.variable_sets_service.list(self.org_name, options) + result = list(self.variable_sets_service.list(self.org_name, options)) # Verify the result assert isinstance(result, list) # Verify API call with parameters expected_params = { - "page[number]": "2", + "page[number]": 1, "page[size]": "50", "q": "test", "include": "workspaces,projects", @@ -125,10 +126,10 @@ def test_list_variable_sets_with_options(self): def test_list_variable_sets_invalid_organization(self): """Test listing variable sets with invalid organization.""" with pytest.raises(ValueError, match="Organization name is required"): - self.variable_sets_service.list("") + list(self.variable_sets_service.list("")) with pytest.raises(ValueError, match="Organization name is required"): - self.variable_sets_service.list(None) + list(self.variable_sets_service.list(None)) def test_list_for_workspace_success(self): """Test successful listing of variable sets for workspace.""" @@ -152,7 +153,7 @@ def test_list_for_workspace_success(self): self.mock_transport.request.return_value = mock_response # Call the method - result = self.variable_sets_service.list_for_workspace(self.workspace_id) + result = list(self.variable_sets_service.list_for_workspace(self.workspace_id)) # Assertions assert len(result) == 1 @@ -160,13 +161,15 @@ def test_list_for_workspace_success(self): # Verify API call self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/workspaces/{self.workspace_id}/varsets", params={} + "GET", + f"/api/v2/workspaces/{self.workspace_id}/varsets", + params={"page[number]": 1, "page[size]": 100}, ) def test_list_for_workspace_invalid_id(self): """Test listing for workspace with invalid workspace ID.""" with pytest.raises(ValueError, match="Workspace ID is required"): - self.variable_sets_service.list_for_workspace("") + list(self.variable_sets_service.list_for_workspace("")) def test_list_for_project_success(self): """Test successful listing of variable sets for project.""" @@ -190,7 +193,7 @@ def test_list_for_project_success(self): self.mock_transport.request.return_value = mock_response # Call the method - result = self.variable_sets_service.list_for_project(self.project_id) + result = list(self.variable_sets_service.list_for_project(self.project_id)) # Assertions assert len(result) == 1 @@ -198,13 +201,15 @@ def test_list_for_project_success(self): # Verify API call self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/projects/{self.project_id}/varsets", params={} + "GET", + f"/api/v2/projects/{self.project_id}/varsets", + params={"page[number]": 1, "page[size]": 100}, ) def test_list_for_project_invalid_id(self): """Test listing for project with invalid project ID.""" with pytest.raises(ValueError, match="Project ID is required"): - self.variable_sets_service.list_for_project("") + list(self.variable_sets_service.list_for_project("")) def test_create_variable_set_success(self): """Test successful variable set creation.""" @@ -731,7 +736,7 @@ def test_list_variables_success(self): self.mock_transport.request.return_value = mock_response # Call the method - result = self.variables_service.list(self.variable_set_id) + result = list(self.variables_service.list(self.variable_set_id)) # Assertions assert len(result) == 1 @@ -748,13 +753,13 @@ def test_list_variables_success(self): self.mock_transport.request.assert_called_once_with( "GET", f"/api/v2/varsets/{self.variable_set_id}/relationships/vars", - params={}, + params={"page[number]": 1, "page[size]": 100}, ) def test_list_variables_invalid_varset_id(self): """Test listing variables with invalid variable set ID.""" with pytest.raises(ValueError, match="Variable set ID is required"): - self.variables_service.list("") + list(self.variables_service.list("")) def test_create_variable_success(self): """Test successful variable creation."""