diff --git a/Makefile b/Makefile index 11c59d7..731afd8 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ .PHONY: help fmt fmt-check lint check test install dev-install type-check clean all venv activate PYTHON := python3 -SRC_DIR := tfe +SRC_DIR := src/tfe TEST_DIR := tests VENV := .venv VENV_PYTHON := $(VENV)/bin/python diff --git a/doc/.keep b/doc/.keep new file mode 100644 index 0000000..1b6ba71 --- /dev/null +++ b/doc/.keep @@ -0,0 +1 @@ +## \ No newline at end of file diff --git a/examples/ws_list.py b/examples/ws_list.py new file mode 100644 index 0000000..159fa3e --- /dev/null +++ b/examples/ws_list.py @@ -0,0 +1,12 @@ +from tfe import TFEClient, TFEConfig + + +def main(): + client = TFEClient(TFEConfig.from_env()) + org = "tfe-xxxxx" + for ws in client.workspaces.list(org): + print("WS:", ws.name, ws.id) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index 6a441bd..76afdbc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,17 +1,26 @@ [build-system] -requires = ["setuptools>=61.0", "wheel"] -build-backend = "setuptools.build_meta" +requires = ["hatchling>=1.21"] +build-backend = "hatchling.build" [project] -name = "python-tfe" -version = "0.1.0" -description = "Python client library for Terraform Enterprise/Cloud API" +name = "tfe" +version = "0.1.0a1" +description = "Official Python SDK for HashiCorp Terraform Cloud / Terraform Enterprise (TFE) API v2" readme = "README.md" license = { text = "MPL-2.0" } authors = [{ name = "HashiCorp" }] -requires-python = ">=3.10" -dependencies = ["requests>=2.25.0"] +requires-python = ">=3.9" +dependencies = [ + "httpx>=0.27.0,<0.29.0", + "pydantic>=2.6,<3", + "typing-extensions>=4.8", + "anyio>=4.0", + "h2>=4.3.0" +] classifiers = [ + "License :: OSI Approved :: Mozilla Public License 2.0 (MPL 2.0)", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", "Programming Language :: Python", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", @@ -32,11 +41,8 @@ dev = [ "Repository" = "https://github.com/hashicorp/python-tfe" "Bug Tracker" = "https://github.com/hashicorp/python-tfe/issues" -[tool.setuptools.packages.find] -where = ["tfe"] - -[tool.setuptools.package-dir] -"" = "tfe" +[tool.hatch.build.targets.wheel] +packages = ["src/tfe"] # Ruff configuration [tool.ruff] @@ -97,7 +103,7 @@ warn_unused_configs = true disallow_untyped_defs = true disallow_incomplete_defs = true check_untyped_defs = true -disallow_untyped_decorators = true +disallow_untyped_decorators = false no_implicit_optional = true warn_redundant_casts = true warn_unused_ignores = true diff --git a/src/tfe/__init__.py b/src/tfe/__init__.py new file mode 100644 index 0000000..8c1b6c6 --- /dev/null +++ b/src/tfe/__init__.py @@ -0,0 +1,5 @@ +from . import errors +from .client import TFEClient +from .config import TFEConfig + +__all__ = ["TFEConfig", "TFEClient", "errors"] diff --git a/src/tfe/_http.py b/src/tfe/_http.py new file mode 100644 index 0000000..b8a351f --- /dev/null +++ b/src/tfe/_http.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import time +from collections.abc import Mapping +from typing import Any + +import anyio +import httpx + +from ._jsonapi import build_headers, parse_error_payload +from .errors import ( + AuthError, + NotFound, + RateLimited, + ServerError, + TFEError, +) + +_RETRY_STATUSES = {429, 502, 503, 504} + + +class HTTPTransport: + def __init__( + self, + address: str, + token: str, + *, + timeout: float, + verify_tls: bool, + user_agent_suffix: str | None, + max_retries: int, + backoff_base: float, + backoff_cap: float, + backoff_jitter: bool, + http2: bool, + proxies: dict | None, + ca_bundle: str | None, + ): + self.base = address.rstrip("/") + self.headers = build_headers(user_agent_suffix) + if token: + self.headers["Authorization"] = f"Bearer {token}" + self.timeout = timeout + self.verify = verify_tls + self.max_retries = max_retries + self.backoff_base = backoff_base + self.backoff_cap = backoff_cap + self.backoff_jitter = backoff_jitter + self.http2 = http2 + self.proxies = proxies + self.ca_bundle = ca_bundle + self._sync = httpx.Client( + http2=http2, timeout=timeout, verify=ca_bundle or verify_tls + ) # proxies=proxies + self._async = httpx.AsyncClient( + http2=http2, timeout=timeout, verify=ca_bundle or verify_tls + ) # proxies=proxies + + def request( + self, + method: str, + path: str, + *, + params: Mapping[str, Any] | None = None, + json_body: Mapping[str, Any] | None = None, + headers: dict[str, str] | None = None, + allow_redirects: bool = True, + ) -> httpx.Response: + url = f"{self.base}{path}" + hdrs = dict(self.headers) + if headers: + hdrs.update(headers) + attempt = 0 + while True: + try: + resp = self._sync.request( + method, + url, + params=params, + json=json_body, + headers=hdrs, + follow_redirects=allow_redirects, + ) + except httpx.HTTPError as e: + if attempt >= self.max_retries: + raise ServerError(str(e)) from e + self._sleep(attempt, None) + attempt += 1 + continue + if resp.status_code in _RETRY_STATUSES and attempt < self.max_retries: + retry_after = _parse_retry_after(resp) + self._sleep(attempt, retry_after) + attempt += 1 + continue + self._raise_if_error(resp) + return resp + + async def arequest( + self, + method: str, + path: str, + *, + params: Mapping[str, Any] | None = None, + json_body: Mapping[str, Any] | None = None, + headers: dict[str, str] | None = None, + allow_redirects: bool = True, + ) -> httpx.Response: + url = f"{self.base}{path}" + hdrs = dict(self.headers) + hdrs.update(headers or {}) + attempt = 0 + while True: + try: + resp = await self._async.request( + method, + url, + params=params, + json=json_body, + headers=hdrs, + follow_redirects=allow_redirects, + ) + except httpx.HTTPError as e: + if attempt >= self.max_retries: + raise ServerError(str(e)) from e + await self._asleep(attempt, None) + attempt += 1 + continue + if resp.status_code in _RETRY_STATUSES and attempt < self.max_retries: + retry_after = _parse_retry_after(resp) + await self._asleep(attempt, retry_after) + attempt += 1 + continue + self._raise_if_error(resp) + return resp + + def _sleep(self, attempt: int, retry_after: float | None) -> None: + if retry_after is not None: + time.sleep(retry_after) + return + delay = min(self.backoff_cap, self.backoff_base * (2**attempt)) + time.sleep(delay) + + async def _asleep(self, attempt: int, retry_after: float | None) -> None: + if retry_after is not None: + await anyio.sleep(retry_after) + return + delay = min(self.backoff_cap, self.backoff_base * (2**attempt)) + await anyio.sleep(delay) + + def _raise_if_error(self, resp: httpx.Response) -> None: + status = resp.status_code + + if 200 <= status < 300: + return + try: + payload: Any = resp.json() + except Exception: + payload = {} + errors = parse_error_payload(payload) + msg: str = f"HTTP {status}" + if errors: + maybe_detail = errors[0].get("detail") + maybe_title = errors[0].get("title") + if isinstance(maybe_detail, str) and maybe_detail: + msg = maybe_detail + elif isinstance(maybe_title, str) and maybe_title: + msg = maybe_title + + if status in (401, 403): + raise AuthError(msg, status=status, errors=errors) + if status == 404: + raise NotFound(msg, status=status, errors=errors) + if status == 429: + ra = _parse_retry_after(resp) + raise RateLimited(msg, status=status, errors=errors, retry_after=ra) + if status >= 500: + raise ServerError(msg, status=status, errors=errors) + raise TFEError(msg, status=status, errors=errors) + + +def _parse_retry_after(resp: httpx.Response) -> float | None: + ra = resp.headers.get("Retry-After") + if not ra: + return None + try: + return float(ra) + except Exception: + return None diff --git a/src/tfe/_jsonapi.py b/src/tfe/_jsonapi.py new file mode 100644 index 0000000..ada4bcb --- /dev/null +++ b/src/tfe/_jsonapi.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from typing import Any + + +def build_headers(user_agent_suffix: str | None = None) -> dict[str, str]: + ua = "python-tfe/0.1" + if user_agent_suffix: + ua = f"{ua} {user_agent_suffix}" + return { + "Accept": "application/vnd.api+json", + "Content-Type": "application/vnd.api+json", + "User-Agent": ua, + } + + +def parse_error_payload(payload: dict[str, Any]) -> list[dict]: + errs = payload.get("errors") + if isinstance(errs, list): + return errs + if "message" in payload: + return [{"detail": payload.get("message")}] + return [] diff --git a/src/tfe/aclient.py b/src/tfe/aclient.py new file mode 100644 index 0000000..9580ba5 --- /dev/null +++ b/src/tfe/aclient.py @@ -0,0 +1,27 @@ +""" +Async TFE Client: This client should not be used for now. +""" + +from __future__ import annotations + +from ._http import HTTPTransport +from .config import TFEConfig + + +class AsyncTFEClient: + def __init__(self, config: TFEConfig | None = None): + cfg = config or TFEConfig.from_env() + self._transport = HTTPTransport( + cfg.address, + cfg.token, + timeout=cfg.timeout, + verify_tls=cfg.verify_tls, + user_agent_suffix=cfg.user_agent_suffix, + max_retries=cfg.max_retries, + backoff_base=cfg.backoff_base, + backoff_cap=cfg.backoff_cap, + backoff_jitter=cfg.backoff_jitter, + http2=cfg.http2, + proxies=cfg.proxies, + ca_bundle=cfg.ca_bundle, + ) diff --git a/src/tfe/client.py b/src/tfe/client.py new file mode 100644 index 0000000..e4e2754 --- /dev/null +++ b/src/tfe/client.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from ._http import HTTPTransport +from .config import TFEConfig +from .resources.organizations import Organizations +from .resources.projects import Projects +from .resources.workspaces import Workspaces + + +class TFEClient: + def __init__(self, config: TFEConfig | None = None): + cfg = config or TFEConfig.from_env() + self._transport = HTTPTransport( + cfg.address, + cfg.token, + timeout=cfg.timeout, + verify_tls=cfg.verify_tls, + user_agent_suffix=cfg.user_agent_suffix, + max_retries=cfg.max_retries, + backoff_base=cfg.backoff_base, + backoff_cap=cfg.backoff_cap, + backoff_jitter=cfg.backoff_jitter, + http2=cfg.http2, + proxies=cfg.proxies, + ca_bundle=cfg.ca_bundle, + ) + self.organizations = Organizations(self._transport) + self.projects = Projects(self._transport) + self.workspaces = Workspaces(self._transport) + + def close(self) -> None: + pass diff --git a/src/tfe/config.py b/src/tfe/config.py new file mode 100644 index 0000000..6a70e34 --- /dev/null +++ b/src/tfe/config.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import os + +from pydantic import BaseModel, Field + + +class TFEConfig(BaseModel): + address: str = Field( + default_factory=lambda: os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + token: str = Field(default_factory=lambda: os.getenv("TFE_TOKEN", "")) + timeout: float = float(os.getenv("TFE_TIMEOUT", "30")) + verify_tls: bool = os.getenv("TFE_VERIFY_TLS", "true").lower() not in ( + "0", + "false", + "no", + ) + user_agent_suffix: str | None = None + max_retries: int = int(os.getenv("TFE_MAX_RETRIES", "5")) + backoff_base: float = 0.5 + backoff_cap: float = 8.0 + backoff_jitter: bool = True + http2: bool = True + proxies: dict[str, str] | None = None + ca_bundle: str | None = os.getenv("SSL_CERT_FILE", None) + + @classmethod + def from_env(cls) -> TFEConfig: + return cls() diff --git a/src/tfe/errors.py b/src/tfe/errors.py new file mode 100644 index 0000000..6742980 --- /dev/null +++ b/src/tfe/errors.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any + + +class TFEError(Exception): + def __init__( + self, + message: str, + *, + status: int | None = None, + errors: list[dict] | None = None, + ): + super().__init__(message) + self.status = status + self.errors = errors or [] + + +class AuthError(TFEError): ... + + +class NotFound(TFEError): ... + + +class RateLimited(TFEError): + def __init__( + self, + message: str, + *, + retry_after: float | None = None, + **kw: Any, + ) -> None: + super().__init__(message, **kw) + self.retry_after = retry_after + + +class ValidationError(TFEError): ... + + +class ServerError(TFEError): ... + + +class UnsupportedInCloud(TFEError): ... + + +class UnsupportedInEnterprise(TFEError): ... diff --git a/src/tfe/resources/_base.py b/src/tfe/resources/_base.py new file mode 100644 index 0000000..9736d6c --- /dev/null +++ b/src/tfe/resources/_base.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from collections.abc import AsyncIterator, Iterator +from typing import Any + +from .._http import HTTPTransport + + +class _Service: + def __init__(self, t: HTTPTransport) -> None: + self.t = t + + def _list( + self, path: str, *, params: dict | None = None + ) -> Iterator[dict[str, Any]]: + page = 1 + while True: + p = dict(params or {}) + p.setdefault("page[number]", page) + p.setdefault("page[size]", 100) + r = self.t.request("GET", path, params=p) + data = r.json().get("data", []) + yield from data + if len(data) < p["page[size]"]: + break + page += 1 + + +""" +Warning: Do Not Use this Async Service as its not stable with HashiCorp API. +""" + + +class _AService: + def __init__(self, t: HTTPTransport) -> None: + self.t = t + + async def _alist( + self, path: str, *, params: dict | None = None + ) -> AsyncIterator[dict[str, Any]]: + page = 1 + while True: + p = dict(params or {}) + p.setdefault("page[number]", page) + p.setdefault("page[size]", 100) + r = await self.t.arequest("GET", path, params=p) + data = r.json().get("data", []) + for item in data: + yield item + if len(data) < p["page[size]"]: + break + page += 1 diff --git a/src/tfe/resources/admin/settings.py b/src/tfe/resources/admin/settings.py new file mode 100644 index 0000000..44a654d --- /dev/null +++ b/src/tfe/resources/admin/settings.py @@ -0,0 +1,17 @@ +from __future__ import annotations + +from typing import Any + +from .._base import _AService, _Service + + +class AdminSettings(_Service): + def terraform_versions(self) -> Any: + r = self.t.request("GET", "/api/v2/admin/terraform-versions") + return r.json() + + +class AdminSettingsAsync(_AService): + async def terraform_versions(self) -> Any: + r = await self.t.arequest("GET", "/api/v2/admin/terraform-versions") + return r.json() diff --git a/src/tfe/resources/organizations.py b/src/tfe/resources/organizations.py new file mode 100644 index 0000000..fde1603 --- /dev/null +++ b/src/tfe/resources/organizations.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..types import Organization +from ._base import _Service + + +def _safe_str(v: Any, default: str = "") -> str: + return v if isinstance(v, str) else (str(v) if v is not None else default) + + +class Organizations(_Service): + def list(self) -> Iterator[Organization]: + for item in self._list("/api/v2/organizations"): + attr = item.get("attributes", {}) or {} + org_id = _safe_str(item.get("id")) + name = _safe_str(attr.get("name") or item.get("id")) + email = attr.get("email") if isinstance(attr.get("email"), str) else None + yield Organization(id=org_id, name=name, email=email) + + def get(self, name: str) -> Organization: + r = self.t.request("GET", f"/api/v2/organizations/{name}") + d = r.json()["data"] + attr = d.get("attributes", {}) or {} + org_id = _safe_str(d.get("id")) + org_name = _safe_str(attr.get("name") or d.get("id")) + email = attr.get("email") if isinstance(attr.get("email"), str) else None + return Organization(id=org_id, name=org_name, email=email) diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py new file mode 100644 index 0000000..6647edc --- /dev/null +++ b/src/tfe/resources/projects.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..types import Project +from ._base import _Service + + +def _safe_str(v: Any, default: str = "") -> str: + return v if isinstance(v, str) else (str(v) if v is not None else default) + + +class Projects(_Service): + def list(self, organization: str) -> Iterator[Project]: + path = f"/api/v2/organizations/{organization}/projects" + for item in self._list(path): + attr = item.get("attributes", {}) or {} + proj_id = _safe_str(item.get("id")) + name = _safe_str(attr.get("name")) + yield Project(id=proj_id, name=name, organization=organization) diff --git a/src/tfe/resources/workspaces.py b/src/tfe/resources/workspaces.py new file mode 100644 index 0000000..9f778eb --- /dev/null +++ b/src/tfe/resources/workspaces.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import builtins +from collections.abc import Iterator +from typing import Any + +from ..types import ExecutionMode, Workspace +from ._base import _Service + + +def _safe_str(v: Any, default: str = "") -> str: + return v if isinstance(v, str) else (str(v) if v is not None else default) + + +def _em_safe(v: Any) -> ExecutionMode | None: + # Only accept strings; map to enum if known, else None + if not isinstance(v, str): + return None + return ExecutionMode._value2member_map_.get(v) # type: ignore[return-value] + + +def _ws_from(d: dict[str, Any], org: str | None = None) -> Workspace: + attr: dict[str, Any] = d.get("attributes", {}) or {} + + # Coerce to required string fields (empty string fallback keeps mypy happy) + id_str: str = _safe_str(d.get("id")) + name_str: str = _safe_str(attr.get("name")) + org_str: str = _safe_str(org if org is not None else attr.get("organization")) + + # Optional fields + em: ExecutionMode | None = _em_safe(attr.get("execution-mode")) + + proj_id: str | None = None + proj = attr.get("project") + if isinstance(proj, dict): + proj_id = proj.get("id") if isinstance(proj.get("id"), str) else None + + tags_val = attr.get("tags", []) or [] + tags_list: list[str] = list(tags_val) if isinstance(tags_val, list | tuple) else [] + + return Workspace( + id=id_str, + name=name_str, + organization=org_str, + execution_mode=em, + project_id=proj_id, + tags=tags_list, + ) + + +class Workspaces(_Service): + def list( + self, organization: str, *, search: str | None = None + ) -> Iterator[Workspace]: + params: dict[str, Any] = {} + if search: + params["search[name]"] = search + path = f"/api/v2/organizations/{organization}/workspaces" + for item in self._list(path, params=params): + yield _ws_from(item, organization) + + def get(self, id_or_name: str, organization: str | None = None) -> Workspace: + if organization: + r = self.t.request( + "GET", f"/api/v2/organizations/{organization}/workspaces/{id_or_name}" + ) + else: + r = self.t.request("GET", f"/api/v2/workspaces/{id_or_name}") + return _ws_from(r.json()["data"], organization) + + def create( + self, + organization: str, + name: str, + *, + execution_mode: str | None = "remote", + project_id: str | None = None, + tags: builtins.list[str] | None = None, + ) -> Workspace: + body: dict[str, Any] = { + "data": {"type": "workspaces", "attributes": {"name": name}} + } + if execution_mode: + body["data"]["attributes"]["execution-mode"] = execution_mode + if project_id: + body["data"].setdefault("relationships", {}) + body["data"]["relationships"]["project"] = { + "data": {"type": "projects", "id": project_id} + } + if tags: + body["data"]["attributes"]["tags"] = list(tags) + + r = self.t.request( + "POST", f"/api/v2/organizations/{organization}/workspaces", json_body=body + ) + return _ws_from(r.json()["data"], organization) + + def update(self, id: str, **attrs: Any) -> Workspace: + body: dict[str, Any] = { + "data": {"type": "workspaces", "id": id, "attributes": {}} + } + for k, v in attrs.items(): + kk = k.replace("_", "-") + # Map enum back to string if provided + if kk == "execution-mode" and isinstance(v, ExecutionMode): + v = v.value + body["data"]["attributes"][kk] = v + r = self.t.request("PATCH", f"/api/v2/workspaces/{id}", json_body=body) + return _ws_from(r.json()["data"], None) + + def delete(self, id: str) -> None: + self.t.request("DELETE", f"/api/v2/workspaces/{id}") diff --git a/src/tfe/types.py b/src/tfe/types.py new file mode 100644 index 0000000..478614e --- /dev/null +++ b/src/tfe/types.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from enum import Enum + +from pydantic import BaseModel, Field + + +class ExecutionMode(str, Enum): + REMOTE = "remote" + AGENT = "agent" + LOCAL = "local" + + +class RunStatus(str, Enum): + PLANNING = "planning" + PLANNED = "planned" + APPLIED = "applied" + CANCELED = "canceled" + ERRORED = "errored" + + +class Organization(BaseModel): + id: str + name: str + email: str | None = None + + +class Project(BaseModel): + id: str + name: str + organization: str + + +class Workspace(BaseModel): + id: str + name: str + organization: str + execution_mode: ExecutionMode | None = None + project_id: str | None = None + tags: list[str] = Field(default_factory=list) diff --git a/src/tfe/utils.py b/src/tfe/utils.py new file mode 100644 index 0000000..968bb4d --- /dev/null +++ b/src/tfe/utils.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +import time +from collections.abc import Callable + + +def poll_until( + fn: Callable[[], bool], + *, + interval_s: float = 5.0, + timeout_s: float | None = 600, +) -> bool: + start = time.time() + while True: + value = fn() + if value: + return True + if timeout_s is not None and (time.time() - start) > timeout_s: + raise TimeoutError("Timed out") + time.sleep(interval_s) diff --git a/tests/units/test_client.py b/tests/units/test_client.py deleted file mode 100644 index 591bf50..0000000 --- a/tests/units/test_client.py +++ /dev/null @@ -1,104 +0,0 @@ -from unittest.mock import Mock, patch - -import pytest - -from tfe import client, config - - -@pytest.fixture -def test_config(): - return config.Config(address="https://app.terraform.io", token="test-token") - - -@pytest.fixture -def mock_response(): - response = Mock() - response.headers = { - "TFP-API-Version": "2.5.0", - "X-TFE-Version": "v202308-1", - "TFP-AppName": "HCP Terraform", - } - response.raise_for_status.return_value = None - return response - - -class TestClient: - @patch("requests.Session.get") - def test_client_initialization(self, mock_get, test_config, mock_response): - """Test basic client setup works.""" - mock_get.return_value = mock_response - - client_instance = client.Client(config=test_config) - - assert client_instance.config.address == "https://app.terraform.io" - assert client_instance.config.token == "test-token" - assert client_instance.base_url == "https://app.terraform.io/api/v2/" - assert ( - client_instance.registry_base_url - == "https://app.terraform.io/api/registry/" - ) - - @patch("requests.Session.get") - def test_url_normalization(self, mock_get, mock_response): - """Test that paths get normalized with trailing slashes.""" - mock_get.return_value = mock_response - - cfg = config.Config( - address="https://example.com", - token="test", - base_path="/custom/api", # no trailing slash - registry_base_path="/registry", # no trailing slash - ) - - client_instance = client.Client(config=cfg) - - assert client_instance.base_url == "https://example.com/custom/api/" - assert client_instance.registry_base_url == "https://example.com/registry/" - - @patch("requests.Session.get") - def test_api_metadata_extraction(self, mock_get, test_config, mock_response): - """Test that API metadata gets extracted from response headers.""" - mock_get.return_value = mock_response - - client_instance = client.Client(config=test_config) - - assert client_instance.remote_api_version == "2.5.0" - assert client_instance.remote_tfe_version == "v202308-1" - assert client_instance.app_name == "HCP Terraform" - - @patch("requests.Session.get") - def test_cloud_vs_enterprise_detection(self, mock_get, test_config): - """Test detection between cloud and enterprise instances.""" - # Test HCP Terraform (cloud) - cloud_response = Mock() - cloud_response.headers = {"TFP-AppName": "HCP Terraform"} - cloud_response.raise_for_status.return_value = None - mock_get.return_value = cloud_response - - cloud_client = client.Client(config=test_config) - assert cloud_client.is_cloud() is True - assert cloud_client.is_enterprise() is False - - # Test Terraform Enterprise - enterprise_response = Mock() - enterprise_response.headers = {"TFP-AppName": "Terraform Enterprise"} - enterprise_response.raise_for_status.return_value = None - mock_get.return_value = enterprise_response - - enterprise_client = client.Client(config=test_config) - assert enterprise_client.is_cloud() is False - assert enterprise_client.is_enterprise() is True - - @patch("requests.Session.get") - def test_fake_api_version_for_testing(self, mock_get, test_config, mock_response): - """Test the fake API version setter for testing scenarios.""" - mock_get.return_value = mock_response - - client_instance = client.Client(config=test_config) - - # Original version from mock - assert client_instance.remote_api_version == "2.5.0" - - # Set fake version - client_instance.set_fake_remote_api_version("3.0.0") - assert client_instance.remote_api_version == "3.0.0" diff --git a/tests/units/test_config.py b/tests/units/test_config.py deleted file mode 100644 index 107413f..0000000 --- a/tests/units/test_config.py +++ /dev/null @@ -1,97 +0,0 @@ -import pytest -import requests - -from tfe import config - - -@pytest.fixture(autouse=True) -def reset_environment(monkeypatch): - """Reset environment variables before each test.""" - monkeypatch.delenv("TFE_ADDRESS", raising=False) - monkeypatch.delenv("TFE_TOKEN", raising=False) - monkeypatch.delenv("TFE_HOST", raising=False) - monkeypatch.setenv("TFE_TOKEN", "abc123") - yield - - -@pytest.fixture -def cfg(): - """Provide a fresh Config instance with clean environment.""" - return config.Config() - - -@pytest.fixture -def test_session(): - """Provide a clean requests session without default headers.""" - session = requests.Session() - session.headers["User-Agent"] = "test" - session.headers["Authorization"] = "Bearer test" - return session - - -class TestConfig: - def test_default_config(self, cfg): - """Test that default configuration values are set correctly.""" - assert cfg.address == config.DEFAULT_ADDRESS - assert cfg.base_path == config.DEFAULT_BASE_PATH - assert cfg.registry_base_path == config.DEFAULT_REGISTRY_PATH - assert isinstance(cfg.http_client, requests.Session) - assert "User-Agent" in cfg.http_client.headers - assert cfg.retry_log_hook is None - assert cfg.retry_server_errors is False - - def test_env_address_and_token(self, monkeypatch): - """Test that environment variables TFE_ADDRESS and TFE_TOKEN are read correctly.""" - monkeypatch.setenv("TFE_ADDRESS", "https://custom.tfe") - cfg = config.Config() - assert cfg.address == "https://custom.tfe" - assert cfg.token == "abc123" - - def test_env_host_fallback(self, monkeypatch): - """Test that TFE_HOST is used as fallback when TFE_ADDRESS is not set.""" - monkeypatch.setenv("TFE_HOST", "host.tfe") - cfg = config.Config() - assert cfg.address == "https://host.tfe" - - def test_explicit_address_override(self): - """Test that explicitly passed address overrides environment variables.""" - cfg = config.Config(address="https://explicit.tfe") - assert cfg.address == "https://explicit.tfe" - - def test_headers_update(self): - """Test that custom headers are properly merged with default headers.""" - custom_headers = {"Authorization": "Bearer testtoken", "X-Test": "yes"} - cfg = config.Config(headers=custom_headers) - assert "Authorization" in cfg.http_client.headers - assert cfg.http_client.headers["Authorization"] == "Bearer testtoken" - assert "X-Test" in cfg.http_client.headers - assert cfg.http_client.headers["X-Test"] == "yes" - assert "User-Agent" in cfg.http_client.headers - - def test_retry_log_hook_and_server_errors(self): - """Test that retry configuration is properly set.""" - - def dummy_hook(retries, response): - pass - - cfg = config.Config(retry_log_hook=dummy_hook, retry_server_errors=True) - assert cfg.retry_log_hook == dummy_hook - assert cfg.retry_server_errors is True - - def test_custom_session(self, test_session): - """Test that User-Agent is set when session has no default User-Agent.""" - cfg = config.Config(http_client=test_session) - assert "User-Agent" in cfg.http_client.headers - assert cfg.http_client.headers["User-Agent"] == "test" - assert cfg.http_client.headers["Authorization"] == "Bearer test" - - def test_validate_config(self, monkeypatch): - """Test that configuration validation works as expected.""" - with pytest.raises(ValueError, match="API token is required") as _: - monkeypatch.setenv("TFE_TOKEN", "") - _ = config.Config(token="") - - with pytest.raises(ValueError, match="Address must include protocol") as _: - monkeypatch.setenv("TFE_TOKEN", "test-token") - monkeypatch.setenv("TFE_ADDRESS", "test.foo.bar") - _ = config.Config() diff --git a/tests/units/test_transport.py b/tests/units/test_transport.py new file mode 100644 index 0000000..cad04eb --- /dev/null +++ b/tests/units/test_transport.py @@ -0,0 +1,21 @@ +from tfe._http import HTTPTransport +from tfe.config import TFEConfig + + +def test_http_transport_init(): + cfg = TFEConfig() + t = HTTPTransport( + cfg.address, + "", + timeout=cfg.timeout, + verify_tls=cfg.verify_tls, + user_agent_suffix=None, + max_retries=1, + backoff_base=0.01, + backoff_cap=0.02, + backoff_jitter=False, + http2=False, + proxies=None, + ca_bundle=None, + ) + assert t.base.startswith("https://") diff --git a/tfe/__init__.py b/tfe/__init__.py deleted file mode 100644 index 60fbccb..0000000 --- a/tfe/__init__.py +++ /dev/null @@ -1,12 +0,0 @@ -""" -Python client library for Terraform Enterprise/Cloud API. - -This package provides a Python interface to the Terraform Enterprise -and Terraform Cloud APIs, allowing you to programmatically manage -workspaces, runs, state files, and other TFE/TFC resources. -""" - -from tfe.client import Client, TFEClientError -from tfe.config import Config - -__all__ = ["Client", "TFEClientError", "Config"] diff --git a/tfe/client.py b/tfe/client.py deleted file mode 100644 index 94c5c27..0000000 --- a/tfe/client.py +++ /dev/null @@ -1,90 +0,0 @@ -""" -Main client class for Terraform Enterprise/Cloud API. -""" - -import logging -from urllib.parse import urljoin - -from tfe.config import Config - -logger = logging.getLogger(__name__) - - -class TFEClientError(Exception): - """Base exception for TFE client errors.""" - - pass - - -class Client: - """ - Client is the Terraform Enterprise API client. It provides the basic - functionality to interact with the Terraform API. - """ - - def __init__(self, config: Config | None = None) -> None: - self.config = config or Config() - self._setup_urls() - - self._api_version = "" - self._tfe_version = "" - self._app_name = "" - self._fetch_api_metadata() - - def _setup_urls(self) -> None: - """Parse and setup base URLs.""" - # Ensure base path ends with / - base_path = self.config.base_path - if not base_path.endswith("/"): - base_path += "/" - - registry_path = self.config.registry_base_path - if not registry_path.endswith("/"): - registry_path += "/" - - self.base_url = urljoin(self.config.address, base_path) - self.registry_base_url = urljoin(self.config.address, registry_path) - - def _fetch_api_metadata(self) -> None: - """Fetch API metadata from the server.""" - ping_url = urljoin(self.base_url, "ping") - headers = { - "Accept": "application/vnd.api+json", - } - if self.config.headers: - headers.update(self.config.headers) - - response = self.config.http_client.get(ping_url, headers=headers) - response.raise_for_status() - - # Extract metadata from headers - self._api_version = response.headers.get("TFP-API-Version", "") - self._tfe_version = response.headers.get("X-TFE-Version", "") - self._app_name = response.headers.get("TFP-AppName", "") - - @property - def remote_api_version(self) -> str: - """Return the server's declared API version string.""" - return self._api_version - - @property - def remote_tfe_version(self) -> str: - """Return the server's declared TFE version string.""" - return self._tfe_version - - @property - def app_name(self) -> str: - """Return the name of the instance.""" - return self._app_name - - def is_cloud(self) -> bool: - """Return True if the client is configured against HCP Terraform.""" - return self._app_name == "HCP Terraform" - - def is_enterprise(self) -> bool: - """Return True if the client is configured against Terraform Enterprise.""" - return not self.is_cloud() - - def set_fake_remote_api_version(self, version: str) -> None: - """Set a fake API version for testing purposes.""" - self._api_version = version diff --git a/tfe/config.py b/tfe/config.py deleted file mode 100644 index f412f88..0000000 --- a/tfe/config.py +++ /dev/null @@ -1,86 +0,0 @@ -import logging -import os -from collections.abc import Callable -from dataclasses import dataclass, field -from urllib.parse import urlparse - -import requests - -logger = logging.getLogger(__name__) - -DEFAULT_ADDRESS = "https://app.terraform.io" -DEFAULT_BASE_PATH = "/api/v2/" -DEFAULT_REGISTRY_PATH = "/api/registry" - - -@dataclass -class Config: - # Address of the Terraform Enterprise API - address: str = field(default="") - - # Base path for which the API is served - base_path: str = DEFAULT_BASE_PATH - - # Base path for the Terraform Enterprise Registry API - registry_base_path: str = DEFAULT_REGISTRY_PATH - - # API token used to access the terraform enterprise API - token: str = field(default="") - - # Headers to include in API requests - # TODO: Do we need headers ? we can pass them directly to http_client, but this will differ from the go-tfe module - headers: dict[str, str] = field(default_factory=dict) - - # Custom request session which needs to be used - http_client: requests.Session = field(default_factory=requests.Session) - - # Callable to run before any request is retried - retry_log_hook: Callable[[int, requests.Response], None] | None = None - - # Enable/Disable retry logic - retry_server_errors: bool = False - - def _set_address(self) -> None: - tfe_address = os.getenv("TFE_ADDRESS", "") - if tfe_address: - self.address = tfe_address - - if not self.address: - if os.getenv("TFE_HOST"): - self.address = f"https://{os.getenv('TFE_HOST')}" - else: - self.address = DEFAULT_ADDRESS - - def _set_token(self) -> None: - if not self.token: - self.token = os.getenv("TFE_TOKEN", "") - - if ( - self.token - and "Authorization" not in self.http_client.headers - and "Authorization" not in self.headers - ): - self.headers["Authorization"] = f"Bearer {self.token}" - - def _set_user_agent(self) -> None: - if ( - "User-Agent" not in self.http_client.headers - and "User-Agent" not in self.headers - ): - self.headers["User-Agent"] = "python-tfe" - - def _validate_config(self) -> None: - if not self.http_client.headers.get("Authorization"): - raise ValueError( - "API token is required, please set the TFE_TOKEN environment variable or the token field in the configuration." - ) - parsed_url = urlparse(self.address) - if not parsed_url.scheme: - raise ValueError("Address must include protocol (http/https)") - - def __post_init__(self) -> None: - self._set_address() - self._set_token() - self._set_user_agent() - self.http_client.headers.update(self.headers) - self._validate_config()