chore(cassandra): audit sso user migration

services/cassandra-auth/bootstrap-jobs/cassandra-ldap-federation-job.yaml

@@ -1,11 +1,11 @@
 # services/cassandra-auth/bootstrap-jobs/cassandra-ldap-federation-job.yaml
-# One-off job for sso/cassandra-ldap-federation-1.
+# One-off job for sso/cassandra-ldap-federation-2.
 # Purpose: attach the Cassandra Keycloak realm to the shared OpenLDAP directory.
 # Run unsuspended for migration, then suspend/replace with a new job name for future runs.
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: cassandra-ldap-federation-1
+  name: cassandra-ldap-federation-2
   namespace: sso
 spec:
   ttlSecondsAfterFinished: 3600
@@ -366,5 +366,32 @@ spec:
                   )
                   print(f"Cassandra LDAP user check {username}: {'found' if found else 'missing'}")
 
+              for audit_realm in ("veles", realm):
+                  status, count, _ = request(
+                      "GET",
+                      f"{base_url}/admin/realms/{audit_realm}/users/count",
+                      token,
+                  )
+                  if status == 200:
+                      print(f"SSO audit realm {audit_realm}: user_count={count}")
+                  for term in (item.strip() for item in os.environ.get("VERIFY_USERS", "").split(",")):
+                      if not term:
+                          continue
+                      encoded = urllib.parse.quote(term)
+                      status, users, _ = request(
+                          "GET",
+                          f"{base_url}/admin/realms/{audit_realm}/users?search={encoded}",
+                          token,
+                      )
+                      users = users if status == 200 and isinstance(users, list) else []
+                      matches = []
+                      for user in users:
+                          username = str(user.get("username") or "")
+                          email = str(user.get("email") or "")
+                          if term.casefold() in username.casefold() or term.casefold() in email.casefold():
+                              matches.append(username if not email else f"{username}<{email}>")
+                      joined = ", ".join(matches[:8]) if matches else "none"
+                      print(f"SSO audit realm {audit_realm} search {term}: {joined}")
+
               print("Cassandra LDAP federation ready")
               PY