Throw RuntimeError with message on job failure

Matatika · Mar 22, 2022 · ba7a49e · ba7a49e
1 parent e860c44
commit ba7a49e
Show file tree

Hide file tree

Showing 3 changed files with 68 additions and 0 deletions.
diff --git a/tap_auth0/streams.py b/tap_auth0/streams.py
@@ -101,6 +101,13 @@ def _poll_job(self, get_job_request: requests.PreparedRequest, count=1) -> Any:
         if status == "completed":
             return job
 
+        if status == "failed":
+            id_ = job["id"]
+            summary: dict[str, int] = job["summary"]
+            summary_format = ", ".join(f"{k}: {v}" for k, v in summary.items())
+
+            raise RuntimeError(f"Job '{id_}' failed ({summary_format})")
+
         time.sleep(job_poll_interval_ms / 1000)
         return self._poll_job(get_job_request, count=count + 1)
 

diff --git a/tap_auth0/tests/test_sync.py b/tap_auth0/tests/test_sync.py
@@ -90,6 +90,54 @@ def test_auth0_sync_users(self):
         self.assertIsInstance(test_utils.SINGER_MESSAGES[1], singer.RecordMessage)
         self.assertIsInstance(test_utils.SINGER_MESSAGES[2], singer.StateMessage)
 
+    @responses.activate
+    def test_auth0_sync_users_failed(self):
+        """Test sync users with failed job"""
+
+        tap = test_utils.set_up_tap_with_custom_catalog(
+            self.mock_config, ["stream_auth0_users"]
+        )
+
+        responses.add(
+            responses.POST,
+            "https://test.auth0.com/oauth/token",
+            json={"access_token": "12345", "expires_in": 3622},
+            status=200,
+        )
+
+        job_id = "12345"
+        job = test_utils.users_export_job_pending(job_id)
+        responses.add(
+            responses.POST,
+            "https://test.auth0.com/api/v2/jobs/users-exports",
+            status=200,
+            json=job,
+        )
+
+        job = test_utils.users_export_job_processing(job_id)
+        responses.add(
+            responses.POST,
+            "https://test.auth0.com/api/v2/jobs/users-exports",
+            status=200,
+            json=job,
+        )
+
+        job = test_utils.users_export_job_failed(job_id)
+        responses.add(
+            responses.GET,
+            f"https://test.auth0.com/api/v2/jobs/{job_id}",
+            status=200,
+            json=job,
+        )
+
+        with self.assertRaises(RuntimeError) as err:
+            tap.sync_all()
+
+        self.assertIn(f"Job '{job_id}' failed", str(err.exception))
+
+        self.assertEqual(len(test_utils.SINGER_MESSAGES), 1)
+        self.assertIsInstance(test_utils.SINGER_MESSAGES[0], singer.SchemaMessage)
+
     @responses.activate
     def test_auth0_sync_clients(self):
         """Test sync clients."""

diff --git a/tap_auth0/tests/utils.py b/tap_auth0/tests/utils.py
@@ -31,6 +31,19 @@ def users_export_job_completed(job_id: str):
     }
 
 
+def users_export_job_failed(job_id: str):
+    return {
+        "status": "failed",
+        "id": job_id,
+        "summary": {
+            "failed": len(users_data),
+            "updated": 0,
+            "inserted": 0,
+            "total": len(users_data),
+        },
+    }
+
+
 clients_data = {"start": 0, "total": 100, "clients": [{"client_id": "client_id_12345"}]}
 logs_data = [{"log_id": "log_id_12345"}]