Run artifacts support

README.md

@@ -478,6 +478,7 @@ This functions gives you more flexibility in reporting and make your reports mor
 
 **Available functions**
 - [add_artifact](#add-artifact)
+- [add_run_artifact](#add-run-artifact)
 - [add_meta](#add-meta)
 - [add_label](#add-label)
 - [link_jira](#link-jira)
@@ -496,6 +497,22 @@ def test_my_test():
     add_artifact(path_to_file)
     assert True
 ```
+#### Add Run Artifact
+Attaches a file to the test run report (not to an individual test). The file will be uploaded to S3 at the end of the run.
+Use this for run-level artifacts such as coverage reports, allure archives, or other files not tied to a specific test.
+
+**Note:** S3 must be configured
+
+```python
+from pytestomatio.functions import add_run_artifact
+
+def test_my_test():
+    add_run_artifact('path/to/coverage-report.html')
+    assert True
+```
+
+Works with pytest-xdist: artifacts collected from all workers are merged and uploaded once by the main process.
+
 #### Add Meta
 Adds meta information to test. Meta information is a key:value pair(s), which is used to add additional information to the test report. E.g. browser, environment, etc.
 

pyproject.toml

@@ -13,7 +13,7 @@ version_provider = "pep621"
 update_changelog_on_bump = false
 [project]
 name = "pytestomatio"
-version = "2.13.0"
+version = "2.14.0b0"
 
 dependencies = [
     "requests>=2.32.4",

pytestomatio/connect/connector.py

@@ -354,6 +354,19 @@ def batch_tests_upload(self, run_id: str,
                 self._is_report_failed(response.status_code)
 
     # TODO: I guess this class should be just an API client and used within testRun (testRunConfig)
+    def upload_run_artifacts(self, run_id: str, artifacts: list) -> None:
+        log.info(f'Uploading run artifacts. Run id: {run_id}')
+        url = f'{self.base_url}/api/reporter/{run_id}?api_key={self.api_key}'
+        try:
+            response = self._send_request_with_retry('put', url, json={"artifacts": artifacts})
+            if response.status_code == 200:
+                log.info(f'Run artifacts uploaded')
+            else:
+                self._show_status_message(response.status_code)
+                log.error(f'Failed to upload run artifacts. Status code: {response.status_code}')
+        except Exception as e:
+            log.error(f'Failed to upload run artifacts: {e}')
+
     def finish_test_run(self, run_id: str, is_final=False) -> None:
         log.info(f'Finishing test run. Run id: {run_id}')
         status_event = 'finish_parallel' if is_final else 'finish'

pytestomatio/functions.py

@@ -1,4 +1,5 @@
 from pytestomatio.services.artifact_storage import artifact_storage
+from pytestomatio.services.run_artifact_storage import run_artifact_storage
 from pytestomatio.services.meta_storage import meta_storage
 from pytestomatio.services.link_storage import link_storage
 from pytestomatio.utils.helper import get_current_test_id
@@ -61,3 +62,11 @@ def add_artifact(path: str):
     if not test_id:
         return
     artifact_storage.put(test_id, path)
+
+
+def add_run_artifact(path: str):
+    """
+    Add an artifact to the current test run
+    :param path: path to artifact
+    """
+    run_artifact_storage.put(path)

pytestomatio/main.py

@@ -20,6 +20,7 @@
 from pytestomatio.testomatio.filter_plugin import TestomatioFilterPlugin
 
 from pytestomatio.services.artifact_storage import artifact_storage
+from pytestomatio.services.run_artifact_storage import run_artifact_storage
 from pytestomatio.services.meta_storage import meta_storage
 from pytestomatio.services.link_storage import link_storage
 
@@ -344,6 +345,9 @@ def pytest_testnodedown(node, error):
     if not hasattr(node, 'workeroutput') or not hasattr(pytest, 'testomatio') or \
             node.config.getoption(testomatio) is None or node.config.getoption(testomatio) != 'report':
         return
+    for path in node.workeroutput.get('run_artifacts', []):
+        run_artifact_storage.put(path)
+
     if pytest.testomatio.test_run_config.disable_batch:
         return
 
@@ -361,6 +365,11 @@ def pytest_sessionfinish(session, exitstatus):
         return
 
     run: TestRunConfig = pytest.testomatio.test_run_config
+
+    # xdist worker process - pass run artifacts to main process regardless of batch mode
+    if hasattr(session.config, 'workerinput'):
+        session.config.workeroutput['run_artifacts'] = run_artifact_storage.get()
+
     if not run.disable_batch:
 
         # xdist worker process - write test results to worker output. They will be reported from master process
@@ -383,7 +392,23 @@ def pytest_unconfigure(config: Config):
     if config.getoption(testomatio) != 'report':
         run.clear_run_id()
         return
-    elif run.proceed:
+
+    # upload run artifacts in main process
+    if not hasattr(config, 'workerinput') and not run.disable_artifacts:
+        attached = run_artifact_storage.get()
+        if attached:
+            if not pytest.testomatio.s3_connector:
+                run_details = pytest.testomatio.connector.update_test_run(**run.to_dict())
+                s3_details = read_env_s3_keys(run_details) if run_details else None
+                if s3_details and all(s3_details):
+                    pytest.testomatio.s3_connector = S3Connector(*s3_details)
+                    pytest.testomatio.s3_connector.login()
+            if pytest.testomatio.s3_connector:
+                urls = pytest.testomatio.s3_connector.upload_files([(path, None) for path in attached])
+                run_artifact_storage.clear()
+                pytest.testomatio.connector.upload_run_artifacts(run.test_run_id, urls)
+
+    if run.proceed:
         if not hasattr(config, 'workerinput'):  # for xdist, only master clear run id
             run.clear_run_id()
         return

pytestomatio/services/run_artifact_storage.py

@@ -0,0 +1,22 @@
+class RunArtifactStorage:
+
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._paths = []
+        return cls._instance
+
+    def put(self, path: str):
+        if path not in self._paths:
+            self._paths.append(path)
+
+    def get(self) -> list:
+        return self._paths
+
+    def clear(self):
+        self._paths.clear()
+
+
+run_artifact_storage = RunArtifactStorage()

tests/test_connect/test_connector.py

@@ -419,6 +419,40 @@ def test_update_test_run_success(self, mock_put, connector):
 
         assert result == {"uid": "run_123"}
 
+    @patch('requests.Session.put')
+    def test_upload_run_artifacts_success(self, mock_put, connector):
+        """Test successful run artifacts upload"""
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_put.return_value = mock_response
+        artifacts = ['https://s3.example.com/run/screenshot.png', 'https://s3.example.com/run/log.txt']
+
+        connector.upload_run_artifacts('run_123', artifacts)
+
+        mock_put.assert_called_once_with(
+            f'{connector.base_url}/api/reporter/run_123?api_key={connector.api_key}',
+            json={"artifacts": artifacts}
+        )
+
+    @patch('requests.Session.put')
+    def test_upload_run_artifacts_http_error(self, mock_put, connector):
+        """Test upload_run_artifacts logs error on non-200 response"""
+        mock_response = Mock()
+        mock_response.status_code = 500
+        mock_response.json.return_value = {}
+        mock_put.return_value = mock_response
+
+        connector.upload_run_artifacts('run_123', ['https://s3.example.com/artifact.png'])
+
+        mock_put.assert_called_once()
+
+    @patch('requests.Session.put')
+    def test_upload_run_artifacts_exception(self, mock_put, connector):
+        """Test upload_run_artifacts handles exception without raising"""
+        mock_put.side_effect = Exception("Connection failed")
+
+        connector.upload_run_artifacts('run_123', ['https://s3.example.com/artifact.png'])
+
     @patch('requests.Session.post')
     def test_update_test_status_success(self, mock_post, connector):
         """Test successful test status update"""

tests/test_main.py

@@ -1248,6 +1248,101 @@ def test_unconfigure_main_process_for_proceed_run(self):
         pytest.testomatio.connector.assert_not_called()
         assert pytest.testomatio.test_run_config.clear_run_id.call_count == 1
 
+    @patch('pytestomatio.main.run_artifact_storage')
+    @patch('pytestomatio.main.time.sleep')
+    def test_unconfigure_uploads_run_artifacts(self, mock_sleep, mock_storage):
+        """Test run artifacts are uploaded to S3 and sent to connector in main process"""
+        mock_config = Mock(spec=['addinivalue_line', 'getini', 'getoption', 'pluginmanager'])
+        mock_config.getoption.return_value = 'report'
+        assert not hasattr(mock_config, 'workerinput')
+
+        pytest.testomatio = Mock()
+        pytest.testomatio.test_run_config.test_run_id = 'run_123'
+        pytest.testomatio.test_run_config.proceed = None
+        pytest.testomatio.test_run_config.disable_artifacts = False
+        pytest.testomatio.s3_connector.upload_files.return_value = ['https://s3.example.com/artifact.png']
+        mock_storage.get.return_value = ['/local/artifact.png']
+
+        main.pytest_unconfigure(mock_config)
+
+        pytest.testomatio.s3_connector.upload_files.assert_called_once_with([('/local/artifact.png', None)])
+        pytest.testomatio.connector.upload_run_artifacts.assert_called_once_with(
+            'run_123', ['https://s3.example.com/artifact.png']
+        )
+        mock_storage.clear.assert_called_once()
+
+    @patch('pytestomatio.main.run_artifact_storage')
+    @patch('pytestomatio.main.time.sleep')
+    def test_unconfigure_skips_run_artifacts_when_disabled(self, mock_sleep, mock_storage):
+        """Test run artifacts are not uploaded when disable_artifacts is set"""
+        mock_config = Mock(spec=['addinivalue_line', 'getini', 'getoption', 'pluginmanager'])
+        mock_config.getoption.return_value = 'report'
+
+        pytest.testomatio = Mock()
+        pytest.testomatio.test_run_config.test_run_id = 'run_123'
+        pytest.testomatio.test_run_config.proceed = None
+        pytest.testomatio.test_run_config.disable_artifacts = True
+
+        main.pytest_unconfigure(mock_config)
+
+        mock_storage.get.assert_not_called()
+        pytest.testomatio.connector.upload_run_artifacts.assert_not_called()
+
+    @patch('pytestomatio.main.run_artifact_storage')
+    @patch('pytestomatio.main.time.sleep')
+    def test_unconfigure_skips_run_artifacts_when_no_s3(self, mock_sleep, mock_storage):
+        """Test run artifacts are not uploaded when s3_connector is not configured and fallback also fails"""
+        mock_config = Mock(spec=['addinivalue_line', 'getini', 'getoption', 'pluginmanager'])
+        mock_config.getoption.return_value = 'report'
+
+        pytest.testomatio = Mock()
+        pytest.testomatio.test_run_config.test_run_id = 'run_123'
+        pytest.testomatio.test_run_config.proceed = None
+        pytest.testomatio.test_run_config.disable_artifacts = False
+        pytest.testomatio.test_run_config.to_dict.return_value = {'id': 'run_123'}
+        pytest.testomatio.s3_connector = None
+        pytest.testomatio.connector.update_test_run.return_value = None
+        mock_storage.get.return_value = ['/local/artifact.png']
+
+        main.pytest_unconfigure(mock_config)
+
+        pytest.testomatio.connector.upload_run_artifacts.assert_not_called()
+
+    @patch('pytestomatio.main.run_artifact_storage')
+    def test_unconfigure_run_artifacts_not_uploaded_from_worker(self, mock_storage):
+        """Test run artifacts are not uploaded from xdist worker process"""
+        mock_config = Mock()
+        mock_config.workerinput = Mock()
+        mock_config.getoption.return_value = 'report'
+
+        pytest.testomatio = Mock()
+        pytest.testomatio.test_run_config.proceed = None
+
+        main.pytest_unconfigure(mock_config)
+
+        mock_storage.get.assert_not_called()
+        pytest.testomatio.connector.upload_run_artifacts.assert_not_called()
+
+    @patch('pytestomatio.main.run_artifact_storage')
+    def test_unconfigure_uploads_run_artifacts_for_proceed_run(self, mock_storage):
+        """Test run artifacts are uploaded even when proceed is set"""
+        mock_config = Mock(spec=['addinivalue_line', 'getini', 'getoption', 'pluginmanager'])
+        mock_config.getoption.return_value = 'report'
+        assert not hasattr(mock_config, 'workerinput')
+
+        pytest.testomatio = Mock()
+        pytest.testomatio.test_run_config.test_run_id = 'run_123'
+        pytest.testomatio.test_run_config.proceed = True
+        pytest.testomatio.test_run_config.disable_artifacts = False
+        pytest.testomatio.s3_connector.upload_files.return_value = ['https://s3.example.com/artifact.png']
+        mock_storage.get.return_value = ['/local/artifact.png']
+
+        main.pytest_unconfigure(mock_config)
+
+        pytest.testomatio.connector.upload_run_artifacts.assert_called_once_with(
+            'run_123', ['https://s3.example.com/artifact.png']
+        )
+
 @pytest.mark.smoke
 class TestPytestRuntestLogfinish:
     """Tests for pytest_runtest_logfinish hook"""

tests/test_services/test_run_artifact_storage.py

@@ -0,0 +1,74 @@
+import pytest
+from pytestomatio.services.run_artifact_storage import RunArtifactStorage, run_artifact_storage
+
+
+class TestRunArtifactStorage:
+
+    @pytest.fixture
+    def storage(self):
+        instance = RunArtifactStorage()
+        instance._paths = []
+        return instance
+
+    def test_put_stores_path(self, storage):
+        storage.put('/path/to/artifact.png')
+
+        assert storage._paths == ['/path/to/artifact.png']
+
+    def test_put_multiple_paths_accumulates(self, storage):
+        storage.put('/path/one.png')
+        storage.put('/path/two.png')
+
+        assert storage._paths == ['/path/one.png', '/path/two.png']
+
+    def test_put_deduplicates_same_path(self, storage):
+        storage.put('/path/artifact.png')
+        storage.put('/path/artifact.png')
+
+        assert storage._paths == ['/path/artifact.png']
+
+    def test_get_returns_stored_paths(self, storage):
+        storage.put('/path/artifact.png')
+
+        assert storage.get() == ['/path/artifact.png']
+
+    def test_get_empty_storage(self, storage):
+        assert storage.get() == []
+
+    def test_clear_empties_storage(self, storage):
+        storage.put('/path/artifact.png')
+
+        storage.clear()
+
+        assert storage.get() == []
+
+    def test_clear_on_empty_storage_does_not_raise(self, storage):
+        storage.clear()
+
+    def test_get_returns_internal_list(self, storage):
+        storage.put('/path/artifact.png')
+
+        assert storage.get() is storage._paths
+
+
+class TestRunArtifactStorageSingleton:
+
+    @pytest.fixture(autouse=True)
+    def cleanup(self):
+        yield
+        run_artifact_storage._paths.clear()
+
+    def test_is_instance_of_RunArtifactStorage(self):
+        assert isinstance(run_artifact_storage, RunArtifactStorage)
+
+    def test_singleton_returns_same_instance(self):
+        from pytestomatio.services.run_artifact_storage import run_artifact_storage as storage2
+
+        assert run_artifact_storage is storage2
+
+    def test_singleton_shares_state(self):
+        from pytestomatio.services.run_artifact_storage import run_artifact_storage as storage2
+
+        run_artifact_storage.put('/path/artifact.png')
+
+        assert storage2.get() == ['/path/artifact.png']