str:
+ """Best-effort human label for the SDK method exercised, from method + path."""
+ try:
+ path = url.split("://", 1)[-1]
+ path = "/" + path.split("/", 1)[1] if "/" in path else path
+ except (IndexError, AttributeError):
+ path = url
+ # Strip /v3 prefix for readability.
+ if "/v3/" in path:
+ path = path.split("/v3", 1)[1]
+ m = method.upper()
+ table = [
+ (r"/user-session", "POST", "client.login()"),
+ (r"/user-session", "DELETE", "client.logout()"),
+ (r"/user", "GET", "client.user().fetch()"),
+ (r"/stacks", "POST", "client.stack().create()"),
+ (r"/stacks", "GET", "client.stack().fetch()"),
+ (r"/content_types", "POST", "stack.content_types().create()"),
+ (r"/content_types", "GET", "stack.content_types().find()"),
+ (r"/entries", "POST", "content_type.entry().create()"),
+ (r"/entries", "GET", "content_type.entry().find()"),
+ (r"/assets", "POST", "stack.assets().upload()"),
+ (r"/assets", "GET", "stack.assets().find()"),
+ (r"/global_fields", "POST", "stack.global_fields().create()"),
+ (r"/global_fields", "GET", "stack.global_fields().find()"),
+ (r"/environments", "POST", "stack.environments().create()"),
+ (r"/locales", "POST", "stack.locale().create()"),
+ (r"/webhooks", "POST", "stack.webhooks().create()"),
+ (r"/workflows", "POST", "stack.workflows().create()"),
+ (r"/taxonomies", "POST", "stack.taxonomy().create()"),
+ (r"/releases", "POST", "stack.releases().create()"),
+ ]
+ import re
+
+ for pat, meth, label in table:
+ if meth == m and re.search(pat + r"(/|$|\?)", path):
+ return label
+ return f"{m} {path}"
+
+
+# Transient network errors get a couple of automatic retries so a single dev11
+# blip (e.g. a read timeout) doesn't red the whole suite.
+_TRANSIENT_RETRIES = 2
+_RETRY_BACKOFF_SECONDS = 3
+
+
+def _request_with_retry(method, url, kwargs):
+ """Call the real requests.request, retrying transient network failures."""
+ import requests.exceptions as rex
+
+ transient = (rex.ConnectionError, rex.ConnectTimeout, rex.ReadTimeout, rex.Timeout)
+ last_exc = None
+ for attempt in range(_TRANSIENT_RETRIES + 1):
+ try:
+ return _original_request(method, url, **kwargs)
+ except transient as exc:
+ last_exc = exc
+ if attempt < _TRANSIENT_RETRIES:
+ time.sleep(_RETRY_BACKOFF_SECONDS)
+ # Re-open any file handles (consumed by the failed multipart attempt).
+ files = kwargs.get("files")
+ if files:
+ for key, val in list(files.items()):
+ if isinstance(val, (tuple, list)):
+ fh = val[1]
+ name = getattr(fh, "name", None)
+ if name and os.path.exists(name):
+ try:
+ fh.close()
+ except Exception:
+ pass
+ kwargs["files"][key] = (val[0], open(name, "rb")) + tuple(val[2:])
+ else:
+ name = getattr(val, "name", None)
+ if name and os.path.exists(name):
+ try:
+ val.close()
+ except Exception:
+ pass
+ kwargs["files"][key] = open(name, "rb")
+ raise last_exc
+
+
+def _patched_request(method, url, **kwargs):
+ """Drop-in for requests.request that records the call, then delegates."""
+ start = time.time()
+ req_headers = kwargs.get("headers") or {}
+ req_data = kwargs.get("data")
+ req_files = kwargs.get("files")
+ record = {
+ "timestamp": start,
+ "method": str(method).upper(),
+ "url": url,
+ "request_headers": {k: _mask(k, str(v)) for k, v in req_headers.items()},
+ "request_body": _truncate(_stringify(req_data)),
+ "has_files": bool(req_files),
+ "status": None,
+ "status_text": None,
+ "response_headers": {},
+ "response_body": None,
+ "duration_ms": None,
+ "curl": _generate_curl(str(method), url, req_headers, req_data),
+ "sdk_method": _detect_sdk_method(str(method), url),
+ "error": None,
+ }
+ try:
+ response = _request_with_retry(method, url, kwargs)
+ record["status"] = response.status_code
+ record["status_text"] = response.reason
+ record["response_headers"] = dict(response.headers)
+ try:
+ body = json.dumps(response.json(), indent=2)
+ except ValueError:
+ body = response.text
+ record["response_body"] = _truncate(body)
+ return response
+ except Exception as exc: # network error, timeout, etc.
+ record["error"] = str(exc)
+ raise
+ finally:
+ record["duration_ms"] = round((time.time() - start) * 1000, 1)
+ _captured.append(record)
+ if len(_captured) > _MAX_BUFFER:
+ _captured.pop(0)
+
+
+def install() -> None:
+ """Monkeypatch requests.request. Idempotent."""
+ global _original_request
+ if _original_request is None:
+ _original_request = requests.request
+ requests.request = _patched_request
+
+
+def uninstall() -> None:
+ """Restore the original requests.request."""
+ global _original_request
+ if _original_request is not None:
+ requests.request = _original_request
+ _original_request = None
+
+
+def clear() -> None:
+ """Clear the capture buffer (called before each test)."""
+ _captured.clear()
+
+
+def get_all() -> list:
+ """All captured calls currently in the buffer."""
+ return list(_captured)
+
+
+def get_last() -> Optional[dict]:
+ """Most recent captured call, or None."""
+ return _captured[-1] if _captured else None
diff --git a/tests/integration/framework/context.py b/tests/integration/framework/context.py
new file mode 100644
index 0000000..1f87a16
--- /dev/null
+++ b/tests/integration/framework/context.py
@@ -0,0 +1,87 @@
+"""
+Shared test context and cross-file data store for the integration suite.
+
+`TestContext` holds the dynamically created stack credentials and auth state for a
+single test run. `test_data` is a shared store used to pass created-resource UIDs
+between test files (e.g. a content type created in test_12 is referenced by the
+entry tests in test_14).
+
+These are exposed to tests through session-scoped fixtures in conftest.py
+(`ctx`, `stack`, `store`) rather than imported directly, keeping the global mutable
+state contained to the framework layer.
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+
+@dataclass
+class TestContext:
+ """Holds auth + dynamically created stack state for one test run."""
+
+ # SDK client
+ client: Any = None
+ stack: Any = None
+
+ # Authentication
+ authtoken: Optional[str] = None
+ user_uid: Optional[str] = None
+
+ # Stack (dynamically created)
+ stack_api_key: Optional[str] = None
+ stack_uid: Optional[str] = None
+ stack_name: Optional[str] = None
+
+ # Management token (dynamically created)
+ management_token: Optional[str] = None
+ management_token_uid: Optional[str] = None
+
+ # Organization (from env)
+ organization_uid: Optional[str] = None
+
+ # Personalize project (dynamically created, optional)
+ personalize_project_uid: Optional[str] = None
+ personalize_project_name: Optional[str] = None
+
+ # Lifecycle flags
+ is_logged_in: bool = False
+ is_dynamic_stack_created: bool = False
+ is_personalize_created: bool = False
+
+
+def _empty_store() -> dict:
+ """Fresh cross-file UID store. One bucket per resource family."""
+ return {
+ "content_types": {},
+ "entries": {},
+ "assets": {},
+ "folders": {},
+ "global_fields": {},
+ "taxonomies": {},
+ "terms": {},
+ "extensions": {},
+ "environments": {},
+ "locales": {},
+ "workflows": {},
+ "webhooks": {},
+ "labels": {},
+ "roles": {},
+ "delivery_tokens": {},
+ "management_tokens": {},
+ "releases": {},
+ "branches": {},
+ "aliases": {},
+ "variant_groups": {},
+ "variants": {},
+ }
+
+
+# Shared, mutable across the whole run. Reset via `reset_store()` in session teardown.
+test_data: dict = _empty_store()
+
+
+def reset_store() -> None:
+ """Clear all cross-file UID buckets (called in session teardown)."""
+ global test_data
+ test_data.clear()
+ test_data.update(_empty_store())
diff --git a/tests/integration/framework/helpers.py b/tests/integration/framework/helpers.py
new file mode 100644
index 0000000..875c8e0
--- /dev/null
+++ b/tests/integration/framework/helpers.py
@@ -0,0 +1,310 @@
+"""
+Test helper utilities for the integration suite.
+
+Provides:
+- unique test-data generators (uids, titles, emails, dates)
+- timing utilities (wait, retry with backoff)
+- response-shape validators (one per resource family)
+- error-response assertions tuned to the Python SDK (which returns a
+ requests.Response on HTTP errors rather than raising)
+- tracked_assert: records expected/actual into the active test's report tracker
+
+Python equivalent of the JS suite's testHelpers.js.
+"""
+
+import random
+import re
+import string
+import time
+from typing import Any, Iterable
+
+# ---------------------------------------------------------------------------
+# Configurable delays (seconds) — reused across dependent operations to reduce
+# flakiness against a live API.
+# ---------------------------------------------------------------------------
+SHORT_DELAY = 2
+API_DELAY = 5
+LONG_DELAY = 10
+
+
+# ---------------------------------------------------------------------------
+# Generators
+# ---------------------------------------------------------------------------
+def short_id(length: int = 5) -> str:
+ """Short lowercase alphanumeric suffix, e.g. 'a1b2c'."""
+ alphabet = string.ascii_lowercase + string.digits
+ return "".join(random.choice(alphabet) for _ in range(length))
+
+
+def generate_valid_uid(prefix: str = "test") -> str:
+ """Contentstack-valid uid: lowercase, underscore-separated."""
+ return f"{prefix}_{short_id()}".lower()
+
+
+def generate_unique_title(base: str = "Test") -> str:
+ return f"{base} {short_id()}"
+
+
+def generate_random_email() -> str:
+ return f"test_{short_id(8)}@example.com"
+
+
+def generate_future_date(days_from_now: int = 7) -> str:
+ from datetime import datetime, timedelta, timezone
+
+ return (datetime.now(timezone.utc) + timedelta(days=days_from_now)).isoformat()
+
+
+def generate_past_date(days_ago: int = 7) -> str:
+ from datetime import datetime, timedelta, timezone
+
+ return (datetime.now(timezone.utc) - timedelta(days=days_ago)).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Timing
+# ---------------------------------------------------------------------------
+def wait(seconds: float) -> None:
+ time.sleep(seconds)
+
+
+def retry(fn, attempts: int = 3, delay: float = 1.0):
+ """Retry `fn` with linear backoff. Re-raises the last error on exhaustion."""
+ last_error = None
+ for attempt in range(1, attempts + 1):
+ try:
+ return fn()
+ except Exception as exc: # noqa: BLE001 - retry is intentionally broad
+ last_error = exc
+ if attempt < attempts:
+ time.sleep(delay * attempt)
+ raise last_error
+
+
+# ---------------------------------------------------------------------------
+# JSON body helper
+# ---------------------------------------------------------------------------
+def body(response) -> dict:
+ """Parse a requests.Response JSON body, returning {} on non-JSON."""
+ try:
+ return response.json()
+ except ValueError:
+ return {}
+
+
+def wait_for_scan(stack, asset_uid, expected, timeout=40, interval=3):
+ """Poll an asset's _asset_scan_status (AM 2.0) until it reaches `expected`.
+
+ Requires the include_asset_scan_status=true query param to surface the field.
+ Returns the last observed status (the caller asserts == expected).
+ """
+ deadline = time.time() + timeout
+ last = None
+ while time.time() < deadline:
+ asset = stack.assets(asset_uid)
+ asset.add_param("include_asset_scan_status", "true")
+ last = body(asset.fetch()).get("asset", {}).get("_asset_scan_status")
+ if last == expected:
+ return last
+ time.sleep(interval)
+ return last
+
+
+# ---------------------------------------------------------------------------
+# Status / error assertions (Python SDK does NOT raise on HTTP errors)
+# ---------------------------------------------------------------------------
+def assert_status(response, *expected_codes: int):
+ """Assert response.status_code is one of expected_codes (helpful message)."""
+ assert response.status_code in expected_codes, (
+ f"Expected status in {expected_codes}, got {response.status_code}. "
+ f"Body: {body(response)}"
+ )
+
+
+def assert_success(response):
+ """Assert a 2xx success status."""
+ assert 200 <= response.status_code < 300, (
+ f"Expected 2xx success, got {response.status_code}. Body: {body(response)}"
+ )
+
+
+def validate_error_body(response):
+ """Assert the error body carries the standard Contentstack error shape."""
+ data = body(response)
+ assert "error_message" in data or "error_code" in data, (
+ f"Expected error body with error_message/error_code, got: {data}"
+ )
+ return data
+
+
+# ---------------------------------------------------------------------------
+# Response-shape validators
+# ---------------------------------------------------------------------------
+def _require_keys(obj: dict, keys: Iterable[str], label: str):
+ for key in keys:
+ assert key in obj, f"{label}: expected key '{key}' in response, got keys {list(obj)}"
+
+
+def validate_content_type_response(response, expected_uid: str = None):
+ data = body(response).get("content_type", body(response))
+ _require_keys(data, ["uid", "title", "schema"], "content_type")
+ assert isinstance(data["schema"], list)
+ assert re.match(r"^[a-z][a-z0-9_]*$", data["uid"]), f"bad uid format: {data['uid']}"
+ if expected_uid:
+ assert data["uid"] == expected_uid
+ return data
+
+
+def validate_entry_response(response, content_type_uid: str = None):
+ data = body(response).get("entry", body(response))
+ _require_keys(data, ["uid", "title", "locale"], "entry")
+ assert re.match(r"^blt[a-f0-9]+$", data["uid"]), f"bad entry uid: {data['uid']}"
+ return data
+
+
+def validate_asset_response(response):
+ data = body(response).get("asset", body(response))
+ _require_keys(data, ["uid", "filename", "url", "content_type"], "asset")
+ assert re.match(r"^(blt|am)[a-f0-9]+$", data["uid"]), f"bad asset uid: {data['uid']}"
+ return data
+
+
+def validate_global_field_response(response, expected_uid: str = None):
+ data = body(response).get("global_field", body(response))
+ _require_keys(data, ["uid", "title", "schema"], "global_field")
+ assert isinstance(data["schema"], list)
+ if expected_uid:
+ assert data["uid"] == expected_uid
+ return data
+
+
+def validate_locale_response(response):
+ data = body(response).get("locale", body(response))
+ _require_keys(data, ["code", "name"], "locale")
+ return data
+
+
+def validate_environment_response(response):
+ data = body(response).get("environment", body(response))
+ _require_keys(data, ["uid", "name"], "environment")
+ return data
+
+
+def validate_workflow_response(response):
+ data = body(response).get("workflow", body(response))
+ _require_keys(data, ["uid", "name", "workflow_stages"], "workflow")
+ assert isinstance(data["workflow_stages"], list) and data["workflow_stages"]
+ return data
+
+
+def validate_webhook_response(response):
+ data = body(response).get("webhook", body(response))
+ _require_keys(data, ["uid", "name", "destinations", "channels"], "webhook")
+ return data
+
+
+def validate_role_response(response):
+ data = body(response).get("role", body(response))
+ # Permission rules are returned as 'rules' on create but 'SYS_ACL' on fetch,
+ # so only uid + name are guaranteed across both shapes.
+ _require_keys(data, ["uid", "name"], "role")
+ return data
+
+
+def validate_release_response(response):
+ data = body(response).get("release", body(response))
+ _require_keys(data, ["uid", "name"], "release")
+ return data
+
+
+def validate_token_response(response):
+ data = body(response).get("token", body(response))
+ _require_keys(data, ["uid", "name", "token"], "token")
+ return data
+
+
+def validate_branch_response(response):
+ data = body(response).get("branch", body(response))
+ _require_keys(data, ["uid", "source"], "branch")
+ return data
+
+
+def validate_taxonomy_response(response):
+ data = body(response).get("taxonomy", body(response))
+ _require_keys(data, ["uid", "name"], "taxonomy")
+ return data
+
+
+def validate_term_response(response):
+ data = body(response).get("term", body(response))
+ _require_keys(data, ["uid", "name"], "term")
+ return data
+
+
+# ---------------------------------------------------------------------------
+# tracked_assert — records expected/actual to the active test's report tracker.
+# The tracker is set by conftest before each test; if absent, this degrades to a
+# plain assert so helpers remain usable outside a pytest run.
+# ---------------------------------------------------------------------------
+_active_tracker = None
+
+
+def set_active_tracker(tracker) -> None:
+ global _active_tracker
+ _active_tracker = tracker
+
+
+def _record(description: str, expected: Any, actual: Any, passed: bool):
+ if _active_tracker is not None:
+ _active_tracker.append(
+ {
+ "description": description,
+ "expected": _fmt(expected),
+ "actual": _fmt(actual),
+ "passed": passed,
+ }
+ )
+
+
+def _fmt(value: Any) -> str:
+ text = repr(value)
+ return text if len(text) <= 200 else text[:200] + "..."
+
+
+def tracked_assert(actual: Any, description: str = ""):
+ """Chainable tracked assertion, e.g.
+ tracked_assert(resp.status_code, 'status').equals(201)
+ """
+
+ class _Chain:
+ def equals(self, expected):
+ passed = actual == expected
+ _record(description or "equals", expected, actual, passed)
+ assert passed, f"{description}: expected {expected!r}, got {actual!r}"
+ return self
+
+ def is_in(self, options):
+ passed = actual in options
+ _record(description or "is_in", options, actual, passed)
+ assert passed, f"{description}: expected one of {options!r}, got {actual!r}"
+ return self
+
+ def is_type(self, typ):
+ passed = isinstance(actual, typ)
+ _record(description or "is_type", typ.__name__, type(actual).__name__, passed)
+ assert passed, f"{description}: expected {typ.__name__}, got {type(actual).__name__}"
+ return self
+
+ def truthy(self):
+ passed = bool(actual)
+ _record(description or "truthy", "truthy", actual, passed)
+ assert passed, f"{description}: expected truthy, got {actual!r}"
+ return self
+
+ def matches(self, pattern):
+ passed = bool(re.match(pattern, str(actual)))
+ _record(description or "matches", pattern, actual, passed)
+ assert passed, f"{description}: {actual!r} does not match {pattern}"
+ return self
+
+ return _Chain()
diff --git a/tests/integration/framework/report.py b/tests/integration/framework/report.py
new file mode 100644
index 0000000..4e6cb7b
--- /dev/null
+++ b/tests/integration/framework/report.py
@@ -0,0 +1,462 @@
+"""
+Custom dashboard HTML report for the integration suite.
+
+Pure Python, zero dependencies. Collects a TestRecord per test (from conftest hooks)
+and renders a single self-contained HTML file at session finish.
+
+Design goals (modelled on the JS suite's Mochawesome output, not a copy):
+ - compact summary chips (no oversized cards or full-width bars)
+ - human-readable test titles (from docstring, else derived from the test name),
+ grouped resource -> describe-block (class) -> test
+ - each test expands to show every HTTP call with method/url, a colour-coded
+ status, request headers, request body, response body, and copy-ready cURL —
+ each in a collapsible block like Mochawesome's context entries
+ - HONEST status: a test that passed its assertions but whose HTTP calls returned
+ a >=400 status is flagged amber ("passed with warnings") rather than silent green,
+ so lenient/tolerant assertions can't hide a real API failure.
+"""
+
+import html
+import os
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import List
+
+
+@dataclass
+class TestRecord:
+ nodeid: str
+ name: str
+ resource: str
+ outcome: str # 'passed' | 'failed' | 'skipped'
+ duration_ms: float
+ message: str = ""
+ calls: List[dict] = field(default_factory=list)
+ assertions: List[dict] = field(default_factory=list)
+ description: str = ""
+ group: str = "" # test class / describe block
+
+ # Keywords/markers identifying a deliberately-negative test, where a >=400
+ # response is the EXPECTED, asserted outcome (so it is not a warning).
+ _NEGATIVE_MARKERS = (
+ "negative", "nonexistent", "invalid", "without", "duplicate", "missing",
+ "requires", "bad", "no_such", "raises", "not_found", "404",
+ )
+
+ @property
+ def is_negative(self) -> bool:
+ hay = f"{self.group} {self.name}".lower()
+ return any(m in hay for m in self._NEGATIVE_MARKERS)
+
+ @property
+ def has_warning(self) -> bool:
+ """Passed a positive test, but an HTTP call returned >= 400.
+
+ Negative tests are excluded: for them a 4xx is the expected, asserted
+ result. This flags only positive operations whose lenient assertions
+ swallowed an unexpected API error (e.g. a create that returned 422).
+ """
+ if self.outcome != "passed" or self.is_negative:
+ return False
+ return any((c.get("status") or 0) >= 400 for c in self.calls)
+
+
+# ---------------------------------------------------------------------------
+# Title humanisation
+# ---------------------------------------------------------------------------
+def _humanize_test(name: str) -> str:
+ n = name[5:] if name.startswith("test_") else name
+ n = n.replace("_", " ").strip()
+ return (n[:1].upper() + n[1:]) if n else name
+
+
+def _humanize_group(cls: str) -> str:
+ # 'TestContentTypeCRUD' -> 'Content Type CRUD'
+ s = cls[4:] if cls.startswith("Test") else cls
+ out, prev_lower = [], False
+ for i, ch in enumerate(s):
+ if ch.isupper() and prev_lower:
+ out.append(" ")
+ out.append(ch)
+ prev_lower = ch.islower()
+ return "".join(out).strip()
+
+
+def _region_from_host(host: str) -> str:
+ if not host:
+ return "NA"
+ head = host.split("-api")[0] if "-api" in host else ""
+ return head.upper() if head and head != "api" else "NA"
+
+
+def _esc(text) -> str:
+ return html.escape(str(text)) if text is not None else ""
+
+
+# ---------------------------------------------------------------------------
+# Render
+# ---------------------------------------------------------------------------
+def render(records: List[TestRecord], output_path: str, started_at: float, ended_at: float) -> str:
+ total = len(records)
+ passed = sum(1 for r in records if r.outcome == "passed")
+ failed = sum(1 for r in records if r.outcome == "failed")
+ skipped = sum(1 for r in records if r.outcome == "skipped")
+ xfailed = sum(1 for r in records if r.outcome == "xfailed")
+ xpassed = sum(1 for r in records if r.outcome == "xpassed")
+ warnings = sum(1 for r in records if r.has_warning) + xpassed
+ pass_rate = round((passed / total) * 100, 1) if total else 0.0
+ duration_s = round(ended_at - started_at, 1)
+ host = os.getenv("HOST", "")
+ region = _region_from_host(host)
+ timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+
+ # Group: resource -> [records]
+ by_resource: dict = {}
+ for r in records:
+ by_resource.setdefault(r.resource, []).append(r)
+
+ # Coverage chips (compact).
+ cov = []
+ for resource in sorted(by_resource):
+ items = by_resource[resource]
+ p = sum(1 for r in items if r.outcome == "passed")
+ f = sum(1 for r in items if r.outcome == "failed")
+ w = sum(1 for r in items if r.has_warning)
+ cls = "ok" if f == 0 else "bad"
+ warn = f' ⚠{w}' if w else ""
+ cov.append(
+ f'{_esc(resource)}'
+ f'{p}/{len(items)}{warn}
'
+ )
+
+ # Sections: failing/warning resources first.
+ def sort_key(resource):
+ items = by_resource[resource]
+ rank = 0 if any(r.outcome == "failed" for r in items) else (1 if any(r.has_warning for r in items) else 2)
+ return (rank, resource)
+
+ sections = []
+ for resource in sorted(by_resource, key=sort_key):
+ sections.append(_render_resource(resource, by_resource[resource]))
+
+ document = _TEMPLATE.format(
+ total=total, passed=passed, failed=failed, skipped=skipped, warnings=warnings,
+ xfailed=xfailed, pass_rate=pass_rate, duration=duration_s, region=_esc(region),
+ host=_esc(host or "n/a"), timestamp=timestamp,
+ coverage="".join(cov), sections="".join(sections) or "No tests recorded.
",
+ )
+ os.makedirs(os.path.dirname(output_path), exist_ok=True)
+ with open(output_path, "w", encoding="utf-8") as fh:
+ fh.write(document)
+ return output_path
+
+
+def _render_resource(resource: str, items: List[TestRecord]) -> str:
+ p = sum(1 for r in items if r.outcome == "passed")
+ f = sum(1 for r in items if r.outcome == "failed")
+ w = sum(1 for r in items if r.has_warning)
+ s = sum(1 for r in items if r.outcome == "skipped")
+
+ badges = f'{p} passed'
+ if f:
+ badges += f'{f} failed'
+ if w:
+ badges += f'{w} warning'
+ if s:
+ badges += f'{s} skipped'
+
+ # Group by class within resource.
+ by_group: dict = {}
+ for r in items:
+ by_group.setdefault(r.group or "", []).append(r)
+
+ body = []
+ for group in by_group:
+ if group:
+ body.append(f'{_esc(_humanize_group(group))}
')
+ for r in by_group[group]:
+ body.append(_render_test(r))
+
+ open_attr = " open" if (f or w) else ""
+ return (
+ f''
+ f'{_esc(resource)}'
+ f'{badges}
'
+ f'{"".join(body)}
{_esc(r.name)}
']
+ if r.message:
+ parts.append(f'')
+ if r.assertions:
+ parts.append(_render_assertions(r.assertions))
+ if r.calls:
+ parts.append(f'HTTP calls ({len(r.calls)})
'
+ + "".join(_render_call(c) for c in r.calls) + "
'
+ f'{icon}{_esc(title)}'
+ f'{warn_note}{int(r.duration_ms)} ms
'
+ f'{"".join(parts)}
| {"✓" if a.get("passed") else "✗"} | '
+ f'{_esc(a.get("description"))} | {_esc(a.get("expected"))} | '
+ f'{_esc(a.get("actual"))} |
'
+ for a in assertions
+ )
+ return (
+ 'Assertions
'
+ '
| Check | Expected | Actual |
|---|
'
+ f'{rows}
'
+ '
'
+ f'{_esc(method)}'
+ f'{_esc(c.get("url"))}'
+ f'{_esc(slabel)}'
+ f'{_esc(dur)} ms
'
+ f'
{_esc(sdk)}
'
+ f'{"".join(blocks)}
{_esc(label)}
'
+ f'{_esc(content)}
+
CMA Python SDK — Integration Report
+
Region {region} · {host} · {timestamp}
+
+
+
{total}Total
+
{passed}Passed
+
{failed}Failed
+
{warnings}Warnings
+
{xfailed}Known issues
+
{skipped}Skipped
+
{pass_rate}%Pass rate
+
{duration}sDuration
+
+
Coverage by resource
+
{coverage}
+
Test details
+ {sections}
+