Add test infrastructure improvements

Makefile

@@ -14,9 +14,12 @@ lint:
 test:
 	poetry run pytest
 
+# Run tests *with* Allure result output (only writes results)
+results:
+	poetry run pytest --maxfail=1 --disable-warnings --alluredir=allure-results
+
 # Generate Allure results and HTML report
 report:
-	poetry run pytest --alluredir=allure-results
 	allure generate allure-results --clean -o allure-report
 
 # Serve the Allure report interactively
@@ -28,4 +31,4 @@ clean:
 	rm -rf .pytest_cache/ allure-results/ allure-report/
 
 # Full flow: clean state, install, lint, test, and report
-all: clean install test report
+all: clean install results report

pytest.ini

@@ -1,4 +1,5 @@
 [pytest]
 addopts = --alluredir=allure-results
 markers = 
-    integration: mark test as an integration test (vs. unit)
+    integration: mark test as an integration test (vs. unit)
+    unit: mark test as a unit test

tests/conftest.py

@@ -0,0 +1,102 @@
+import os
+import uuid
+
+import allure
+import pytest
+
+from api_testing_framework.client import APIClient
+
+# from tests.utils_allure import find_attachment_path, wait_for_result_with_label
+
+
+@pytest.fixture
+def api_client(request):
+    """
+    Provide a SpotifyClient (or a generic APIClient subclass). If ATTACH_ON_FAILURE is set,
+    APIClient will record each exchane. We attach this instance to the test node so
+    a pytest hook can retrieve and attach exchanges only on failure.
+    """
+
+    client = APIClient(base_url="https://api.example.com", token=None)
+
+    request.node._api_client = client
+    return client
+
+
+def pytest_runtest_makereport(item, call):
+    """
+    After each test's 'call' phase, if the test failed and the client recorded an exchange,
+    attach it to Allure.
+    """
+    if call.when != "call":
+        return
+
+    if call.excinfo is None:
+        return
+
+    api_client = getattr(item, "_api_client", None)
+    if isinstance(api_client, APIClient):
+        api_client._attach_last_exchange_to_allure()
+
+
+@pytest.fixture
+def require_alluredir(pytestconfig):
+    allure_dir = pytestconfig.getoption("--alluredir", default=None)
+    if not allure_dir:
+        pytest.skip(
+            "This test requires --alluredir=<dir> (no Allure artifacts otherwise)"
+        )
+    return allure_dir
+
+
+# @pytest.fixture
+# def assert_truncated_response_after_test(require_alluredir, request):
+#     """
+#     After the test finishes, locate THIS test's result by a label the test sets,
+#     then verify the 'Response Body' attachment ends with '<truncated>'.
+#
+#     NOTE: This fixture has race condition issues in CI where Allure results
+#     may not be flushed to disk before the teardown runs. Commented out until
+#     a more reliable approach is implemented.
+#     """
+#     # The test will set this label on itself (see test code below)
+#     LABEL_NAME = "nodeid"
+#     LABEL_VALUE = request.node.nodeid
+#
+#     yield  # --- test body runs here ---
+#
+#     res = wait_for_result_with_label(
+#         require_alluredir, LABEL_NAME, LABEL_VALUE, timeout=12.0
+#     )
+#     assert res, f"No Allure result found for label {LABEL_NAME}={LABEL_VALUE}."
+#
+#     body_path = find_attachment_path(res, require_alluredir, "Response Body")
+#     assert body_path, "No 'Response Body' attachment found in this test's result."
+#
+#     with open(body_path, "r", encoding="utf-8", errors="ignore") as f:
+#         body = f.read()
+#     assert body.endswith("<truncated>"), "Response body was not truncated."
+
+
+# @pytest.fixture
+# def assert_truncated_after_test(require_alluredir, request):
+#     """
+#     Attach a unique marker at the start (so we can find THIS test's result),
+#     let the test run, then (after teardown) assert the 'Response Body' was truncated.
+#     """
+#     marker = f"ATTACHMENT_MARKER:{uuid.uuid4()}"
+#     allure.attach("marker", name=marker, attachment_type=allure.attachment_type.TEXT)
+
+#     yield  # <-- test body runs here
+
+#     res = wait_for_result_with_marker(require_alluredir, marker, timeout=12.0)
+#     assert res, "No Allure result found for this test (race or mssing --alluredir)."
+
+#     body_path = find_attachment_by_name_in_result(
+#         res, require_alluredir, "Response Body"
+#     )
+#     assert body_path, "No 'Response Body' attachment found in this test's result."
+
+#     with open(body_path, "r", encoding="utf-8", errors="ignore") as f:
+#         body = f.read()
+#     assert body.endswith("<truncated>"), "Response Body was not truncated."

tests/test_manual_attach.py

@@ -0,0 +1,10 @@
+import pytest
+
+from api_testing_framework.client import APIClient
+
+
+@pytest.mark.unit
+def test_manual_attach_using_swapi():
+    client = APIClient(base_url="https://www.swapi.tech/api", token=None)
+    data = client.get("/people", attach=True)
+    assert "total_records" in data

tests/test_per_call_attach.py

@@ -0,0 +1,81 @@
+import allure
+import httpx
+import pytest
+
+from api_testing_framework.client import APIClient
+from api_testing_framework.exceptions import APIError
+
+
+class ErrorTransport(httpx.BaseTransport):
+    """Always return HTTP 500 with a JSON error body."""
+
+    def handle_request(self, request):
+        return httpx.Response(500, json={"error": "server_error"})
+
+
+class BodyTransport(httpx.BaseTransport):
+    """Always returns HTTP 400 with a JSON error body."""
+
+    def handle_request(self, request):
+        return httpx.Response(400, json={"error": "bad_request"})
+
+
+def test_per_call_attach_on_error(monkeypatch):
+    # Capture all attachment names
+    attached = []
+    monkeypatch.setattr(
+        allure, "attach", lambda content, name, attachment_type: attached.append(name)
+    )
+
+    # Create a client that will error
+    client = APIClient(
+        base_url="https://api.example.com",
+        transport=ErrorTransport(),
+        token="dummy-token",
+    )
+
+    # Call with attach=True; expect APIError and attachments recorded immediately
+    with pytest.raises(APIError):
+        client.get("/endpoint-fails", attach=True)
+
+    # Verify we saw the key HTTP attachment names
+    expected = {
+        "HTTP Request",
+        "Request Headers",
+        "HTTP Response Status",
+        "Response Headers",
+        "Response Body",
+    }
+    missing = expected - set(attached)
+    assert not missing, f"Missing attachments: {missing}"
+
+
+def test_per_call_attach_post_with_body(monkeypatch):
+    attached = []
+
+    # Define a fake attach that accepts keyword args
+    def fake_attach(content, name=None, attachment_type=None):
+        attached.append(name)
+
+    monkeypatch.setattr(allure, "attach", fake_attach)
+
+    client = APIClient(
+        base_url="https://api.example.com",
+        transport=BodyTransport(),
+        token="dummy-token",
+    )
+
+    with pytest.raises(APIError):
+        client.post("/fails", json={"foo": "bar"}, attach=True)
+
+    # Now you *should* see Request Body attached, along with the others
+    expected = {
+        "HTTP Request",
+        "Request Headers",
+        "Request Body",
+        "HTTP Response Status",
+        "Response Headers",
+        "Response Body",
+    }
+    missing = expected - set(attached)
+    assert not missing, f"Missing attachments: {missing}"

tests/utils_allure.py

@@ -0,0 +1,121 @@
+import glob
+import json
+import os
+import time
+from typing import Any, Dict, Iterable, Optional
+
+
+def _iter_attachments(node: Dict[str, Any]) -> Iterable[Dict[str, Any]]:
+    for att in node.get("attachments", []) or []:
+        yield att
+    for step in node.get("steps", []) or []:
+        yield from _iter_attachments(step)
+
+
+def _result_has_label(res: Dict[str, Any], name: str, value: str) -> bool:
+    for lab in res.get("labels", []) or []:
+        if lab.get("name") == name and lab.get("value") == value:
+            return True
+    return False
+
+
+def wait_for_result_with_label(
+    allure_dir: str, label_name: str, label_value: str, timeout: float = 12.0
+) -> Optional[Dict[str, Any]]:
+    """Poll allure-results for a test result that has given label."""
+    deadline = time.time() + timeout
+    time.sleep(0.2)
+    pattern = os.path.join(allure_dir, "*-result.json")
+
+    while time.time() < deadline:
+        for path in glob.glob(pattern):
+            try:
+                with open(path, "r", encoding="utf-8") as f:
+                    res = json.load(f)
+            except Exception:
+                continue
+            if _result_has_label(res, label_name, label_value):
+                return res
+        time.sleep(0.1)
+    return None
+
+
+def find_attachment_path(
+    res: Dict[str, Any], allure_dir: str, name: str
+) -> Optional[str]:
+    """Return full path to the first attachment named `name` in this result (top or steps)."""
+    for att in _iter_attachments(res):
+        if att.get("name") == name and att.get("source"):
+            path = os.path.join(allure_dir, att["source"])
+            if os.path.exists(path):
+                return path
+    return None
+
+
+# def _result_has_marker(res: Dict[str, Any], marker_name: str) -> bool:
+#     for att in _iter_attachments(res):
+#         if att.get("name") == marker_name:
+#             return True
+#     return False
+
+
+def find_attachment_by_name_in_result(
+    res: Dict[str, Any], allure_dir: str, attachment_name: str
+) -> Optional[str]:
+    """Return full path to the first attachment named `attachment_name` in this result."""
+    for att in _iter_attachments(res):
+        if att.get("name") == attachment_name:
+            src = att.get("source")
+            if src:
+                path = os.path.join(allure_dir, src)
+                if os.path.exists(path):
+                    return path
+    return None
+
+
+# def wait_for_result_with_marker(
+#     allure_dir: str, marker_name: str, timeout: float = 10.0
+# ) -> Optional[Dict[str, Any]]:
+#     """Poll allure-results for a test result that contains our marker attachment."""
+
+#     deadline = time.time() + timeout
+#     # small delay helps when tests are very fast
+#     time.sleep(0.2)
+#     while time.time() < deadline:
+#         for path in glob.glob(os.path.join(allure_dir, "*-result.json")):
+#             try:
+#                 with open(path, "r", encoding="utf-8") as f:
+#                     res = json.load(f)
+#             except Exception:
+#                 continue
+#             if _result_has_marker(res, marker_name):
+#                 return res
+#         time.sleep(0.1)
+#     return None
+
+
+def find_response_body_attachment(
+    allure_dir: str, name_hint: str, timeout: float = 5.0
+):
+    """Wait up to `timeout` seconds for Allure result whose name/fullName
+    contains `name_hint`; then return the full path to the 'Response Body' attachment.
+    """
+
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        result_files = glob.glob(os.path.join(allure_dir, "*-result.json"))
+        for path in result_files:
+            try:
+                with open(path, "r", encoding="utf-8") as f:
+                    res = json.load(f)
+            except Exception:
+                continue
+            name = res.get("name") or res.get("fullName") or ""
+            if name_hint in name:
+                for att in res.get("attachments", []):
+                    if att.get("name") == "Response Body":
+                        src = os.path.join(allure_dir, att.get("source", ""))
+                        if os.path.exists(src):
+                            return src
+        time.sleep(0.1)
+    return None