|
| 1 | +# Copyright IBM Corp. 2025, 2026 |
| 2 | +# SPDX-License-Identifier: MPL-2.0 |
| 3 | + |
| 4 | +"""Read health assessment (drift detection / continuous validation) results. |
| 5 | +
|
| 6 | +``GET /api/v2/assessment-results/:id`` returns the assessment summary; the |
| 7 | +``/json-output``, ``/json-schema`` and ``/log-output`` companion endpoints |
| 8 | +return the underlying plan JSON, provider schema, and Terraform JSON log. |
| 9 | +
|
| 10 | +Those output endpoints do not adhere to JSON:API and (per the API docs) require |
| 11 | +a **user or team token with admin access to the workspace** — organization |
| 12 | +tokens cannot read them. |
| 13 | +
|
| 14 | +API reference: |
| 15 | +https://developer.hashicorp.com/terraform/cloud-docs/api-docs/assessment-results |
| 16 | +""" |
| 17 | + |
| 18 | +from __future__ import annotations |
| 19 | + |
| 20 | +from typing import Any |
| 21 | + |
| 22 | +import httpx |
| 23 | + |
| 24 | +from .._jsonapi import attach_jsonapi |
| 25 | +from ..errors import InvalidAssessmentResultIDError, TFEError |
| 26 | +from ..models.assessment_result import AssessmentResult |
| 27 | +from ..utils import valid_string_id |
| 28 | +from ._base import _Service |
| 29 | + |
| 30 | + |
| 31 | +def _assessment_result_from( |
| 32 | + data: dict[str, Any], included: list[dict[str, Any]] | None = None |
| 33 | +) -> AssessmentResult: |
| 34 | + """Parse a JSON:API assessment-results resource into an AssessmentResult.""" |
| 35 | + attrs = dict(data.get("attributes") or {}) |
| 36 | + attrs["id"] = data.get("id") |
| 37 | + return attach_jsonapi(AssessmentResult.model_validate(attrs), data, included) |
| 38 | + |
| 39 | + |
| 40 | +class AssessmentResults(_Service): |
| 41 | + """Service for reading workspace health assessment results.""" |
| 42 | + |
| 43 | + def read(self, assessment_result_id: str) -> AssessmentResult: |
| 44 | + """Read an assessment result by its ID.""" |
| 45 | + if not valid_string_id(assessment_result_id): |
| 46 | + raise InvalidAssessmentResultIDError() |
| 47 | + r = self.t.request( |
| 48 | + "GET", f"/api/v2/assessment-results/{assessment_result_id}" |
| 49 | + ) |
| 50 | + body = r.json() |
| 51 | + data = (body or {}).get("data") or {} if isinstance(body, dict) else {} |
| 52 | + included = body.get("included") if isinstance(body, dict) else None |
| 53 | + return _assessment_result_from(data, included) |
| 54 | + |
| 55 | + def json_output(self, assessment_result_id: str) -> dict[str, Any] | None: |
| 56 | + """Return the JSON plan output for an assessment result. |
| 57 | +
|
| 58 | + Only available once the assessment has succeeded and produced JSON |
| 59 | + output. Returns ``None`` when output is not yet ready (HTTP 204); the |
| 60 | + transport raises (e.g. ``NotFound``) when the assessment produced no |
| 61 | + output, such as when it did not succeed. Requires a user/team token with |
| 62 | + workspace admin access. |
| 63 | + """ |
| 64 | + if not valid_string_id(assessment_result_id): |
| 65 | + raise InvalidAssessmentResultIDError() |
| 66 | + resp = self._follow_blob( |
| 67 | + f"/api/v2/assessment-results/{assessment_result_id}/json-output" |
| 68 | + ) |
| 69 | + return self._as_json(resp) |
| 70 | + |
| 71 | + def json_schema(self, assessment_result_id: str) -> dict[str, Any] | None: |
| 72 | + """Return the JSON provider schema for an assessment result. |
| 73 | +
|
| 74 | + Returns ``None`` when the schema is not yet ready (HTTP 204); the |
| 75 | + transport raises when the assessment produced no schema (e.g. it did not |
| 76 | + succeed). Requires a user/team token with workspace admin access. |
| 77 | + """ |
| 78 | + if not valid_string_id(assessment_result_id): |
| 79 | + raise InvalidAssessmentResultIDError() |
| 80 | + resp = self._follow_blob( |
| 81 | + f"/api/v2/assessment-results/{assessment_result_id}/json-schema" |
| 82 | + ) |
| 83 | + return self._as_json(resp) |
| 84 | + |
| 85 | + def log_output(self, assessment_result_id: str) -> str: |
| 86 | + """Return the Terraform JSON log output for an assessment result as text. |
| 87 | +
|
| 88 | + Returns an empty string when there is no log output yet (HTTP 204). |
| 89 | + Requires a user/team token with workspace admin access. |
| 90 | + """ |
| 91 | + if not valid_string_id(assessment_result_id): |
| 92 | + raise InvalidAssessmentResultIDError() |
| 93 | + resp = self._follow_blob( |
| 94 | + f"/api/v2/assessment-results/{assessment_result_id}/log-output" |
| 95 | + ) |
| 96 | + return resp.text if resp is not None else "" |
| 97 | + |
| 98 | + def _follow_blob(self, path: str) -> httpx.Response | None: |
| 99 | + """Fetch a non-JSON:API output endpoint, following a blob redirect. |
| 100 | +
|
| 101 | + These endpoints may 307-redirect to a HashiCorp object-storage URL |
| 102 | + (Archivist), which requires the API bearer; we re-issue the request to |
| 103 | + the ``Location`` with auth (matching the plan ``json-output`` flow). |
| 104 | + Returns ``None`` when the API responds ``204 No Content``. |
| 105 | + """ |
| 106 | + resp = self.t.request("GET", path, allow_redirects=False) |
| 107 | + if resp.status_code == 204: |
| 108 | + return None |
| 109 | + if resp.status_code in (301, 302, 303, 307, 308): |
| 110 | + location = resp.headers.get("Location") or resp.headers.get("location") |
| 111 | + if not location: |
| 112 | + raise TFEError( |
| 113 | + "assessment-results output redirect did not include a Location header" |
| 114 | + ) |
| 115 | + return self.t.request("GET", location) |
| 116 | + return resp |
| 117 | + |
| 118 | + @staticmethod |
| 119 | + def _as_json(resp: httpx.Response | None) -> dict[str, Any] | None: |
| 120 | + if resp is None: |
| 121 | + return None |
| 122 | + try: |
| 123 | + data = resp.json() |
| 124 | + except Exception: |
| 125 | + return None |
| 126 | + if data is None: |
| 127 | + return None |
| 128 | + return data if isinstance(data, dict) else {"data": data} |
0 commit comments