diff --git a/Makefile b/Makefile
index 031f948..6c98320 100644
--- a/Makefile
+++ b/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
\ No newline at end of file
+all: clean install results report
\ No newline at end of file
diff --git a/pytest.ini b/pytest.ini
index 7bb600f..4bd91c6 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,4 +1,5 @@
[pytest]
addopts = --alluredir=allure-results
markers =
- integration: mark test as an integration test (vs. unit)
\ No newline at end of file
+ integration: mark test as an integration test (vs. unit)
+ unit: mark test as a unit test
\ No newline at end of file
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..5d07544
--- /dev/null
+++ b/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=
(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 ''.
+#
+# 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(""), "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(""), "Response Body was not truncated."
diff --git a/tests/test_manual_attach.py b/tests/test_manual_attach.py
new file mode 100644
index 0000000..ab02062
--- /dev/null
+++ b/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
diff --git a/tests/test_per_call_attach.py b/tests/test_per_call_attach.py
new file mode 100644
index 0000000..2ffef61
--- /dev/null
+++ b/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}"
diff --git a/tests/utils_allure.py b/tests/utils_allure.py
new file mode 100644
index 0000000..be62254
--- /dev/null
+++ b/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