Changed: conftest file

konfuzio_sdk/api.py

@@ -170,7 +170,7 @@ def konfuzio_session(
     retry_strategy = Retry(
         total=num_retries,
         status_forcelist=[429, 500, 502, 503, 504],
-        backoff_factor=3,
+        backoff_factor=2,
     )
     session = requests.Session()
     session.mount('https://', adapter=TimeoutHTTPAdapter(max_retries=retry_strategy, timeout=timeout))

tests/conftest.py

@@ -0,0 +1,37 @@
+import pytest
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    report = outcome.get_result()
+
+    if report.when == 'call' and report.failed and call.excinfo is not None:
+        if call.excinfo is not None:
+            exc_value = call.excinfo.value
+            error_message = str(exc_value)
+            if '502' in error_message or 'Read timed out' in error_message:
+                setattr(report, 'wasxfail', False)
+                setattr(report, 'rerun', True)
+                report.outcome = 'failed'
+
+
+def pytest_addoption(parser):
+    parser.addoption('--n_reruns', action='store', default=5, help='Number of times to rerun failed tests')
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_protocol(item, nextitem):
+    try:
+        reruns = int(item.config.getoption('--n_reruns'))
+    except TypeError:
+        reruns = 0
+    for i in range(reruns + 1):
+        if i > 0:
+            item.ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
+        outcome = yield
+        report = outcome.get_result()
+        if not getattr(report, 'rerun', False):
+            break
+        if i == reruns:
+            report.outcome = 'failed'

tests/test_api.py

@@ -686,19 +686,6 @@ def test_get_all_project_ais_with_invalid_permissions(
         self.assertEqual(result['categorization']['error'].__str__(), exception_message)
         self.assertEqual(result['filesplitting']['error'].__str__(), exception_message)
 
-    def test_restore_snapshot(self):
-        """Test restoring a snapshot using snapshotrestores endpoint."""
-        project_id = restore_snapshot(snapshot_id=65)
-        all_projects = get_project_list()
-        assert project_id in [project['id'] for project in all_projects['results']]
-        new_project = Project(id_=project_id)
-        for document in new_project.documents + new_project.test_documents:
-            document.dataset_status = 0
-            document.save_meta_data()
-            document.delete(delete_online=True)
-        r = delete_project(project_id=project_id)
-        assert r.status_code == 204
-
     @classmethod
     def tearDownClass(cls) -> None:
         """Remove the project created specifically for this test pipeline."""