Optional HTTP request/response attachment in Allure Reports

.gitignore

@@ -4,6 +4,7 @@ __pycache__/
 .pytest_cache/
 .env
 .vscode/settings.json
+.DS_Store
 
 # Ignore Allure results and reports
 allure-results/

scripts/allure_helper.py

@@ -46,7 +46,7 @@ def serve_allure_report(results_dir: str) -> None:
 
 
 def main():
-    project_root = os.path.dirname(os.path.abspath(__file__))
+    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     parser = argparse.ArgumentParser(
         description="Clean, generate, and serve Allure reports."
     )

src/api_testing_framework/client.py

@@ -1,6 +1,8 @@
 from typing import Any, Dict, Optional
 
+import allure
 import httpx
+from httpx import Request, Response
 from tenacity import (
     retry,
     retry_if_exception_type,
@@ -45,6 +47,8 @@ def __init__(
             client_args["transport"] = transport
 
         self._client = httpx.Client(**client_args)
+        self._last_request: Optional[httpx.Request] = None
+        self._last_response: Optional[httpx.Response] = None
 
     def _refresh_token_if_needed(self) -> None:
         """
@@ -58,36 +62,206 @@ def _handle_response(self, response: httpx.Response) -> dict:
             raise APIError(response.status_code, data.get("error", response.text), data)
         return data
 
+    def _attach_last_exchange_to_allure(self) -> None:
+        """
+        Attach the most recent httpx.Request and httpx.Response
+        (stored in self._last_request / self._last_response) into Allure
+        """
+        if not (self._last_request and self._last_response):
+            return
+
+        # Attach HTTP request
+        request: httpx.Request = self._last_request
+        allure.attach(
+            f"{request.method} {request.url}",
+            name="HTTP Request",
+            attachment_type=allure.attachment_type.TEXT,
+        )
+
+        # Attach request headers
+        headers = "\n".join(f"{k}: {v}" for k, v in request.headers.items())
+        allure.attach(
+            headers, name="Request Headers", attachment_type=allure.attachment_type.TEXT
+        )
+
+        # Attach request body
+        if request.content:
+            try:
+                body_text = request.content.decode("utf-8")
+                atype = (
+                    allure.attachment_type.JSON
+                    if body_text.strip().startswith("{")
+                    else allure.attachment_type.TEXT
+                )
+            except Exception:
+                body_text = "<binary_content>"
+                atype = allure.attachment_type.TEXT
+
+            allure.attach(body_text, name="Request Body", attachment_type=atype)
+
+        # Attach response status
+        response: httpx.Response = self._last_response
+        allure.attach(
+            f"{response.status_code} {response.reason_phrase}",
+            name="HTTP Response Status",
+            attachment_type=allure.attachment_type.TEXT,
+        )
+
+        # Attach response headers
+        response_headers = "\n".join(f"{k}: {v}" for k, v in response.headers.items())
+        allure.attach(
+            response_headers,
+            name="Response Headers",
+            attachment_type=allure.attachment_type.TEXT,
+        )
+
+        # Attach response body
+        response_text = response.text or ""
+        try:
+            import json
+
+            json.loads(response_text)
+            atype = allure.attachment_type.JSON
+        except Exception:
+            atype = allure.attachment_type.TEXT
+
+        allure.attach(response_text, name="Response Body", attachment_type=atype)
+
+    def _record_request(self, request: httpx.Request) -> None:
+        """Store the outgoing Request object for later attachment."""
+        self._last_request = request
+
     @retry(
         reraise=True,
         stop=stop_after_attempt(3),
         wait=wait_exponential(multiplier=1, min=1, max=10),
         retry=retry_if_exception_type(APIError),
     )
-    def get(self, path: str, params: Dict[str, Any] = None) -> dict:
-        """GET with retries on APIError"""
+    def get(
+        self, path: str, params: Dict[str, Any] = None, *, attach: bool = False
+    ) -> dict:
+        """
+        GET request; if attach=True, record & attach the request/response in Allure
+        """
         self._refresh_token_if_needed()
-        resp = self._client.get(path, params=params)
-        return self._handle_response(resp)
+
+        # Build request
+        request = self._client.build_request("GET", path, params=params)
+        if attach:
+            self._record_request(request)
+
+        # Send and record response
+        response = self._client.send(request)
+        if attach:
+            self._last_response = response
+
+        # Handle status; if it errors, attach before raising
+        try:
+            data = self._handle_response(response)
+        except APIError:
+            if attach:
+                self._attach_last_exchange_to_allure()
+            raise
+
+        # On success, attach if requested
+        if attach:
+            self._attach_last_exchange_to_allure()
+        return data
 
     @retry(
         reraise=True,
         stop=stop_after_attempt(3),
         wait=wait_exponential(multiplier=1, min=1, max=10),
         retry=retry_if_exception_type(APIError),
     )
-    def post(self, path: str, json: Dict[str, Any] = None) -> dict:
-        """POST with retries on APIError"""
+    def post(
+        self, path: str, json: Dict[str, Any] = None, *, attach: bool = False
+    ) -> dict:
+        """
+        POST request; if attach=True, record & attach the request/response in Allure
+        """
         self._refresh_token_if_needed()
-        resp = self._client.post(path, json=json)
-        return self._handle_response(resp)
 
-    def put(self, path: str, json: Dict[str, Any] = None) -> dict:
+        # Build request
+        request = self._client.build_request("POST", path, json=json)
+        if attach:
+            self._record_request(request)
+
+        # Send and record response
+        response = self._client.send(request)
+        if attach:
+            self._last_response = response
+
+        # Handle status; if errors, attach before raising
+        try:
+            data = self._handle_response(response)
+        except APIError:
+            if attach:
+                self._attach_last_exchange_to_allure()
+            raise
+
+        # On success, attach if requested
+        if attach:
+            self._attach_last_exchange_to_allure()
+        return data
+
+    def put(
+        self, path: str, json: Dict[str, Any] = None, *, attach: bool = False
+    ) -> dict:
+        """
+        PUT request; if attach=True, record & attach the request/response in Allure.
+        """
         self._refresh_token_if_needed()
-        resp = self._client.put(path, json=json)
-        return self._handle_response(resp)
 
-    def delete(self, path: str) -> dict:
+        # Build the request
+        request = self._client.build_request("PUT", path, json=json)
+        if attach:
+            self._record_request(request)
+
+        # Send and record response
+        response = self._client.send(request)
+        if attach:
+            self._last_response = response
+
+        # Handle status; if it errors, attach before raising
+        try:
+            data = self._handle_response(response)
+        except APIError:
+            if attach:
+                self._attach_last_exchange_to_allure()
+            raise
+
+        # On success, attach if requested
+        if attach:
+            self._attach_last_exchange_to_allure()
+        return data
+
+    def delete(self, path: str, *, attach: bool = False) -> dict:
+        """
+        DELETE request; if attach=True, record & attach the request/response in Allure.
+        """
+
         self._refresh_token_if_needed()
-        resp = self._client.delete(path)
-        return self._handle_response(resp)
+
+        # Build request
+        request = self._client.build_request("DELETE", path)
+        if attach:
+            self._record_request(request)
+
+        # Send and record response
+        response = self._client.send(request)
+        if attach:
+            self._last_response = response
+
+        # Handle status; if it errors, attach before raising
+        try:
+            data = self._handle_response(response)
+        except APIError:
+            if attach:
+                self._attach_last_exchange_to_allure()
+            raise
+
+        # On success, attach if requested
+        if attach:
+            self._attach_last_exchange_to_allure()
+        return data

src/api_testing_framework/spotify/client.py

@@ -57,7 +57,7 @@ def get_new_releases(self, limit: int = 20) -> NewReleasesResponse:
         """
         Fetch new album releases from Spotify and return a validated model.
         """
-        raw = self.get(f"/browse/new-releases?limit={limit}")
+        raw = self.get(f"/browse/new-releases?limit={limit}", attach=False)
         return NewReleasesResponse.model_validate(raw)
 
     def get_artist_top_tracks(
@@ -66,5 +66,5 @@ def get_artist_top_tracks(
         """
         Fetch the top tracks for a given artist in the specified market.
         """
-        raw = self.get(f"/artists/{artist_id}/top-tracks?market={market}")
+        raw = self.get(f"/artists/{artist_id}/top-tracks?market={market}", attach=False)
         return TopTracksResponse.model_validate(raw)

tests/test_client.py

@@ -1,14 +1,23 @@
+import allure
 import httpx
 import pytest
 
 from api_testing_framework.client import APIClient
+from api_testing_framework.exceptions import APIError
 
 
 class DummyTransport(httpx.BaseTransport):
     def handle_request(self, request):
         return httpx.Response(200, json={"ok": True})
 
 
+class ErrorTransport(httpx.BaseTransport):
+    """A dummy transport that always returns HTTP 500 with a JSON body"""
+
+    def handle_request(self, request):
+        return httpx.Response(500, json={"error": "internal_server_error"})
+
+
 @pytest.fixture
 def client():
     return APIClient(
@@ -22,3 +31,50 @@ def client():
 def test_get_returns_response(client):
     data = client.get("/foo")
     assert data == {"ok": True}
+
+
+def test_allure_attachment(client):
+    data = client.get("/bar", attach=True)
+    assert data == {"ok": True}
+
+
+def test_allure_without_attachment(client):
+    data = client.get("/foo")
+    assert data == {"ok": True}
+
+
+def test_attach_on_500_response(monkeypatch):
+    """
+    Verify that when the server returns a 500, calling client.get(..., attach=True)
+    raises APIError and triggers Allure attachments for request/response.
+    """
+    attached = []
+
+    def fake_attach(content, name, attachment_type):
+        attached.append((name, content))
+
+    monkeypatch.setattr(allure, "attach", fake_attach)
+
+    client = APIClient(
+        base_url="https://api.example.com",
+        transport=ErrorTransport(),
+        token="dummy-token",
+    )
+
+    with pytest.raises(APIError):
+        client.get("/endpoint-that-errors", attach=True)
+
+    assert attached, "Expected at least one Allure attachment on 500 error"
+
+    attachment_names = {name for (name, _) in attached}
+    expected_names = {
+        "HTTP Request",
+        "Request Headers",
+        "HTTP Response Status",
+        "Response Headers",
+        "Response Body",
+    }
+
+    assert expected_names.issubset(
+        attachment_names
+    ), f"Missing attachment(s): {expected_names - attachment_names}"