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