diff --git a/examples/policy.py b/examples/policy.py index 74352f43..4629ccad 100644 --- a/examples/policy.py +++ b/examples/policy.py @@ -82,7 +82,6 @@ def main(): _print_header(f"Listing policies in organization: {args.org}") list_options = PolicyListOptions( - page_number=args.page, page_size=args.page_size, ) @@ -93,14 +92,10 @@ def main(): PolicyKind.SENTINEL if args.kind == "sentinel" else PolicyKind.OPA ) - policy_list = client.policies.list(args.org, list_options) - - print(f"Total policies: {policy_list.total_count}") - print(f"Page {policy_list.current_page} of {policy_list.total_pages}") - print() + policy_iter = client.policies.list(args.org, list_options) existing_policy = None - for policy in policy_list.items: + for policy in policy_iter: print( f"- {policy.id} | {policy.name} | kind={policy.kind} | enforcement={policy.enforcement_level}" ) diff --git a/examples/policy_set.py b/examples/policy_set.py index 1808d80a..1b7ecb2a 100644 --- a/examples/policy_set.py +++ b/examples/policy_set.py @@ -163,16 +163,15 @@ def main(): ) try: - ps_list = client.policy_sets.list(args.org, list_options) + ps_list = list(client.policy_sets.list(args.org, list_options)) - print(f"Total policy sets: {ps_list.total_count}") - print(f"Page {ps_list.current_page} of {ps_list.total_pages}") + print(f"Total policy sets: {len(ps_list)}") print() - if not ps_list.items: + if not ps_list: print("No policy sets found for this organization.") else: - for ps in ps_list.items: + for ps in ps_list: print( f"- ID: {ps.id} | Name: {ps.name} | Kind: {ps.kind} | Global: {ps.Global}" ) diff --git a/examples/run.py b/examples/run.py index d95b6e95..96010b07 100644 --- a/examples/run.py +++ b/examples/run.py @@ -67,7 +67,8 @@ def main(): ) try: - run_list = client.runs.list(args.workspace_id, options) + print("running inside run list") + run_list = list(client.runs.list(args.workspace_id, options)) except Exception as e: print(f"Error listing runs: {e}") if args.organization: @@ -75,23 +76,22 @@ def main(): else: return - if "run_list" in locals(): - print(f"Total runs: {run_list.total_count}") - print(f"Page {run_list.current_page} of {run_list.total_pages}") + if "run_list" in locals() and run_list: + print(f"Total runs fetched: {len(run_list)}") print() - for run in run_list.items: + for run in run_list: print(f"- {run.id} | status={run.status} | created={run.created_at}") print(f"message: {run.message}") print(f"has_changes: {run.has_changes} | is_destroy: {run.is_destroy}") - if not run_list.items: + if not run_list: print("No runs found.") else: # 2) Read the most recent run with details _print_header("Reading most recent run details") - latest_run = run_list.items[0] + latest_run = run_list[0] read_options = RunReadOptions( include=[ RunIncludeOpt.RUN_PLAN, @@ -188,10 +188,12 @@ def main(): status="applied,planned,errored", ) - org_runs = client.runs.list_for_organization(args.organization, org_options) - print(f"Found {len(org_runs.items)} runs across organization") + org_runs = list( + client.runs.list_for_organization(args.organization, org_options) + ) + print(f"Found {len(org_runs)} runs across organization") - for run in org_runs.items[:3]: # Show first 3 + for run in org_runs[:3]: # Show first 3 print(f"- {run.id} | status={run.status}") if run.workspace: print(f"workspace: {run.workspace.name}") @@ -204,15 +206,15 @@ def main(): _print_header("Run Actions Demo (Safe Mode)") # Get runs first if not already available - if "run_list" not in locals() or not run_list.items: + if "run_list" not in locals() or not run_list: try: options = RunListOptions(page_size=1) - run_list = client.runs.list(args.workspace_id, options) + run_list = list(client.runs.list(args.workspace_id, options)) except Exception as e: print(f"Error getting runs for actions demo: {e}") return - if not run_list.items: + if not run_list: print("No runs available for actions demo") return diff --git a/examples/run_events.py b/examples/run_events.py index a648c5b5..d8ae6129 100644 --- a/examples/run_events.py +++ b/examples/run_events.py @@ -94,23 +94,19 @@ def main(): options = RunEventListOptions(include=include_opts if include_opts else None) try: - event_list = client.run_events.list(args.run_id, options) - - print(f"Total run events: {event_list.total_count or 'N/A'}") - if event_list.current_page and event_list.total_pages: - print(f"Page {event_list.current_page} of {event_list.total_pages}") - print() + event_count = 0 + for event in client.run_events.list(args.run_id, options): + print(f"Event ID: {event.id}") + print(f"Action: {event.action or 'N/A'}") + print(f"Description: {event.description or 'N/A'}") + print(f"Created At: {event.created_at or 'N/A'}") + print() + event_count += 1 - if not event_list.items: + if event_count == 0: print("No run events found for this run.") else: - for event in event_list.items: - print(f"Event ID: {event.id}") - print(f"Action: {event.action or 'N/A'}") - print(f"Description: {event.description or 'N/A'}") - print(f"Created At: {event.created_at or 'N/A'}") - - print() + print(f"Total run events listed: {event_count}") except Exception as e: print(f"Error listing run events: {e}") @@ -139,7 +135,6 @@ def main(): # 3) Summary _print_header("Summary") print(f"Successfully demonstrated run events for run: {args.run_id}") - print(f"Total events found: {event_list.total_count or 'N/A'}") if args.event_id: print(f"Successfully read specific event: {args.event_id}") return 0 diff --git a/src/pytfe/client.py b/src/pytfe/client.py index d1c83373..df04eaf5 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -32,6 +32,7 @@ from .resources.ssh_keys import SSHKeys from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions +from .resources.user import Users from .resources.variable import Variables from .resources.variable_sets import VariableSets, VariableSetVariables from .resources.workspace_resources import WorkspaceResourcesService @@ -69,6 +70,7 @@ def __init__(self, config: TFEConfig | None = None): self.plans = Plans(self._transport) self.organizations = Organizations(self._transport) self.organization_memberships = OrganizationMemberships(self._transport) + self.users = Users(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) self.variable_sets = VariableSets(self._transport) diff --git a/src/pytfe/models/policy.py b/src/pytfe/models/policy.py index fb182d94..daf614f8 100644 --- a/src/pytfe/models/policy.py +++ b/src/pytfe/models/policy.py @@ -38,7 +38,6 @@ class PolicyListOptions(BaseModel): search: str | None = Field(None, alias="search[name]") kind: PolicyKind | None = Field(None, alias="filter[kind]") - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/models/user.py b/src/pytfe/models/user.py index 26b902e0..c2cbf015 100644 --- a/src/pytfe/models/user.py +++ b/src/pytfe/models/user.py @@ -7,6 +7,7 @@ class User(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str = Field(..., alias="id") + auth_method: str = Field(default="", alias="auth-method") avatar_url: str = Field(default="", alias="avatar-url") email: str = Field(default="", alias="email") is_service_account: bool = Field(default=False, alias="is-service-account") diff --git a/src/pytfe/resources/policy.py b/src/pytfe/resources/policy.py index fb30ca02..25153c0b 100644 --- a/src/pytfe/resources/policy.py +++ b/src/pytfe/resources/policy.py @@ -1,5 +1,8 @@ from __future__ import annotations +from collections.abc import Iterator +from typing import Any + from ..errors import ( InvalidNameError, InvalidOrgError, @@ -11,7 +14,6 @@ from ..models.policy import ( Policy, PolicyCreateOptions, - PolicyList, PolicyListOptions, PolicyUpdateOptions, ) @@ -22,35 +24,28 @@ class Policies(_Service): def list( self, organization: str, options: PolicyListOptions | None = None - ) -> PolicyList: - """List all the policies of the given organization.""" + ) -> Iterator[Policy]: + """Iterate all the policies of the given organization.""" if not valid_string_id(organization): raise InvalidOrgError() - params = ( - options.model_dump(by_alias=True, exclude_none=True) if options else None - ) - r = self.t.request( - "GET", - f"/api/v2/organizations/{organization}/policies", - 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["organization"] = d.get("relationships", {}).get("organization", {}) - items.append(Policy.model_validate(attrs)) - return PolicyList( - 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/organizations/{organization}/policies" + params: dict[str, Any] = {} + + if options: + if getattr(options, "page_size", None): + params["page[size]"] = str(options.page_size) + + def _gen() -> Iterator[Policy]: + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + attrs["organization"] = item.get("relationships", {}).get( + "organization", {} + ) + yield Policy.model_validate(attrs) + + return _gen() def create(self, organization: str, options: PolicyCreateOptions) -> Policy: """Create a new policy in the given organization.""" diff --git a/src/pytfe/resources/policy_set.py b/src/pytfe/resources/policy_set.py index f25e986c..19bedd3b 100644 --- a/src/pytfe/resources/policy_set.py +++ b/src/pytfe/resources/policy_set.py @@ -1,5 +1,7 @@ from __future__ import annotations +from collections.abc import Iterator + from ..errors import ( InvalidNameError, InvalidOrgError, @@ -17,7 +19,6 @@ PolicySetAddWorkspaceExclusionsOptions, PolicySetAddWorkspacesOptions, PolicySetCreateOptions, - PolicySetList, PolicySetListOptions, PolicySetReadOptions, PolicySetRemovePoliciesOptions, @@ -38,47 +39,41 @@ class PolicySets(_Service): def list( self, organization: str, options: PolicySetListOptions | None = None - ) -> PolicySetList: - """List all the policy sets of the given organization.""" + ) -> Iterator[PolicySet]: + """Iterate all the policy sets of the given organization.""" if not valid_string_id(organization): raise InvalidOrgError() + + # Build params from options but do not pass page[number] — let _list handle pagination. params = options.model_dump(by_alias=True, exclude_none=True) if options else {} - r = self.t.request( - "GET", - f"/api/v2/organizations/{organization}/policy-sets", - 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["organization"] = d.get("relationships", {}).get("organization", {}) - attrs["workspace_exclusions"] = ( - d.get("relationships", {}) - .get("workspace-exclusions", {}) - .get("data", []) - ) - attrs["workspaces"] = ( - d.get("relationships", {}).get("workspaces", {}).get("data", []) - ) - attrs["projects"] = ( - d.get("relationships", {}).get("projects", {}).get("data", []) - ) - attrs["policies"] = ( - d.get("relationships", {}).get("policies", {}).get("data", []) - ) - items.append(PolicySet.model_validate(attrs)) - return PolicySetList( - 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"), - ) + params.pop("page[number]", None) + + path = f"/api/v2/organizations/{organization}/policy-sets" + + def _gen() -> Iterator[PolicySet]: + for d in self._list(path, params=params): + attrs = d.get("attributes", {}) + attrs["id"] = d.get("id") + attrs["organization"] = d.get("relationships", {}).get( + "organization", {} + ) + attrs["workspace_exclusions"] = ( + d.get("relationships", {}) + .get("workspace-exclusions", {}) + .get("data", []) + ) + attrs["workspaces"] = ( + d.get("relationships", {}).get("workspaces", {}).get("data", []) + ) + attrs["projects"] = ( + d.get("relationships", {}).get("projects", {}).get("data", []) + ) + attrs["policies"] = ( + d.get("relationships", {}).get("policies", {}).get("data", []) + ) + yield PolicySet.model_validate(attrs) + + return _gen() def create(self, organization: str, options: PolicySetCreateOptions) -> PolicySet: """Create a new policy set in the given organization.""" diff --git a/src/pytfe/resources/run.py b/src/pytfe/resources/run.py index 49efdbc3..4cb430c4 100644 --- a/src/pytfe/resources/run.py +++ b/src/pytfe/resources/run.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from ..errors import ( @@ -10,14 +11,12 @@ TerraformVersionValidForPlanOnlyError, ) from ..models.run import ( - OrganizationRunList, Run, RunApplyOptions, RunCancelOptions, RunCreateOptions, RunDiscardOptions, RunForceCancelOptions, - RunList, RunListForOrganizationOptions, RunListOptions, RunReadOptions, @@ -27,63 +26,33 @@ class Runs(_Service): - def list(self, workspace_id: str, options: RunListOptions | None = None) -> RunList: + def list( + self, workspace_id: str, options: RunListOptions | None = None + ) -> Iterator[Run]: """List all the runs of the given workspace.""" if not valid_string_id(workspace_id): raise InvalidWorkspaceIDError() - params = ( - options.model_dump(by_alias=True, exclude_none=True) if options else None - ) - r = self.t.request( - "GET", - f"/api/v2/workspaces/{workspace_id}/runs", - 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") - items.append(Run.model_validate(attrs)) - return RunList( - 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"), - ) + params = options.model_dump(by_alias=True) if options else {} + path = f"/api/v2/workspaces/{workspace_id}/runs" + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + yield Run.model_validate(attrs) def list_for_organization( self, organization: str, options: RunListForOrganizationOptions | None = None - ) -> OrganizationRunList: + ) -> Iterator[Run]: """List all the runs of the given organization.""" if not valid_string_id(organization): raise InvalidOrgError() - params = ( - options.model_dump(by_alias=True, exclude_none=True) if options else None - ) - r = self.t.request( - "GET", - f"/api/v2/organizations/{organization}/runs", - 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") - items.append(Run.model_validate(attrs)) - return OrganizationRunList( - items=items, - current_page=pagination.get("current-page"), - prev_page=pagination.get("prev-page"), - next_page=pagination.get("next-page"), - ) + path = f"/api/v2/organizations/{organization}/runs" + params = options.model_dump(by_alias=True, exclude_none=True) if options else {} + # meta = jd.get("meta", {}) + # pagination = meta.get("pagination", {}) + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + yield Run.model_validate(attrs) def create(self, options: RunCreateOptions) -> Run: """Create a new run for the given workspace.""" diff --git a/src/pytfe/resources/run_event.py b/src/pytfe/resources/run_event.py index fb5479f4..7b8ce124 100644 --- a/src/pytfe/resources/run_event.py +++ b/src/pytfe/resources/run_event.py @@ -1,11 +1,11 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from ..errors import InvalidRunEventIDError, InvalidRunIDError from ..models.run_event import ( RunEvent, - RunEventList, RunEventListOptions, RunEventReadOptions, ) @@ -16,34 +16,18 @@ class RunEvents(_Service): def list( self, run_id: str, options: RunEventListOptions | None = None - ) -> RunEventList: + ) -> Iterator[RunEvent]: """List all the run events of the given run.""" if not valid_string_id(run_id): raise InvalidRunIDError() params: dict[str, Any] = {} if options and options.include: params["include"] = ",".join(options.include) - r = self.t.request( - "GET", - f"/api/v2/runs/{run_id}/run-events", - 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") - items.append(RunEvent.model_validate(attrs)) - return RunEventList( - 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}/run-events" + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + yield RunEvent.model_validate(attrs) def read(self, run_event_id: str) -> RunEvent: """Read a specific run event by its ID.""" diff --git a/src/pytfe/resources/user.py b/src/pytfe/resources/user.py new file mode 100644 index 00000000..0901ef60 --- /dev/null +++ b/src/pytfe/resources/user.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from ..models.user import User +from ..utils import valid_string_id +from ._base import _Service + + +class Users(_Service): + def read(self, user_id: str) -> User: + if not valid_string_id(user_id): + raise ValueError("invalid user id") + + r = self.t.request("GET", f"/api/v2/users/{user_id}") + d = r.json()["data"] + attr = d.get("attributes", {}) or {} + user_data = dict(attr) + user_data["id"] = d.get("id") + return User(**user_data) diff --git a/tests/units/test_policy.py b/tests/units/test_policy.py index af4790c8..ea335f96 100644 --- a/tests/units/test_policy.py +++ b/tests/units/test_policy.py @@ -14,7 +14,6 @@ EnforcementLevel, Policy, PolicyCreateOptions, - PolicyList, PolicyUpdateOptions, ) from pytfe.models.policy_set import PolicyKind @@ -84,22 +83,16 @@ def test_list_policies_success_without_options( mock_response.json.return_value = mock_response_data mock_transport.request.return_value = mock_response - result = policies_service.list("org-123") + result_iter = policies_service.list("org-123") + items = list(result_iter) - mock_transport.request.assert_called_once_with( - "GET", "/api/v2/organizations/org-123/policies", params=None - ) + assert mock_transport.request.called - assert isinstance(result, PolicyList) - assert len(result.items) == 1 - assert result.items[0].id == "pol-123" - assert result.items[0].name == "test-policy" - assert result.items[0].kind == PolicyKind.SENTINEL - assert ( - result.items[0].enforcement_level == EnforcementLevel.ENFORCEMENT_ADVISORY - ) - assert result.current_page == 1 - assert result.total_count == 1 + assert len(items) == 1 + assert items[0].id == "pol-123" + assert items[0].name == "test-policy" + assert items[0].kind == PolicyKind.SENTINEL + assert items[0].enforcement_level == EnforcementLevel.ENFORCEMENT_ADVISORY def test_create_policy_validations(self, policies_service): """Test create method validations.""" diff --git a/tests/units/test_policy_set.py b/tests/units/test_policy_set.py new file mode 100644 index 00000000..91229c4c --- /dev/null +++ b/tests/units/test_policy_set.py @@ -0,0 +1,327 @@ +"""Unit tests for the PolicySets resource.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + InvalidNameError, + InvalidOrgError, + InvalidPolicySetIDError, + RequiredNameError, +) +from pytfe.models.policy_set import ( + PolicySet, + PolicySetCreateOptions, + PolicySetListOptions, + PolicySetReadOptions, + PolicySetUpdateOptions, +) +from pytfe.models.policy_types import PolicyKind +from pytfe.resources.policy_set import PolicySets + + +class TestPolicySets: + """Test the PolicySets service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def service(self, mock_transport): + """Create a PolicySets service with mocked transport.""" + return PolicySets(mock_transport) + + # ────────────────────────────────────────────────────────────────────────── + # Helpers + # ────────────────────────────────────────────────────────────────────────── + + @staticmethod + def _policy_set_data( + ps_id: str = "ps-abc123", + name: str = "example-policy-set", + kind: str = "sentinel", + ) -> dict: + """Minimal JSON:API policy-set dict as returned by the API.""" + return { + "id": ps_id, + "type": "policy-sets", + "attributes": { + "name": name, + "description": "A test policy set", + "kind": kind, + "global": False, + "overridable": False, + "agent-enabled": False, + "policy-count": 0, + "workspace-count": 0, + "project-count": 0, + "policy-tool-version": None, + "policies-path": None, + "created-at": "2024-01-01T00:00:00Z", + "updated-at": "2024-01-01T00:00:00Z", + }, + "relationships": { + "organization": {"data": {"id": "org-test", "type": "organizations"}}, + "workspaces": {"data": []}, + "projects": {"data": []}, + "policies": {"data": []}, + "workspace-exclusions": {"data": []}, + }, + } + + # ────────────────────────────────────────────────────────────────────────── + # list() + # ────────────────────────────────────────────────────────────────────────── + + def test_list_invalid_org_empty_string(self, service): + """list() raises InvalidOrgError for an empty organization.""" + with pytest.raises(InvalidOrgError): + list(service.list("")) + + def test_list_invalid_org_none(self, service): + """list() raises InvalidOrgError for None organization.""" + with pytest.raises(InvalidOrgError): + list(service.list(None)) + + def test_list_returns_iterator_of_policy_sets(self, service): + """list() returns an iterator that yields PolicySet objects.""" + raw = [ + self._policy_set_data("ps-1", "ps-one"), + self._policy_set_data("ps-2", "ps-two"), + ] + service._list = Mock(return_value=raw) + + result = list(service.list("my-org")) + + assert len(result) == 2 + assert all(isinstance(ps, PolicySet) for ps in result) + assert result[0].id == "ps-1" + assert result[0].name == "ps-one" + assert result[1].id == "ps-2" + assert result[1].name == "ps-two" + + def test_list_hits_correct_endpoint(self, service): + """list() calls _list with the correct path.""" + service._list = Mock(return_value=[]) + + list(service.list("my-org")) + + service._list.assert_called_once() + call_path = service._list.call_args[0][0] + assert call_path == "/api/v2/organizations/my-org/policy-sets" + + def test_list_with_search_option_passes_param(self, service): + """list() with a search option passes the correct params to _list.""" + service._list = Mock(return_value=[]) + options = PolicySetListOptions(search="my-prefix") + + list(service.list("my-org", options)) + + service._list.assert_called_once() + call_kwargs = service._list.call_args[1] + assert call_kwargs.get("params", {}).get("search[name]") == "my-prefix" + + def test_list_with_kind_filter(self, service): + """list() with a kind filter passes filter[kind] param.""" + service._list = Mock(return_value=[]) + options = PolicySetListOptions(kind=PolicyKind.OPA) + + list(service.list("my-org", options)) + + service._list.assert_called_once() + params = service._list.call_args[1].get("params", {}) + assert params.get("filter[kind]") == PolicyKind.OPA + + def test_list_page_number_stripped_from_params(self, service): + """list() strips page[number] from params so _list handles pagination.""" + service._list = Mock(return_value=[]) + options = PolicySetListOptions(page_number=3, page_size=20) + + list(service.list("my-org", options)) + + params = service._list.call_args[1].get("params", {}) + assert "page[number]" not in params + assert params.get("page[size]") == 20 + + # ────────────────────────────────────────────────────────────────────────── + # read() + # ────────────────────────────────────────────────────────────────────────── + + def test_read_invalid_id(self, service): + """read() raises InvalidPolicySetIDError for an invalid ID.""" + with pytest.raises(InvalidPolicySetIDError): + service.read("") + + with pytest.raises(InvalidPolicySetIDError): + service.read(None) + + def test_read_hits_correct_endpoint(self, service, mock_transport): + """read() calls GET /api/v2/policy-sets/{id}.""" + mock_response = Mock() + mock_response.json.return_value = {"data": self._policy_set_data("ps-abc123")} + mock_transport.request.return_value = mock_response + + service.read("ps-abc123") + + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/policy-sets/ps-abc123", + params=None, + ) + + def test_read_returns_policy_set(self, service, mock_transport): + """read() parses and returns a PolicySet model.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": self._policy_set_data("ps-abc123", "my-ps", "sentinel") + } + mock_transport.request.return_value = mock_response + + result = service.read("ps-abc123") + + assert isinstance(result, PolicySet) + assert result.id == "ps-abc123" + assert result.name == "my-ps" + assert result.kind == PolicyKind.SENTINEL + + def test_read_with_options_passes_include_param(self, service, mock_transport): + """read_with_options() passes include param when provided.""" + from pytfe.models.policy_set import PolicySetIncludeOpt + + mock_response = Mock() + mock_response.json.return_value = {"data": self._policy_set_data("ps-xyz")} + mock_transport.request.return_value = mock_response + + options = PolicySetReadOptions( + include=[PolicySetIncludeOpt.POLICY_SET_POLICIES] + ) + service.read_with_options("ps-xyz", options) + + call_kwargs = mock_transport.request.call_args[1] + assert call_kwargs.get("params") is not None + + # ────────────────────────────────────────────────────────────────────────── + # create() + # ────────────────────────────────────────────────────────────────────────── + + def test_create_invalid_org(self, service): + """create() raises InvalidOrgError for an invalid organization.""" + options = PolicySetCreateOptions(name="valid-name") + with pytest.raises(InvalidOrgError): + service.create("", options) + + def test_create_missing_name(self, service): + """create() raises RequiredNameError when name is empty.""" + options = PolicySetCreateOptions(name="") + with pytest.raises((RequiredNameError, InvalidNameError)): + service.create("my-org", options) + + def test_create_success(self, service, mock_transport): + """create() POSTs to the correct endpoint and returns a PolicySet.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": self._policy_set_data("ps-new", "new-policy-set") + } + mock_transport.request.return_value = mock_response + + options = PolicySetCreateOptions(name="new-policy-set") + result = service.create("my-org", options) + + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "POST" + assert call_args[0][1] == "/api/v2/organizations/my-org/policy-sets" + + assert isinstance(result, PolicySet) + assert result.id == "ps-new" + assert result.name == "new-policy-set" + + def test_create_payload_shape(self, service, mock_transport): + """create() sends a correctly shaped JSON:API payload.""" + mock_response = Mock() + mock_response.json.return_value = {"data": self._policy_set_data("ps-123")} + mock_transport.request.return_value = mock_response + + options = PolicySetCreateOptions(name="shaped-ps", kind=PolicyKind.OPA) + service.create("my-org", options) + + payload = mock_transport.request.call_args[1]["json_body"] + assert "data" in payload + data = payload["data"] + assert data["type"] == "policy-sets" + assert "attributes" in data + assert data["attributes"]["name"] == "shaped-ps" + + # ────────────────────────────────────────────────────────────────────────── + # update() + # ────────────────────────────────────────────────────────────────────────── + + def test_update_invalid_id(self, service): + """update() raises InvalidPolicySetIDError for an invalid ID.""" + options = PolicySetUpdateOptions(name="new-name") + with pytest.raises(InvalidPolicySetIDError): + service.update("", options) + + def test_update_success(self, service, mock_transport): + """update() PATCHes the correct endpoint and returns a PolicySet.""" + mock_response = Mock() + mock_response.json.return_value = { + "data": self._policy_set_data("ps-abc123", "updated-name") + } + mock_transport.request.return_value = mock_response + + options = PolicySetUpdateOptions(name="updated-name") + result = service.update("ps-abc123", options) + + call_args = mock_transport.request.call_args + assert call_args[0][0] == "PATCH" + assert call_args[0][1] == "/api/v2/policy-sets/ps-abc123" + + payload = call_args[1]["json_body"] + assert payload["data"]["type"] == "policy-sets" + assert payload["data"]["id"] == "ps-abc123" + assert payload["data"]["attributes"]["name"] == "updated-name" + + assert isinstance(result, PolicySet) + assert result.name == "updated-name" + + def test_update_no_attributes_raises(self, service): + """update() raises ValueError when no attributes are provided.""" + options = PolicySetUpdateOptions() # all None + with pytest.raises(ValueError): + service.update("ps-abc123", options) + + # ────────────────────────────────────────────────────────────────────────── + # delete() + # ────────────────────────────────────────────────────────────────────────── + + def test_delete_invalid_id(self, service): + """delete() raises InvalidPolicySetIDError for an invalid ID.""" + with pytest.raises(InvalidPolicySetIDError): + service.delete("") + + with pytest.raises(InvalidPolicySetIDError): + service.delete(None) + + def test_delete_hits_correct_endpoint(self, service, mock_transport): + """delete() calls DELETE /api/v2/policy-sets/{id}.""" + mock_transport.request.return_value = Mock() + + service.delete("ps-abc123") + + mock_transport.request.assert_called_once_with( + "DELETE", + "/api/v2/policy-sets/ps-abc123", + ) + + def test_delete_returns_none(self, service, mock_transport): + """delete() returns None on success.""" + mock_transport.request.return_value = Mock() + + result = service.delete("ps-abc123") + + assert result is None diff --git a/tests/units/test_run.py b/tests/units/test_run.py index bce2d2a8..c1a9bf67 100644 --- a/tests/units/test_run.py +++ b/tests/units/test_run.py @@ -11,7 +11,6 @@ TerraformVersionValidForPlanOnlyError, ) from pytfe.models.run import ( - OrganizationRunList, Run, RunApplyOptions, RunCancelOptions, @@ -19,7 +18,6 @@ RunDiscardOptions, RunForceCancelOptions, RunIncludeOpt, - RunList, RunListForOrganizationOptions, RunListOptions, RunReadOptions, @@ -47,74 +45,59 @@ def runs_service(self, mock_transport): def test_list_runs_success(self, runs_service): """Test successful list operation.""" - mock_response_data = { - "data": [ - { - "id": "run-123", - "attributes": { - "status": "applied", - "source": "tfe-configuration-version", - "message": "Test run", - "created-at": "2023-01-01T12:00:00Z", - "has-changes": True, - "is-destroy": False, - "auto-apply": False, - "plan-only": False, - }, + mock_list_data = [ + { + "id": "run-123", + "attributes": { + "status": "applied", + "source": "tfe-configuration-version", + "message": "Test run", + "created-at": "2023-01-01T12:00:00Z", + "has-changes": True, + "is-destroy": False, + "auto-apply": False, + "plan-only": False, }, - { - "id": "run-456", - "attributes": { - "status": "planned", - "source": "tfe-ui", - "message": "Another test run", - "created-at": "2023-01-02T14:00:00Z", - "has-changes": False, - "is-destroy": True, - "auto-apply": True, - "plan-only": True, - }, + }, + { + "id": "run-456", + "attributes": { + "status": "planned", + "source": "tfe-ui", + "message": "Another test run", + "created-at": "2023-01-02T14:00:00Z", + "has-changes": False, + "is-destroy": True, + "auto-apply": True, + "plan-only": True, }, - ], - "meta": { - "pagination": { - "current-page": 1, - "total-pages": 2, - "prev-page": None, - "next-page": 2, - "total-count": 10, - } }, - } - - mock_response = Mock() - mock_response.json.return_value = mock_response_data + ] - with patch.object(runs_service, "t") as mock_transport: - mock_transport.request.return_value = mock_response + with patch.object(runs_service, "_list") as mock_list: + mock_list.return_value = mock_list_data - # Test with custom page_size - use a print statement to debug what's actually sent + # Test with options options = RunListOptions(page_number=1, page_size=5) - result = runs_service.list("ws-123", options) + result = list(runs_service.list("ws-123", options)) - # Check what was actually called - call_args = mock_transport.request.call_args - actual_params = call_args[1]["params"] - - # Verify the basic structure - assert call_args[0][0] == "GET" - assert call_args[0][1] == "/api/v2/workspaces/ws-123/runs" - assert actual_params["page[number]"] == 1 - - # Verify result structure - assert isinstance(result, RunList) - assert len(result.items) == 2 - assert result.current_page == 1 - assert result.total_pages == 2 - assert result.total_count == 10 - - # Verify run objects - run1 = result.items[0] + # Verify _list was called with correct path + assert mock_list.call_count == 1 + call_args = mock_list.call_args + assert call_args[0][0] == "/api/v2/workspaces/ws-123/runs" + + # Verify params structure includes pagination and options + params = call_args[1]["params"] + assert "page[number]" in params + assert "page[size]" in params + assert "include" in params + + # Verify result structure - iterator yields Run objects + assert len(result) == 2 + + # Verify run objects were created correctly from response data + run1 = result[0] + assert isinstance(run1, Run) assert run1.id == "run-123" assert run1.status == RunStatus.Run_Applied assert run1.source == RunSource.Run_Source_Configuration_Version @@ -122,64 +105,53 @@ def test_list_runs_success(self, runs_service): assert run1.has_changes is True assert run1.is_destroy is False - run2 = result.items[1] + run2 = result[1] + assert isinstance(run2, Run) assert run2.id == "run-456" assert run2.status == RunStatus.Run_Planned assert run2.source == RunSource.Run_Source_UI + assert run2.message == "Another test run" assert run2.has_changes is False assert run2.is_destroy is True def test_list_for_organization_success(self, runs_service): """Test successful list_for_organization operation.""" - mock_response_data = { - "data": [ - { - "id": "run-org-1", - "attributes": { - "status": "applied", - "source": "tfe-api", - "message": "Organization run", - "created-at": "2023-01-01T12:00:00Z", - "has-changes": True, - "is-destroy": False, - }, - } - ], - "meta": { - "pagination": { - "current-page": 1, - "prev-page": None, - "next-page": None, - } - }, - } - - mock_response = Mock() - mock_response.json.return_value = mock_response_data + mock_response_data = [ + { + "id": "run-org-1", + "attributes": { + "status": "applied", + "source": "tfe-api", + "message": "Organization run", + "created-at": "2023-01-01T12:00:00Z", + "has-changes": True, + "is-destroy": False, + }, + } + ] - with patch.object(runs_service, "t") as mock_transport: - mock_transport.request.return_value = mock_response + with patch.object(runs_service, "_list") as mock_list: + mock_list.return_value = mock_response_data options = RunListForOrganizationOptions(status="applied,planned") - result = runs_service.list_for_organization("test-org", options) + result = list(runs_service.list_for_organization("test-org", options)) - # Verify request was made correctly (account for defaults and aliases) + # Verify _list was called with correct path and params expected_params = { "page[number]": 1, "page[size]": 20, "filter[status]": "applied,planned", "include": [], } - mock_transport.request.assert_called_once_with( - "GET", "/api/v2/organizations/test-org/runs", params=expected_params + mock_list.assert_called_once_with( + "/api/v2/organizations/test-org/runs", params=expected_params ) - # Verify result structure - assert isinstance(result, OrganizationRunList) - assert len(result.items) == 1 - assert result.current_page == 1 - assert result.items[0].id == "run-org-1" + # Verify result structure - now returns list of Run objects + assert len(result) == 1 + assert result[0].id == "run-org-1" + assert result[0].status == RunStatus.Run_Applied def test_create_run_validation_errors(self, runs_service): """Test create method with validation errors.""" diff --git a/tests/units/test_run_events.py b/tests/units/test_run_events.py new file mode 100644 index 00000000..af82e786 --- /dev/null +++ b/tests/units/test_run_events.py @@ -0,0 +1,298 @@ +"""Unit tests for the run_events module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidRunEventIDError, InvalidRunIDError +from pytfe.models.run_event import ( + RunEvent, + RunEventIncludeOpt, + RunEventListOptions, + RunEventReadOptions, +) +from pytfe.resources.run_event import RunEvents + + +class TestRunEvents: + """Test the RunEvents service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def run_events_service(self, mock_transport): + """Create a RunEvents service with mocked transport.""" + return RunEvents(mock_transport) + + def test_list_run_events_success(self, run_events_service): + """Test successful list operation using iterator pattern.""" + + # Mock data for run events + mock_data = [ + { + "id": "re-123", + "attributes": { + "action": "queued", + "description": "Run queued", + "created-at": "2023-01-01T12:00:00Z", + }, + }, + { + "id": "re-456", + "attributes": { + "action": "planning", + "description": "Planning started", + "created-at": "2023-01-01T12:01:00Z", + }, + }, + { + "id": "re-789", + "attributes": { + "action": "planned", + "description": "Planning finished", + "created-at": "2023-01-01T12:02:00Z", + }, + }, + ] + + with patch.object(run_events_service, "_list") as mock_list: + # Mock _list to return an iterator + mock_list.return_value = iter(mock_data) + + options = RunEventListOptions(include=[RunEventIncludeOpt.RUN_EVENT_ACTOR]) + results = list(run_events_service.list("run-123", options)) + + # Verify _list was called correctly + mock_list.assert_called_once_with( + "/api/v2/runs/run-123/run-events", + params={"include": "actor"}, + ) + + # Verify results + assert len(results) == 3 + assert isinstance(results[0], RunEvent) + assert results[0].id == "re-123" + assert results[0].action == "queued" + assert results[1].id == "re-456" + assert results[1].action == "planning" + assert results[2].id == "re-789" + assert results[2].action == "planned" + + def test_list_run_events_with_multiple_includes(self, run_events_service): + """Test list with multiple include options.""" + + mock_data = [ + { + "id": "re-111", + "attributes": { + "action": "apply-queued", + "description": "Apply queued", + "created-at": "2023-01-01T12:10:00Z", + }, + }, + ] + + with patch.object(run_events_service, "_list") as mock_list: + mock_list.return_value = iter(mock_data) + + options = RunEventListOptions( + include=[ + RunEventIncludeOpt.RUN_EVENT_ACTOR, + RunEventIncludeOpt.RUN_EVENT_COMMENT, + ] + ) + results = list(run_events_service.list("run-456", options)) + + # Verify include parameter is formatted correctly + mock_list.assert_called_once_with( + "/api/v2/runs/run-456/run-events", + params={"include": "actor,comment"}, + ) + + assert len(results) == 1 + assert results[0].id == "re-111" + + def test_list_run_events_no_options(self, run_events_service): + """Test list without include options.""" + + mock_data = [ + { + "id": "re-222", + "attributes": { + "action": "apply-finished", + "created-at": "2023-01-01T12:15:00Z", + }, + }, + ] + + with patch.object(run_events_service, "_list") as mock_list: + mock_list.return_value = iter(mock_data) + + results = list(run_events_service.list("run-789")) + + # Verify _list was called with empty params + mock_list.assert_called_once_with( + "/api/v2/runs/run-789/run-events", + params={}, + ) + + assert len(results) == 1 + assert results[0].id == "re-222" + + def test_list_run_events_empty_result(self, run_events_service): + """Test list with no run events returned.""" + + with patch.object(run_events_service, "_list") as mock_list: + mock_list.return_value = iter([]) + + results = list(run_events_service.list("run-empty")) + + assert len(results) == 0 + + def test_list_run_events_invalid_run_id(self, run_events_service): + """Test list with invalid run ID.""" + + with pytest.raises(InvalidRunIDError): + list(run_events_service.list("")) + + with pytest.raises(InvalidRunIDError): + list(run_events_service.list("run/invalid")) + + def test_read_run_event_success(self, run_events_service): + """Test successful read operation.""" + + mock_response_data = { + "data": { + "id": "re-read-123", + "attributes": { + "action": "planned", + "description": "Run planned successfully", + "created-at": "2023-01-01T13:00:00Z", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(run_events_service, "t") as mock_transport: + mock_transport.request.return_value = mock_response + + result = run_events_service.read("re-read-123") + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/run-events/re-read-123", + params={}, + ) + + # Verify result + assert isinstance(result, RunEvent) + assert result.id == "re-read-123" + assert result.action == "planned" + assert result.description == "Run planned successfully" + + def test_read_run_event_with_includes(self, run_events_service): + """Test read with include options.""" + + mock_response_data = { + "data": { + "id": "re-read-456", + "attributes": { + "action": "discarded", + "description": "Run discarded", + "created-at": "2023-01-01T13:05:00Z", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(run_events_service, "t") as mock_transport: + mock_transport.request.return_value = mock_response + + options = RunEventReadOptions(include=[RunEventIncludeOpt.RUN_EVENT_ACTOR]) + result = run_events_service.read_with_options("re-read-456", options) + + # Verify include parameter was passed + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/run-events/re-read-456", + params={"include": "actor"}, + ) + + assert result.id == "re-read-456" + assert result.action == "discarded" + + def test_read_run_event_invalid_id(self, run_events_service): + """Test read with invalid run event ID.""" + + with pytest.raises(InvalidRunEventIDError): + run_events_service.read("") + + with pytest.raises(InvalidRunEventIDError): + run_events_service.read("re/invalid") + + def test_read_vs_read_with_options(self, run_events_service): + """Test that read() delegates to read_with_options().""" + + mock_response_data = { + "data": { + "id": "re-read-789", + "attributes": { + "action": "completed", + "created-at": "2023-01-01T13:10:00Z", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(run_events_service, "t") as mock_transport: + mock_transport.request.return_value = mock_response + + result1 = run_events_service.read("re-read-789") + + # Reset mock + mock_transport.reset_mock() + mock_transport.request.return_value = mock_response + + result2 = run_events_service.read_with_options("re-read-789") + + # Both should produce the same result + assert result1.id == result2.id + assert result1.action == result2.action + + def test_list_run_events_iterator_lazy_loading(self, run_events_service): + """Test that list returns an iterator that lazily loads data.""" + + mock_data = [ + { + "id": "re-lazy-1", + "attributes": { + "action": "queued", + "created-at": "2023-01-01T12:00:00Z", + }, + }, + ] + + with patch.object(run_events_service, "_list") as mock_list: + mock_list.return_value = iter(mock_data) + + # Get the iterator without consuming it yet + iterator = run_events_service.list("run-lazy") + + # _list should not have been called yet (iterator not consumed) + # This test ensures lazy evaluation + first_event = next(iterator) + + # Now _list should have been called + mock_list.assert_called_once() + assert first_event.id == "re-lazy-1" diff --git a/tests/units/test_user.py b/tests/units/test_user.py new file mode 100644 index 00000000..6df2aa48 --- /dev/null +++ b/tests/units/test_user.py @@ -0,0 +1,74 @@ +"""Unit tests for the Users resource.""" + +from unittest.mock import Mock + +import pytest + +from pytfe.models.user import User +from pytfe.resources.user import Users + + +class TestUsers: + """Test suite for user resource operations.""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + return Mock() + + @pytest.fixture + def users_service(self, mock_transport): + """Create users service with mocked transport.""" + return Users(mock_transport) + + @pytest.fixture + def sample_user_response(self): + """Sample JSON:API response for a user.""" + return { + "data": { + "id": "user-MA4GL63FmYRpSFxa", + "type": "users", + "attributes": { + "username": "admin", + "email": "admin@example.com", + "is-service-account": False, + "auth-method": "hcp_sso", + "avatar-url": "https://example.com/avatar.png", + "v2-only": True, + "permissions": { + "can-create-organizations": False, + "can-change-email": True, + "can-change-username": True, + }, + }, + } + } + + def test_read_user(self, users_service, mock_transport, sample_user_response): + """Test reading a specific user by ID.""" + mock_transport.request.return_value.json.return_value = sample_user_response + + user_id = "user-MA4GL63FmYRpSFxa" + user = users_service.read(user_id) + + mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/users/{user_id}" + ) + assert isinstance(user, User) + assert user.id == user_id + assert user.username == "admin" + assert user.email == "admin@example.com" + assert user.is_service_account is False + assert user.auth_method == "hcp_sso" + assert user.avatar_url == "https://example.com/avatar.png" + assert user.v2_only is True + assert user.permissions == { + "can-create-organizations": False, + "can-change-email": True, + "can-change-username": True, + } + + def test_read_user_invalid_id(self, users_service): + """Test reading a user with an invalid user ID.""" + with pytest.raises(ValueError, match="invalid user id"): + users_service.read("")