feat(attachments): attach logs, HAR files, videos, traces and HTML to tests

README.md

@@ -57,6 +57,28 @@ execution = angles_reporter.save_test()
 ```
 
 
+## Attachments
+
+A test can attach files to its results: console logs, network HAR files, videos, Playwright traces, page HTML snapshots and images. Angles shows each one on the test (or the step) with a viewer that suits it. The file extension decides how it is shown, so keep the real one: `.log`/`.txt`, `.json`, `.har`, `.webm`/`.mp4`, `.zip` (shown as a Playwright trace when the name contains "trace"), `.html`/`.htm`, `.png`/`.jpg`/`.jpeg`/`.gif`/`.webp`. Requires an Angles server with test attachment support.
+
+```python
+angles_reporter.fail_step("Order confirmation", "Order confirmed", "Payment declined", "")
+
+# attach to the step you just reported, e.g. the page as it was when the assertion failed
+angles_reporter.attach_data_to_last_step(page.content(), "page.html")
+angles_reporter.attach_file_to_last_step("artifacts/failure.png")
+
+# attach to the whole test
+angles_reporter.attach_file(page.video.path(), "checkout.webm")
+angles_reporter.attach_file("artifacts/trace.zip")
+angles_reporter.attach_file("artifacts/network.har")
+angles_reporter.attach_data("\n".join(console_lines), "console.log")
+
+angles_reporter.save_test()
+```
+
+Uploads happen straight away (they only need the build id), so this works in batch mode too. A rejected upload (e.g. an unsupported extension) raises `AnglesApiError`.
+
 ## Batch mode
 
 By default every call to `save_test()` sends the test execution to the Angles API straight away. If you'd rather send the whole test run in a single request at the end (e.g. for large runs), you can enable batch mode. The build is still created up-front and screenshots are still uploaded individually as the tests run (they need the build id), but the executions are gathered by the reporter until you call `save_all_tests()`.

angles_python_client/__init__.py

@@ -4,7 +4,7 @@
 
 - A singleton reporter (`angles_reporter`) similar to the JS default export.
 - Request classes: BuildRequests, TeamRequests, EnvironmentRequests, ScreenshotRequests,
-  ExecutionRequests, BaselineRequests, MetricRequests, AnglesRequests.
+  ExecutionRequests, BaselineRequests, MetricRequests, AnglesRequests, AttachmentRequests.
 """
 
 from .http import AnglesHttpClient
@@ -18,6 +18,7 @@
     BaselineRequests,
     MetricRequests,
     AnglesRequests,
+    AttachmentRequests,
 )
 
 __all__ = [
@@ -32,4 +33,5 @@
     "BaselineRequests",
     "MetricRequests",
     "AnglesRequests",
+    "AttachmentRequests",
 ]

angles_python_client/models/execution.py

@@ -23,4 +23,5 @@ class Execution:
     platforms: Optional[List[Platform]] = None
     tags: Optional[List[str]] = None
     meta: Optional[Dict[str, Any]] = None
+    attachments: Optional[List[str]] = None
     status: Optional[ExecutionStates] = None

angles_python_client/models/requests.py

@@ -29,6 +29,8 @@ class CreateExecution:
     platforms: Optional[List[Platform]] = None
     tags: Optional[List[str]] = None
     meta: Optional[Dict[str, Any]] = None
+    # Ids of files attached to the whole test (see AnglesReporter.attach_file).
+    attachments: Optional[List[str]] = None
 
 
 @dataclass

angles_python_client/models/step.py

@@ -2,7 +2,7 @@
 
 import datetime as _dt
 from dataclasses import dataclass
-from typing import Optional
+from typing import List, Optional
 
 from .enums import StepStates
 
@@ -16,3 +16,5 @@ class Step:
     status: Optional[StepStates] = None
     timestamp: Optional[_dt.datetime] = None
     screenshot: Optional[str] = None
+    # Ids of files attached to this step (see AnglesReporter.attach_file_to_last_step).
+    attachments: Optional[List[str]] = None

angles_python_client/reporter.py

@@ -12,6 +12,7 @@
     EnvironmentRequests,
     ExecutionRequests,
     ScreenshotRequests,
+    AttachmentRequests,
 )
 
 
@@ -55,6 +56,7 @@ def _instantiate_clients(self) -> None:
         self.builds = BuildRequests(self.http)
         self.executions = ExecutionRequests(self.http)
         self.screenshots = ScreenshotRequests(self.http)
+        self.attachments = AttachmentRequests(self.http)
 
     @classmethod
     def get_instance(cls) -> "AnglesReporter":
@@ -196,6 +198,57 @@ def compare_screenshot_against_baseline(self, screenshot_id: str) -> Any:
         return self.screenshots.get_baseline_compare(screenshot_id)
 
     # --- steps/actions ---
+    # --- attachments ---
+    def attach_file(self, file_path: str, file_name: Optional[str] = None) -> Dict[str, Any]:
+        """Uploads a file and attaches it to the current test: a video, a Playwright trace,
+        a HAR file, a console log and so on.
+
+        The extension decides how Angles shows it: ``.log``/``.txt``, ``.json``, ``.har``,
+        ``.webm``/``.mp4``, ``.zip`` (a trace when the name contains "trace"), ``.html``/``.htm``
+        or an image. ``file_name`` shows the file under another name (keep the extension).
+        """
+        execution = self._require_execution("attach_file")
+        attachment = self.attachments.upload_test_attachment(self.current_build["_id"], file_path, file_name)
+        execution.attachments = (execution.attachments or []) + [attachment["_id"]]
+        return attachment
+
+    def attach_data(self, data: Any, file_name: str) -> Dict[str, Any]:
+        """Attaches in-memory content (``bytes`` or ``str``) to the current test."""
+        execution = self._require_execution("attach_data")
+        attachment = self.attachments.upload_test_attachment_data(self.current_build["_id"], data, file_name)
+        execution.attachments = (execution.attachments or []) + [attachment["_id"]]
+        return attachment
+
+    def attach_file_to_last_step(self, file_path: str, file_name: Optional[str] = None) -> Dict[str, Any]:
+        """Uploads a file and attaches it to the most recent step, e.g. a screenshot taken
+        when an assertion failed."""
+        step = self._require_last_step("attach_file_to_last_step")
+        attachment = self.attachments.upload_test_attachment(self.current_build["_id"], file_path, file_name)
+        step.attachments = (step.attachments or []) + [attachment["_id"]]
+        return attachment
+
+    def attach_data_to_last_step(self, data: Any, file_name: str) -> Dict[str, Any]:
+        """Attaches in-memory content to the most recent step, e.g.
+        ``attach_data_to_last_step(page.content(), "page.html")`` right after ``fail_step()``."""
+        step = self._require_last_step("attach_data_to_last_step")
+        attachment = self.attachments.upload_test_attachment_data(self.current_build["_id"], data, file_name)
+        step.attachments = (step.attachments or []) + [attachment["_id"]]
+        return attachment
+
+    def _require_execution(self, method: str) -> CreateExecution:
+        if not self.current_build or not self.current_build.get("_id"):
+            raise RuntimeError(f"{method}: no current build set. Call start_build() or set_current_build() first.")
+        if not self.current_execution:
+            raise RuntimeError(f"{method}: no current test started. Call start_test() first.")
+        return self.current_execution
+
+    def _require_last_step(self, method: str) -> Step:
+        self._require_execution(method)
+        steps = self.current_action.steps if self.current_action else None
+        if not steps:
+            raise RuntimeError(f"{method}: add a step (pass_step, fail_step, info, ...) before attaching a file to it.")
+        return steps[-1]
+
     def add_action(self, name: str) -> None:
         self.current_action = Action(name=name, start=_dt.datetime.now(), steps=[])
         if self.current_execution is None:

angles_python_client/requests.py

@@ -316,6 +316,58 @@ def find_uploaded_image_in_screenshot(self, screenshot_id: str, template_file_pa
             return resp.json() if resp.content else None
 
 
+class AttachmentRequests(BaseRequests):
+    """Files an automated test produced: logs, HAR files, videos, traces, HTML snapshots, images.
+
+    The server decides how a file is shown from its extension, so keep the real one:
+    ``.log``/``.txt``, ``.json``, ``.har``, ``.webm``/``.mp4``, ``.zip`` (a Playwright trace
+    when the name contains "trace"), ``.html``/``.htm``, ``.png``/``.jpg``/``.jpeg``/``.gif``/``.webp``.
+    """
+
+    # Videos and traces can be large; the client's default request timeout is for JSON calls.
+    UPLOAD_TIMEOUT_S = 300.0
+
+    def upload_test_attachment(self, build_id: str, file_path: str, file_name: Optional[str] = None) -> Any:
+        """Uploads a file from disk against the build the test is running in."""
+        import os
+
+        full_path = os.path.abspath(file_path)
+        with open(full_path, "rb") as f:
+            return self._upload(build_id, f, file_name or os.path.basename(full_path))
+
+    def upload_test_attachment_data(self, build_id: str, data: Any, file_name: str) -> Any:
+        """Uploads in-memory content (``bytes`` or ``str``). ``file_name``'s extension decides the kind."""
+        if isinstance(data, str):
+            data = data.encode("utf-8")
+        return self._upload(build_id, data, file_name)
+
+    def _upload(self, build_id: str, content: Any, file_name: str) -> Any:
+        files = {"attachment": (file_name, content, "application/octet-stream")}
+        resp = self.http.request(
+            "POST",
+            f"build/{build_id}/attachment",
+            files=files,
+            headers={"Accept": "application/json"},
+            timeout_s=self.UPLOAD_TIMEOUT_S,
+        )
+        return resp.json() if resp.content else None
+
+    def get_attachments(self, execution_id: Optional[str] = None, build_id: Optional[str] = None) -> Any:
+        """Lists the files an execution claimed, or everything uploaded against a build."""
+        params: Dict[str, Any] = {}
+        if execution_id:
+            params["executionId"] = execution_id
+        if build_id:
+            params["buildId"] = build_id
+        return self.get("attachment", params=params)
+
+    def get_attachment_file(self, attachment_id: str) -> bytes:
+        return self.get(f"attachment/{attachment_id}/file", response_type="bytes")
+
+    def delete_attachment(self, attachment_id: str) -> Any:
+        return self.delete(f"attachment/{attachment_id}")
+
+
 class BaselineRequests(BaseRequests):
     def set_baseline(self, screenshot: Dict[str, Any]) -> Any:
         view = screenshot.get("view")

tests/test_reporter_attachments.py

@@ -0,0 +1,104 @@
+import pytest
+
+from angles_python_client.reporter import AnglesReporter
+
+
+class _RecordingAttachmentRequests:
+    def __init__(self):
+        self.uploads = []
+
+    def _stored(self, build_id, file_name):
+        attachment_id = f"attachment-{len(self.uploads) + 1}"
+        self.uploads.append((build_id, file_name))
+        return {"_id": attachment_id, "originalName": file_name, "kind": "log"}
+
+    def upload_test_attachment(self, build_id, file_path, file_name=None):
+        return self._stored(build_id, file_name or file_path.rsplit("/", 1)[-1])
+
+    def upload_test_attachment_data(self, build_id, data, file_name):
+        return self._stored(build_id, file_name)
+
+
+class _RecordingExecutionRequests:
+    def __init__(self):
+        self.saved = []
+
+    def save_execution(self, execution):
+        self.saved.append(execution)
+        return {"_id": "execution-id"}
+
+
+def _reporter():
+    reporter = AnglesReporter(base_url="https://angles.example/rest/api/v1.0/")
+    reporter.attachments = _RecordingAttachmentRequests()
+    reporter.executions = _RecordingExecutionRequests()
+    reporter.set_current_build("build-id")
+    return reporter
+
+
+def test_attach_file_adds_the_id_to_the_current_test():
+    reporter = _reporter()
+    reporter.start_test("test-one", "suite-one")
+
+    first = reporter.attach_file("/tmp/videos/checkout.webm")
+    second = reporter.attach_data("one\ntwo\n", "console.log")
+    reporter.save_test()
+
+    saved = reporter.executions.saved[0]
+    assert saved.attachments == [first["_id"], second["_id"]]
+    assert reporter.attachments.uploads == [("build-id", "checkout.webm"), ("build-id", "console.log")]
+
+
+def test_attach_to_last_step_adds_the_id_to_that_step_only():
+    reporter = _reporter()
+    reporter.start_test("test-one", "suite-one")
+    reporter.add_action("Pay")
+    reporter.pass_step("Open page", "200", "200", "")
+    reporter.fail_step("Confirmation", "Order confirmed", "Declined", "")
+
+    html = reporter.attach_data_to_last_step("<html></html>", "page.html")
+    png = reporter.attach_file_to_last_step("/tmp/failure.png")
+
+    steps = reporter.current_action.steps
+    assert steps[0].attachments is None
+    assert steps[1].attachments == [html["_id"], png["_id"]]
+    assert reporter.current_execution.attachments is None
+
+
+def test_attachments_are_serialised_with_the_execution():
+    from angles_python_client._serialize import jsonable
+
+    reporter = _reporter()
+    reporter.start_test("test-one", "suite-one")
+    reporter.info("collecting")
+    reporter.attach_data_to_last_step("x", "page.html")
+    reporter.attach_file("/tmp/trace.zip")
+
+    body = jsonable(reporter.current_execution)
+    assert body["attachments"] == ["attachment-2"]
+    assert body["actions"][0]["steps"][0]["attachments"] == ["attachment-1"]
+
+
+def test_attachments_survive_batch_mode():
+    reporter = _reporter()
+    reporter.set_batch_mode(True)
+    reporter.start_test("test-one", "suite-one")
+    reporter.attach_file("/tmp/network.har")
+    reporter.save_test()
+
+    assert reporter._batched_executions[0].attachments == ["attachment-1"]
+
+
+def test_attaching_needs_a_started_test():
+    reporter = _reporter()
+    with pytest.raises(RuntimeError, match="start_test"):
+        reporter.attach_file("/tmp/console.log")
+    assert reporter.attachments.uploads == []
+
+
+def test_attaching_to_a_step_needs_a_step():
+    reporter = _reporter()
+    reporter.start_test("test-one", "suite-one")
+    with pytest.raises(RuntimeError, match="add a step"):
+        reporter.attach_file_to_last_step("/tmp/console.log")
+    assert reporter.attachments.uploads == []