diff --git a/examples/team_token.py b/examples/team_token.py new file mode 100644 index 00000000..5c71bfd1 --- /dev/null +++ b/examples/team_token.py @@ -0,0 +1,149 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import TeamTokenCreateOptions, TeamTokenListOptions + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def _print_token(token): + print(f"- ID: {token.id}") + if token.description: + print(f" Description: {token.description}") + print(f" Created At: {token.created_at}") + print(f" Last Used At: {token.last_used_at}") + print(f" Expired At: {token.expired_at}") + if token.team: + print(f" Team ID: {token.team.id}") + if token.created_by: + if token.created_by.user: + print(f" Created By (user): {token.created_by.user.id}") + elif token.created_by.team: + print(f" Created By (team): {token.created_by.team.id}") + elif token.created_by.organization: + print(f" Created By (org): {token.created_by.organization.id}") + print() + + +def main(): + parser = argparse.ArgumentParser(description="Team Tokens demo for python-tfe SDK") + parser.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + parser.add_argument("--organization", required=True, help="Organization name") + parser.add_argument("--team-id", help="Team ID (e.g. team-xxxxx)") + parser.add_argument("--create", action="store_true", help="Create a team token") + parser.add_argument( + "--description", help="Token description (creates a named multi-token)" + ) + parser.add_argument( + "--expired-at", + help="Expiry datetime in ISO 8601 (e.g. 2026-12-31T00:00:00Z)", + ) + parser.add_argument( + "--read", action="store_true", help="Read the legacy token for --team-id" + ) + parser.add_argument( + "--read-by-id", action="store_true", help="Read a token by --token-id" + ) + parser.add_argument("--token-id", help="Token ID (e.g. at-xxxxx)") + parser.add_argument( + "--delete", action="store_true", help="Delete the legacy token for --team-id" + ) + parser.add_argument( + "--delete-by-id", + action="store_true", + help="Delete a token by --token-id", + ) + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) Always list tokens for the organization + _print_header(f"Listing team tokens for organization: {args.organization}") + list_opts = TeamTokenListOptions() + token_count = 0 + for t in client.team_tokens.list(organization=args.organization, options=list_opts): + token_count += 1 + _print_token(t) + + if token_count == 0: + print("No team tokens found.") + else: + print(f"Total: {token_count} team tokens") + + # 2) Create a team token + if args.create: + if not args.team_id: + print("--team-id is required for --create") + else: + from datetime import datetime + + if args.description or args.expired_at: + _print_header(f"Creating named team token for team: {args.team_id}") + create_opts = TeamTokenCreateOptions( + description=args.description, + expired_at=datetime.fromisoformat(args.expired_at) + if args.expired_at + else None, + ) + t = client.team_tokens.create_with_options( + team_id=args.team_id, options=create_opts + ) + else: + _print_header(f"Creating legacy team token for team: {args.team_id}") + t = client.team_tokens.create(team_id=args.team_id) + print("Created team token:") + _print_token(t) + + # 3) Read legacy token by team ID + if args.read: + if not args.team_id: + print("--team-id is required for --read") + else: + _print_header(f"Reading legacy token for team: {args.team_id}") + t = client.team_tokens.read(team_id=args.team_id) + _print_token(t) + + # 4) Read token by token ID + if args.read_by_id: + if not args.token_id: + print("--token-id is required for --read-by-id") + else: + _print_header(f"Reading token by ID: {args.token_id}") + t = client.team_tokens.read_by_id(token_id=args.token_id) + _print_token(t) + + # 5) Delete legacy token by team ID + if args.delete: + if not args.team_id: + print("--team-id is required for --delete") + else: + _print_header(f"Deleting legacy token for team: {args.team_id}") + client.team_tokens.delete(team_id=args.team_id) + print("Deleted.") + + # 6) Delete token by token ID + if args.delete_by_id: + if not args.token_id: + print("--token-id is required for --delete-by-id") + else: + _print_header(f"Deleting token by ID: {args.token_id}") + client.team_tokens.delete_by_id(token_id=args.token_id) + print("Deleted.") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 4aead338..1e71bdee 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -43,6 +43,7 @@ from .resources.state_versions import StateVersions from .resources.team import Teams from .resources.team_project_access import TeamProjectAccesses +from .resources.team_token import TeamTokens from .resources.user import Users from .resources.variable import Variables from .resources.variable_sets import VariableSets, VariableSetVariables @@ -125,6 +126,7 @@ def __init__(self, config: TFEConfig | None = None): # Team project access self.teams = Teams(self._transport) self.team_project_accesses = TeamProjectAccesses(self._transport) + self.team_tokens = TeamTokens(self._transport) # Reserved Tag Key self.reserved_tag_key = ReservedTagKeys(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index 1209a552..efc6305f 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -664,3 +664,11 @@ class RequiredCommentBodyError(TFEError): def __init__(self, message: str = "comment body is required"): super().__init__(message) + + +# Team Token errors +class InvalidTokenIDError(InvalidValues): + """Raised when an invalid authentication token ID is provided.""" + + def __init__(self, message: str = "invalid value for token ID"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index fd263c1d..764c0dab 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -358,6 +358,12 @@ TeamPermissions, TeamUpdateOptions, ) +from .team_token import ( + CreatedByChoice, + TeamToken, + TeamTokenCreateOptions, + TeamTokenListOptions, +) # Variables from .variable import ( @@ -588,6 +594,11 @@ "TeamIncludeOpt", "TeamListOptions", "TeamUpdateOptions", + # Team Tokens + "CreatedByChoice", + "TeamToken", + "TeamTokenCreateOptions", + "TeamTokenListOptions", "Project", "ProjectAddTagBindingsOptions", "ProjectCreateOptions", diff --git a/src/pytfe/models/team_token.py b/src/pytfe/models/team_token.py new file mode 100644 index 00000000..b8d23df7 --- /dev/null +++ b/src/pytfe/models/team_token.py @@ -0,0 +1,56 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + +from .organization import Organization +from .team import Team +from .user import User + + +class TeamToken(BaseModel): + """TeamToken represents a Terraform Enterprise team token.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + created_at: datetime | None = Field(default=None, alias="created-at") + last_used_at: datetime | None = Field(default=None, alias="last-used-at") + description: str | None = Field(default=None, alias="description") + token: str | None = Field(default=None, alias="token") + expired_at: datetime | None = Field(default=None, alias="expired-at") + + # Relations + team: Team | None = None + created_by: CreatedByChoice | None = Field(default=None, alias="created-by") + + +class TeamTokenCreateOptions(BaseModel): + """TeamTokenCreateOptions contains the options for creating a team token.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + description: str | None = Field(default=None, alias="description") + expired_at: datetime | None = Field(default=None, alias="expired-at") + + +class TeamTokenListOptions(BaseModel): + """TeamTokenListOptions contains the options for listing team tokens.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(default=None, alias="page[size]") + query: str | None = Field(default=None, alias="q") + sort: str | None = Field(default=None, alias="sort") + + +class CreatedByChoice(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + organization: Organization | None = None + user: User | None = None + team: Team | None = None diff --git a/src/pytfe/resources/team_token.py b/src/pytfe/resources/team_token.py new file mode 100644 index 00000000..fd971a68 --- /dev/null +++ b/src/pytfe/resources/team_token.py @@ -0,0 +1,151 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import InvalidOrgError, InvalidTeamIDError, InvalidTokenIDError +from ..models.organization import Organization +from ..models.team import Team +from ..models.team_token import ( + CreatedByChoice, + TeamToken, + TeamTokenCreateOptions, + TeamTokenListOptions, +) +from ..models.user import User +from ..utils import valid_string_id +from ._base import _Service + + +class TeamTokens(_Service): + """Service for managing team authentication tokens.""" + + def create(self, team_id: str) -> TeamToken: + """ + Create a new team token using the legacy creation behavior, which creates a token without a description + or regenerates the existing, descriptionless token. + """ + return self.create_with_options(team_id=team_id) + + def create_with_options( + self, + team_id: str, + options: TeamTokenCreateOptions | None = None, + ) -> TeamToken: + """ + CreateWithOptions creates a team token, with options. If no description is provided, it uses the legacy + creation behavior, which regenerates the descriptionless token if it already exists. Otherwise, it create + a new token with the given unique description, allowing for the creation of multiple team tokens. + """ + if not valid_string_id(team_id): + raise InvalidTeamIDError() + + opts = options or TeamTokenCreateOptions() + + if opts.description: + # New multi-token endpoint + path = f"/api/v2/teams/{team_id}/authentication-tokens" + payload_type = "authentication-tokens" + else: + # Legacy single-token endpoint + path = f"/api/v2/teams/{team_id}/authentication-token" + payload_type = "authentication-token" + + attributes: dict[str, Any] = opts.model_dump( + by_alias=True, + exclude_none=True, + exclude={"description"} if not opts.description else set(), + mode="json", + ) + + payload = { + "data": { + "type": payload_type, + "attributes": attributes, + } + } + r = self.t.request("POST", path=path, json_body=payload) + data = r.json().get("data", {}) + return self._team_token_from(data) + + def read(self, team_id: str) -> TeamToken: + """Read the legacy (descriptionless) team token by team ID.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + r = self.t.request("GET", path=f"/api/v2/teams/{team_id}/authentication-token") + data = r.json().get("data", {}) + return self._team_token_from(data) + + def read_by_id(self, token_id: str) -> TeamToken: + """Read a team token by its token ID.""" + if not valid_string_id(token_id): + raise InvalidTokenIDError() + r = self.t.request("GET", path=f"/api/v2/authentication-tokens/{token_id}") + data = r.json().get("data", {}) + return self._team_token_from(data) + + def list( + self, + organization: str, + options: TeamTokenListOptions | None = None, + ) -> Iterator[TeamToken]: + """List all team tokens for the given organization.""" + if not valid_string_id(organization): + raise InvalidOrgError() + path = f"/api/v2/organizations/{organization}/team-tokens" + params: dict[str, Any] = {} + if options: + if options.page_size is not None: + params["page[size]"] = options.page_size + if options.query: + params["q"] = options.query + if options.sort: + params["sort"] = options.sort + for item in self._list(path=path, params=params): + yield self._team_token_from(item) + + def delete(self, team_id: str) -> None: + """Delete the legacy team token by team ID.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + self.t.request("DELETE", path=f"/api/v2/teams/{team_id}/authentication-token") + return None + + def delete_by_id(self, token_id: str) -> None: + """Delete a team token by its token ID.""" + if not valid_string_id(token_id): + raise InvalidTokenIDError() + self.t.request("DELETE", path=f"/api/v2/authentication-tokens/{token_id}") + return None + + def _team_token_from(self, data: dict[str, Any]) -> TeamToken: + """Parse a TeamToken from API response data.""" + attrs = dict(data.get("attributes", {})) + attrs["id"] = data.get("id") + relationships = data.get("relationships", {}) + + team_data = relationships.get("team", {}).get("data") + if team_data and team_data.get("id"): + attrs["team"] = Team.model_construct( + id=team_data["id"], + ) + + created_by_data = relationships.get("created-by", {}).get("data") + if created_by_data and created_by_data.get("id"): + if created_by_data.get("type") == "users": + attrs["created-by"] = CreatedByChoice( + user=User.model_construct(id=created_by_data["id"]) + ) + elif created_by_data.get("type") == "teams": + attrs["created-by"] = CreatedByChoice( + team=Team.model_construct(id=created_by_data["id"]) + ) + elif created_by_data.get("type") == "organizations": + attrs["created-by"] = CreatedByChoice( + organization=Organization.model_construct(id=created_by_data["id"]) + ) + + return TeamToken.model_validate(attrs) diff --git a/tests/units/test_team_token.py b/tests/units/test_team_token.py new file mode 100644 index 00000000..57678671 --- /dev/null +++ b/tests/units/test_team_token.py @@ -0,0 +1,334 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the team_token module.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidOrgError, InvalidTeamIDError, InvalidTokenIDError +from pytfe.models.team import Team +from pytfe.models.team_token import ( + CreatedByChoice, + TeamToken, + TeamTokenCreateOptions, + TeamTokenListOptions, +) +from pytfe.models.user import User +from pytfe.resources.team_token import TeamTokens + + +class TestTeamTokens: + """Test the TeamTokens 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 TeamTokens service with mocked transport.""" + return TeamTokens(mock_transport) + + @pytest.fixture + def token_api_data(self): + """Typical API response for a single team token (user created-by).""" + return { + "id": "at-abc123", + "type": "authentication-tokens", + "attributes": { + "created-at": "2026-05-01T10:00:00.000Z", + "last-used-at": None, + "description": "My token", + "token": "secret-token-value", + "expired-at": "2027-05-01T10:00:00.000Z", + }, + "relationships": { + "team": {"data": {"id": "team-xyz789", "type": "teams"}}, + "created-by": {"data": {"id": "user-111", "type": "users"}}, + }, + } + + @pytest.fixture + def token_api_data_team_creator(self): + """API response where the token was created by a team.""" + return { + "id": "at-team001", + "type": "authentication-tokens", + "attributes": { + "created-at": "2026-05-01T10:00:00.000Z", + "last-used-at": None, + "description": None, + "token": None, + "expired-at": None, + }, + "relationships": { + "team": {"data": {"id": "team-xyz789", "type": "teams"}}, + "created-by": {"data": {"id": "team-abc", "type": "teams"}}, + }, + } + + # ── Model tests ────────────────────────────────────────────────────────── + + def test_team_token_model_fields(self): + """TeamToken model stores all fields.""" + t = TeamToken(id="at-abc123", description="My token", token="secret") + assert t.id == "at-abc123" + assert t.description == "My token" + assert t.token == "secret" + + def test_team_token_defaults(self): + """TeamToken optional fields default to None.""" + t = TeamToken(id="at-min") + assert t.description is None + assert t.token is None + assert t.expired_at is None + assert t.last_used_at is None + assert t.team is None + assert t.created_by is None + + def test_create_options_defaults(self): + """TeamTokenCreateOptions defaults all fields to None.""" + opts = TeamTokenCreateOptions() + assert opts.description is None + assert opts.expired_at is None + + def test_create_options_with_description(self): + """TeamTokenCreateOptions stores description.""" + opts = TeamTokenCreateOptions(description="CI token") + assert opts.description == "CI token" + + def test_create_options_serializes_with_aliases(self): + """TeamTokenCreateOptions serialises with API aliases.""" + from datetime import datetime, timezone + + expiry = datetime(2027, 1, 1, tzinfo=timezone.utc) + opts = TeamTokenCreateOptions(description="Test", expired_at=expiry) + dumped = opts.model_dump(by_alias=True, exclude_none=True) + assert dumped["description"] == "Test" + assert "expired-at" in dumped + + def test_list_options(self): + """TeamTokenListOptions stores pagination and filter params.""" + opts = TeamTokenListOptions(page_size=10, query="my-team", sort="expired-at") + assert opts.page_size == 10 + assert opts.query == "my-team" + assert opts.sort == "expired-at" + + def test_created_by_choice_user(self): + """CreatedByChoice can hold a User.""" + u = User(id="user-123") + choice = CreatedByChoice(user=u) + assert choice.user.id == "user-123" + assert choice.team is None + assert choice.organization is None + + def test_created_by_choice_team(self): + """CreatedByChoice can hold a Team.""" + t = Team(id="team-abc") + choice = CreatedByChoice(team=t) + assert choice.team.id == "team-abc" + assert choice.user is None + + # ── Parser tests ───────────────────────────────────────────────────────── + + def test_team_token_from_full_data(self, service, token_api_data): + """_team_token_from parses attributes and typed relation stubs.""" + result = service._team_token_from(token_api_data) + + assert isinstance(result, TeamToken) + assert result.id == "at-abc123" + assert result.description == "My token" + assert result.token == "secret-token-value" + + # team relation is a typed Team stub + assert isinstance(result.team, Team) + assert result.team.id == "team-xyz789" + + # created_by is a CreatedByChoice wrapping a User stub + assert isinstance(result.created_by, CreatedByChoice) + assert isinstance(result.created_by.user, User) + assert result.created_by.user.id == "user-111" + + def test_team_token_from_team_creator(self, service, token_api_data_team_creator): + """_team_token_from handles team-type created-by relation.""" + result = service._team_token_from(token_api_data_team_creator) + + assert isinstance(result.team, Team) + assert isinstance(result.created_by, CreatedByChoice) + assert isinstance(result.created_by.team, Team) + assert result.created_by.team.id == "team-abc" + + def test_team_token_from_no_relationships(self, service): + """_team_token_from handles missing relationship data.""" + data = { + "id": "at-min", + "attributes": {"description": None, "token": None}, + "relationships": {}, + } + result = service._team_token_from(data) + + assert result.id == "at-min" + assert result.team is None + assert result.created_by is None + + def test_team_token_from_null_relationship_data(self, service): + """_team_token_from handles null data inside relationship.""" + data = { + "id": "at-null", + "attributes": {}, + "relationships": { + "team": {"data": None}, + "created-by": {"data": None}, + }, + } + result = service._team_token_from(data) + assert result.team is None + assert result.created_by is None + + # ── Resource method tests ───────────────────────────────────────────────── + + def test_create_legacy_success(self, service, mock_transport, token_api_data): + """create() without description uses legacy endpoint.""" + mock_response = Mock() + mock_response.json.return_value = {"data": token_api_data} + mock_transport.request.return_value = mock_response + + result = service.create(team_id="team-xyz789") + + args, kwargs = mock_transport.request.call_args + assert args[0] == "POST" + assert kwargs["path"] == "/api/v2/teams/team-xyz789/authentication-token" + assert kwargs["json_body"]["data"]["type"] == "authentication-token" + assert isinstance(result, TeamToken) + assert result.id == "at-abc123" + + def test_create_with_description_uses_new_endpoint( + self, service, mock_transport, token_api_data + ): + """create_with_options() with description uses the multi-token endpoint.""" + mock_response = Mock() + mock_response.json.return_value = {"data": token_api_data} + mock_transport.request.return_value = mock_response + + opts = TeamTokenCreateOptions(description="CI token") + service.create_with_options(team_id="team-xyz789", options=opts) + + args, kwargs = mock_transport.request.call_args + assert kwargs["path"] == "/api/v2/teams/team-xyz789/authentication-tokens" + assert kwargs["json_body"]["data"]["type"] == "authentication-tokens" + + def test_create_with_options_invalid_team_id(self, service): + """create_with_options() raises InvalidTeamIDError for a bad team ID.""" + with pytest.raises(InvalidTeamIDError): + service.create_with_options(team_id="not valid!") + + def test_create_invalid_team_id(self, service): + """create() raises InvalidTeamIDError for a bad team ID.""" + with pytest.raises(InvalidTeamIDError): + service.create(team_id="not valid!") + + def test_read_success(self, service, mock_transport, token_api_data): + """read() GETs the legacy endpoint and returns a TeamToken.""" + mock_response = Mock() + mock_response.json.return_value = {"data": token_api_data} + mock_transport.request.return_value = mock_response + + result = service.read(team_id="team-xyz789") + + mock_transport.request.assert_called_once_with( + "GET", path="/api/v2/teams/team-xyz789/authentication-token" + ) + assert isinstance(result, TeamToken) + assert result.id == "at-abc123" + + def test_read_invalid_team_id(self, service): + """read() raises InvalidTeamIDError for a bad team ID.""" + with pytest.raises(InvalidTeamIDError): + service.read(team_id="bad id") + + def test_read_by_id_success(self, service, mock_transport, token_api_data): + """read_by_id() GETs the correct path and returns a TeamToken.""" + mock_response = Mock() + mock_response.json.return_value = {"data": token_api_data} + mock_transport.request.return_value = mock_response + + result = service.read_by_id(token_id="at-abc123") + + mock_transport.request.assert_called_once_with( + "GET", path="/api/v2/authentication-tokens/at-abc123" + ) + assert isinstance(result, TeamToken) + + def test_read_by_id_invalid_token_id(self, service): + """read_by_id() raises InvalidTokenIDError for a bad token ID.""" + with pytest.raises(InvalidTokenIDError): + service.read_by_id(token_id="not valid!") + + def test_list_success(self, service, token_api_data): + """list() yields TeamToken objects from paginated results.""" + service._list = Mock(return_value=[token_api_data]) + + results = list(service.list(organization="my-org")) + + service._list.assert_called_once_with( + path="/api/v2/organizations/my-org/team-tokens", + params={}, + ) + assert len(results) == 1 + assert isinstance(results[0], TeamToken) + assert results[0].id == "at-abc123" + + def test_list_with_options(self, service, token_api_data): + """list() passes pagination and filter params.""" + service._list = Mock(return_value=[token_api_data]) + + opts = TeamTokenListOptions(page_size=5, query="my-team", sort="expired-at") + list(service.list(organization="my-org", options=opts)) + + _, kwargs = service._list.call_args + assert kwargs["params"]["page[size]"] == 5 + assert kwargs["params"]["q"] == "my-team" + assert kwargs["params"]["sort"] == "expired-at" + + def test_list_empty(self, service): + """list() returns empty iterator when no tokens exist.""" + service._list = Mock(return_value=[]) + results = list(service.list(organization="my-org")) + assert results == [] + + def test_list_invalid_org(self, service): + """list() raises InvalidOrgError for a bad organization name.""" + with pytest.raises(InvalidOrgError): + list(service.list(organization="not valid!")) + + def test_delete_success(self, service, mock_transport): + """delete() DELETEs the legacy token endpoint.""" + result = service.delete(team_id="team-xyz789") + + mock_transport.request.assert_called_once_with( + "DELETE", path="/api/v2/teams/team-xyz789/authentication-token" + ) + assert result is None + + def test_delete_invalid_team_id(self, service): + """delete() raises InvalidTeamIDError for a bad team ID.""" + with pytest.raises(InvalidTeamIDError): + service.delete(team_id="bad id") + + def test_delete_by_id_success(self, service, mock_transport): + """delete_by_id() DELETEs by token ID.""" + result = service.delete_by_id(token_id="at-abc123") + + mock_transport.request.assert_called_once_with( + "DELETE", path="/api/v2/authentication-tokens/at-abc123" + ) + assert result is None + + def test_delete_by_id_invalid_token_id(self, service): + """delete_by_id() raises InvalidTokenIDError for a bad token ID.""" + with pytest.raises(InvalidTokenIDError): + service.delete_by_id(token_id="not valid!")