Issue #5 initial pytest plugin impl to upload results on failure to s3

.github/workflows/benchmarks.yaml

@@ -31,6 +31,10 @@ jobs:
         env:
           OPENEO_AUTH_METHOD: client_credentials
           OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED: ${{ secrets.OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED }}
+          UPLOAD_ASSETS_ENDPOINT_URL: "https://s3.waw3-1.cloudferro.com"
+          UPLOAD_ASSETS_BUCKET: "APEx-benchmarks"
+          UPLOAD_ASSETS_ACCESS_KEY_ID: ${{ secrets.UPLOAD_ASSETS_ACCESS_KEY_ID }}
+          UPLOAD_ASSETS_SECRET_ACCESS_KEY: ${{ secrets.UPLOAD_ASSETS_SECRET_ACCESS_KEY }}
       - name: List local reports
         if: always()
         run: ls -alR qa/benchmarks/report

qa/benchmarks/tests/conftest.py

@@ -12,6 +12,7 @@
 
 pytest_plugins = [
     "apex_algorithm_qa_tools.pytest_track_metrics",
+    "apex_algorithm_qa_tools.pytest_upload_assets",
 ]
 
 

qa/benchmarks/tests/test_dummy.py

@@ -36,3 +36,11 @@ def test_produce_files_fail(tmp_path):
     path = tmp_path / "hello.txt"
     path.write_text("Hello, world.\n")
     assert 1 == 2
+
+
+@pytest.mark.parametrize("x", [3, 5])
+def test_upload_assets(tmp_path, upload_assets, x):
+    path = tmp_path / "hello.txt"
+    path.write_text("Hello, world.\n")
+    upload_assets(path)
+    assert x == 5

qa/tools/apex_algorithm_qa_tools/pytest_upload_assets.py

@@ -0,0 +1,118 @@
+"""
+Pytest plugin to collect files generated during benchmark/test
+and upload them to S3 (e.g. on test failure).
+"""
+
+import logging
+import os
+import re
+import time
+from pathlib import Path
+from typing import Callable, Dict, Union
+
+import boto3
+import pytest
+
+_log = logging.getLogger(__name__)
+
+_PLUGIN_NAME = "upload_assets"
+
+
+def pytest_addoption(parser):
+    # TODO: option to inject github run id
+    # TODO: options for S3 bucket, credentials, ...
+    # TODO: option to always upload (also on success).
+    ...
+
+
+def pytest_configure(config: pytest.Config):
+    if (
+        # TODO only register if enough config is available for setup
+        # Don't register on xdist worker nodes
+        not hasattr(config, "workerinput")
+    ):
+        s3_client = boto3.client(
+            service_name="s3",
+            aws_access_key_id=os.environ.get("UPLOAD_ASSETS_ACCESS_KEY_ID"),
+            aws_secret_access_key=os.environ.get("UPLOAD_ASSETS_SECRET_ACCESS_KEY"),
+            # TODO Option for endpoint url
+            endpoint_url=os.environ.get("UPLOAD_ASSETS_ENDPOINT_URL"),
+        )
+        bucket = os.environ.get("UPLOAD_ASSETS_BUCKET")
+        # TODO: do run id through option
+        if os.environ.get("GITHUB_RUN_ID"):
+            run_id = "github-" + os.environ["GITHUB_RUN_ID"]
+        else:
+            run_id = f"local-{int(time.time())}"
+
+        config.pluginmanager.register(
+            S3UploadPlugin(
+                run_id=run_id,
+                s3_client=s3_client,
+                bucket=bucket,
+            ),
+            name=_PLUGIN_NAME,
+        )
+
+
+class _Collector:
+    """
+    Collects test outcomes and files to upload for a single test node.
+    """
+
+    def __init__(self, nodeid: str) -> None:
+        self.nodeid = nodeid
+        self.outcomes: Dict[str, str] = {}
+        self.assets: Dict[str, Path] = {}
+
+    def set_outcome(self, when: str, outcome: str):
+        self.outcomes[when] = outcome
+
+    def collect(self, path: Path, name: str):
+        self.assets[name] = path
+
+
+class S3UploadPlugin:
+    def __init__(self, run_id: str, s3_client, bucket: str) -> None:
+        # TODO: bucket, credentials, githubrunid, ...
+        self.run_id = run_id
+        self.collector: Union[_Collector, None] = None
+        self.s3_client = s3_client
+        self.bucket = bucket
+
+    def pytest_runtest_logstart(self, nodeid, location):
+        self.collector = _Collector(nodeid=nodeid)
+
+    def pytest_runtest_logreport(self, report: pytest.TestReport):
+        self.collector.set_outcome(when=report.when, outcome=report.outcome)
+
+    def pytest_runtest_logfinish(self, nodeid, location):
+        # TODO: option to also upload on success?
+        if self.collector.outcomes.get("call") == "failed":
+            self._upload(self.collector)
+
+        self.collector = None
+
+    def _upload(self, collector: _Collector):
+        for name, path in collector.assets.items():
+            nodeid = re.sub(r"[^a-zA-Z0-9_.-]", "_", collector.nodeid)
+            key = f"{self.run_id}!{nodeid}!{name}"
+            # TODO: get upload info in report?
+            _log.info(f"Uploading {path} to {self.bucket}/{key}")
+            self.s3_client.upload_file(Filename=str(path), Bucket=self.bucket, Key=key)
+
+
+@pytest.fixture
+def upload_assets(pytestconfig, tmp_path) -> Callable[[Path], None]:
+    """
+    Fixture to register a file (under `tmp_path`) for S3 upload
+    after the test failed.
+    """
+    uploader = pytestconfig.pluginmanager.get_plugin(_PLUGIN_NAME)
+
+    def collect(path: Path):
+        assert path.is_relative_to(tmp_path)
+        name = str(path.relative_to(tmp_path))
+        uploader.collector.collect(path=path, name=name)
+
+    return collect

qa/tools/pyproject.toml

@@ -21,4 +21,6 @@ dependencies = [
     "requests>=2.32.0",
     "jsonschema>=4.0.0",
     "openeo>=0.30.0",
+    # TODO: make some of these dependencies optional
+    "boto3>=1.34.0",
 ]