bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

keycloak: allow e2e client execute-actions-email

Browse Source
This commit is contained in:
Brad Stein 2026-01-04 00:58:02 -03:00
parent cadb0daba0
commit eb11eaff4e
3 changed files with 65 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
scripts/tests/test_keycloak_execute_actions_email.py
View File

@ -2,6 +2,7 @@
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
@ -46,7 +47,10 @@ def _request_json(method: str, url: str, token: str, payload: object | None = No
return resp.status, json.loads(body.decode())
except urllib.error.HTTPError as exc:
raw = exc.read().decode(errors="replace")
raise SystemExit(f"HTTP {exc.code} from {url}: {raw}")
try:
return exc.code, json.loads(raw) if raw else None
except json.JSONDecodeError:
return exc.code, raw
def main() -> int:
@ -64,23 +68,34 @@ def main() -> int:
token_url = f"{keycloak_base}/realms/{realm}/protocol/openid-connect/token"
admin_users_url = f"{keycloak_base}/admin/realms/{realm}/users"
token_payload = _post_form(
token_url,
{"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret},
timeout_s=30,
)
access_token = token_payload.get("access_token")
if not isinstance(access_token, str) or not access_token:
raise SystemExit("client credentials token missing access_token")
def get_access_token() -> str:
token_payload = _post_form(
token_url,
{"grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret},
timeout_s=30,
)
access_token = token_payload.get("access_token")
if not isinstance(access_token, str) or not access_token:
raise SystemExit("client credentials token missing access_token")
return access_token
status, users = _request_json(
"GET",
f"{admin_users_url}?{urllib.parse.urlencode({'username': probe_username, 'exact': 'true'})}",
access_token,
timeout_s=30,
)
if status != 200 or not isinstance(users, list):
raise SystemExit("unexpected admin API response when searching for probe user")
access_token = get_access_token()
users: list | None = None
search_url = f"{admin_users_url}?{urllib.parse.urlencode({'username': probe_username, 'exact': 'true'})}"
for attempt in range(1, 11):
status, body = _request_json("GET", search_url, access_token, timeout_s=30)
if status == 200 and isinstance(body, list):
users = body
break
if status == 403 and attempt < 10:
time.sleep(3)
access_token = get_access_token()
continue
raise SystemExit(f"unexpected admin API response when searching for probe user (status={status} body={body})")
if users is None:
raise SystemExit("probe user search did not return a list response")
if not users:
create_payload = {
@ -89,17 +104,27 @@ def main() -> int:
"email": probe_email,
"emailVerified": True,
}
status, _ = _request_json("POST", admin_users_url, access_token, create_payload, timeout_s=30)
if status not in (201, 204):
raise SystemExit(f"unexpected status creating probe user: {status}")
status, users = _request_json(
"GET",
f"{admin_users_url}?{urllib.parse.urlencode({'username': probe_username, 'exact': 'true'})}",
access_token,
timeout_s=30,
)
if status != 200 or not isinstance(users, list) or not users:
raise SystemExit("failed to refetch probe user after creation")
for attempt in range(1, 6):
status, body = _request_json("POST", admin_users_url, access_token, create_payload, timeout_s=30)
if status in (201, 204):
break
if status == 403 and attempt < 5:
time.sleep(3)
access_token = get_access_token()
continue
raise SystemExit(f"unexpected status creating probe user: {status} body={body}")
# Refetch.
for attempt in range(1, 11):
status, body = _request_json("GET", search_url, access_token, timeout_s=30)
if status == 200 and isinstance(body, list) and body:
users = body
break
if status == 403 and attempt < 10:
time.sleep(3)
access_token = get_access_token()
continue
raise SystemExit(f"failed to refetch probe user after creation (status={status} body={body})")
user_id = users[0].get("id")
if not isinstance(user_id, str) or not user_id:
@ -114,9 +139,15 @@ def main() -> int:
}
)
url = f"{admin_users_url}/{urllib.parse.quote(user_id)}/execute-actions-email?{query}"
status, _ = _request_json("PUT", url, access_token, ["UPDATE_PASSWORD"], timeout_s=30)
if status != 204:
raise SystemExit(f"unexpected status from execute-actions-email: {status}")
for attempt in range(1, 6):
status, body = _request_json("PUT", url, access_token, ["UPDATE_PASSWORD"], timeout_s=30)
if status == 204:
break
if status == 403 and attempt < 5:
time.sleep(3)
access_token = get_access_token()
continue
raise SystemExit(f"unexpected status from execute-actions-email: {status} body={body}")
print("PASS: Keycloak execute-actions-email succeeded")
return 0
@ -124,4 +155,3 @@ def main() -> int:
if __name__ == "__main__":
sys.exit(main())

4
services/keycloak/portal-e2e-client-job.yaml
View File

@ -2,7 +2,7 @@
apiVersion: batch/v1
kind: Job
metadata:
name: keycloak-portal-e2e-client-1
name: keycloak-portal-e2e-client-2
namespace: sso
spec:
backoffLimit: 0
@ -211,7 +211,7 @@ spec:
if not rm_uuid:
raise SystemExit("realm-management client has no id")
wanted_roles = ("query-users", "view-users", "impersonation")
wanted_roles = ("query-users", "view-users", "manage-users", "impersonation")
role_reps = []
for role_name in wanted_roles:
status, role = http_json(

3
services/keycloak/portal-e2e-execute-actions-email-test-job.yaml
View File

@ -2,7 +2,7 @@
apiVersion: batch/v1
kind: Job
metadata:
name: keycloak-portal-e2e-execute-actions-email-1
name: keycloak-portal-e2e-execute-actions-email-2
namespace: sso
spec:
backoffLimit: 3
@ -49,4 +49,3 @@ spec:
configMap:
name: portal-e2e-tests
defaultMode: 0555