Make identity deletion replay safe

identity-provisioner/provisioner.py

@@ -153,10 +153,15 @@ mutation Remove($userId: String!, $groupId: Int!) {
         return list(value["users"]), list(value["groups"])
 
     def _user(self, token: str, subject: str) -> dict[str, Any] | None:
-        value = self._gql(token, """
+        try:
+            value = self._gql(token, """
 query User($id: String!) {
   user(userId: $id) { id email displayName groups { id displayName } }
 }""", {"id": subject})
+        except ValueError as exc:
+            if "not found" in str(exc).lower():
+                return None
+            raise
         user = value.get("user")
         return dict(user) if user else None
 

identity-provisioner/tests/test_provisioner.py

@@ -3,7 +3,7 @@ import sys
 import unittest
 
 sys.path.insert(0, str(pathlib.Path(__file__).parents[1]))
-from provisioner import _username, dispatch, DriftResult, Result
+from provisioner import _username, dispatch, DriftResult, LLDAPProvisioner, Result
 
 
 class Fake:
@@ -45,3 +45,10 @@ class ProvisionerTests(unittest.TestCase):
         reconciled = dispatch(Fake(), "/v1/identities/reconcile", payload)
         self.assertEqual("drifted", drift.status)
         self.assertEqual("reconciled", reconciled.status)
+
+    def test_missing_directory_user_is_normalized(self):
+        provisioner = LLDAPProvisioner(base_url="http://directory", admin_password="unused")
+        provisioner._gql = lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            ValueError("User not found")
+        )
+        self.assertIsNone(provisioner._user("token", "absent"))