165 lines
8.6 KiB
Python
165 lines
8.6 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Exercise identity SQL against acceptance PostgreSQL, always rolling back."""
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from apply_weknora_identities import sql_run, target_snapshot
|
||
|
|
from weknora_category_etl import literal, save_json
|
||
|
|
from weknora_identity_etl import (IDENTITY_LEDGER, ORGANIZATION_LEDGER, TARGET_TENANT,
|
||
|
|
identity_sql, identity_verification_sql)
|
||
|
|
|
||
|
|
|
||
|
|
def rejected_case(name, mutation, expected_message, import_sql):
|
||
|
|
# Roll back both the simulated conflict and the attempted import, even when
|
||
|
|
# the expected rejection is caught in the inner subtransaction.
|
||
|
|
return f"""
|
||
|
|
DO $test_conflict$
|
||
|
|
DECLARE blocked boolean := false;
|
||
|
|
BEGIN
|
||
|
|
BEGIN
|
||
|
|
{mutation}
|
||
|
|
BEGIN
|
||
|
|
EXECUTE {literal(import_sql)};
|
||
|
|
EXCEPTION WHEN raise_exception THEN
|
||
|
|
IF position({literal(expected_message)} in SQLERRM) > 0 THEN blocked := true; ELSE RAISE; END IF;
|
||
|
|
END;
|
||
|
|
IF NOT blocked THEN RAISE EXCEPTION 'Expected conflict was not rejected'; END IF;
|
||
|
|
RAISE EXCEPTION USING ERRCODE='Z9001', MESSAGE='rollback accepted test';
|
||
|
|
EXCEPTION WHEN SQLSTATE 'Z9001' THEN NULL;
|
||
|
|
END;
|
||
|
|
END $test_conflict$;
|
||
|
|
SELECT json_build_object('check',{literal(name)},'passed',true);
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
def regression_sql(manifest):
|
||
|
|
query = identity_sql(manifest)
|
||
|
|
uid = next(u["target_id"] for u in manifest["users"] if u["action"] == "create_suspended")
|
||
|
|
sql = f"""
|
||
|
|
BEGIN ISOLATION LEVEL SERIALIZABLE;
|
||
|
|
SET LOCAL statement_timeout='30s';
|
||
|
|
SET LOCAL lock_timeout='5s';
|
||
|
|
{query}
|
||
|
|
CREATE TEMP TABLE _test_users ON COMMIT DROP AS SELECT * FROM users;
|
||
|
|
CREATE TEMP TABLE _test_members ON COMMIT DROP AS SELECT * FROM tenant_members;
|
||
|
|
CREATE TEMP TABLE _test_identities ON COMMIT DROP AS SELECT * FROM {IDENTITY_LEDGER};
|
||
|
|
CREATE TEMP TABLE _test_organizations ON COMMIT DROP AS SELECT * FROM {ORGANIZATION_LEDGER};
|
||
|
|
{query}
|
||
|
|
DO $test_repeat$
|
||
|
|
BEGIN
|
||
|
|
IF EXISTS (SELECT * FROM users EXCEPT SELECT * FROM _test_users)
|
||
|
|
OR EXISTS (SELECT * FROM _test_users EXCEPT SELECT * FROM users)
|
||
|
|
OR EXISTS (SELECT * FROM tenant_members EXCEPT SELECT * FROM _test_members)
|
||
|
|
OR EXISTS (SELECT * FROM _test_members EXCEPT SELECT * FROM tenant_members)
|
||
|
|
OR EXISTS (SELECT * FROM {IDENTITY_LEDGER} EXCEPT SELECT * FROM _test_identities)
|
||
|
|
OR EXISTS (SELECT * FROM {ORGANIZATION_LEDGER} EXCEPT SELECT * FROM _test_organizations)
|
||
|
|
THEN RAISE EXCEPTION 'Exact repeat changed stored rows'; END IF;
|
||
|
|
END $test_repeat$;
|
||
|
|
SELECT json_build_object('check','exact_repeat_is_noop','passed',true);
|
||
|
|
"""
|
||
|
|
cases = [
|
||
|
|
("foreign_tenant_rejected", f"UPDATE users SET tenant_id={TARGET_TENANT + 1} WHERE id={literal(uid)};", "ID/provenance collision"),
|
||
|
|
("cross_tenant_membership_rejected", f"INSERT INTO tenant_members (user_id,tenant_id,role,status) VALUES ({literal(uid)},{TARGET_TENANT + 1},'viewer','active');", "unexpected cross-tenant membership"),
|
||
|
|
("changed_mapping_rejected", f"UPDATE {IDENTITY_LEDGER} SET target_user_id='wrong-mapping' WHERE target_user_id={literal(uid)};", "Durable identity mapping conflict"),
|
||
|
|
("deleted_account_not_revived", f"UPDATE users SET deleted_at=now() WHERE id={literal(uid)};", "ID/provenance collision"),
|
||
|
|
("elevated_role_rejected", f"UPDATE tenant_members SET role='admin' WHERE user_id={literal(uid)} AND tenant_id={TARGET_TENANT};", "membership does not match"),
|
||
|
|
]
|
||
|
|
target = next(u for u in manifest["users"] if u["target_id"] == uid)
|
||
|
|
other_uid = next(u["target_id"] for u in manifest["users"] if u["action"] == "create_suspended" and u["target_id"] != uid)
|
||
|
|
for field in ("username", "email"):
|
||
|
|
mutation = f"DELETE FROM {IDENTITY_LEDGER} WHERE target_user_id={literal(uid)};"
|
||
|
|
mutation += f"UPDATE users SET {field}='etl-conflict-vacated@migration.invalid' WHERE id={literal(uid)};"
|
||
|
|
mutation += f"UPDATE users SET {field}={literal(target['target_' + field].upper())} WHERE id={literal(other_uid)};"
|
||
|
|
cases.append((f"normalized_{field}_collision_rejected", mutation, "username/email collision"))
|
||
|
|
for name, mutation, expected in cases:
|
||
|
|
sql += rejected_case(name, mutation, expected, query)
|
||
|
|
sql += f"""
|
||
|
|
UPDATE users SET preferences='{{}}'::jsonb,username='etl-contact-preservation-check',
|
||
|
|
email='etl-contact-preservation-check@migration.invalid',is_active=true,
|
||
|
|
password_hash='!rollback-only-reset-check' WHERE id={literal(uid)};
|
||
|
|
UPDATE tenant_members SET status='active' WHERE user_id={literal(uid)} AND tenant_id={TARGET_TENANT};
|
||
|
|
CREATE TEMP TABLE _test_activated_users ON COMMIT DROP AS SELECT * FROM users;
|
||
|
|
CREATE TEMP TABLE _test_activated_members ON COMMIT DROP AS SELECT * FROM tenant_members;
|
||
|
|
{query}
|
||
|
|
DO $test_activation$
|
||
|
|
BEGIN
|
||
|
|
IF EXISTS (SELECT * FROM users EXCEPT SELECT * FROM _test_activated_users)
|
||
|
|
OR EXISTS (SELECT * FROM tenant_members EXCEPT SELECT * FROM _test_activated_members)
|
||
|
|
THEN RAISE EXCEPTION 'Repeat overwrote native profile or activation'; END IF;
|
||
|
|
END $test_activation$;
|
||
|
|
SELECT json_build_object('check','native_preferences_contacts_and_activation_preserved','passed',true);
|
||
|
|
SELECT json_build_object('check','durable_mapping_and_account_scope','passed',bool_and(
|
||
|
|
database_matches AND organization_mapping_matches AND identity_mapping_matches
|
||
|
|
AND account_scope_matches AND membership_matches)) FROM ({identity_verification_sql(manifest).rstrip().rstrip(';')}) checks;
|
||
|
|
ROLLBACK;
|
||
|
|
"""
|
||
|
|
return sql
|
||
|
|
|
||
|
|
|
||
|
|
def api_checks(manifest, credentials_path):
|
||
|
|
credentials = json.loads(credentials_path.read_text())
|
||
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
|
|
|
||
|
|
def request(path, token=None, payload=None):
|
||
|
|
headers = {"Content-Type": "application/json"}
|
||
|
|
if token:
|
||
|
|
headers["Authorization"] = "Bearer " + token
|
||
|
|
req = urllib.request.Request("http://127.0.0.1:18080" + path, headers=headers,
|
||
|
|
data=json.dumps(payload).encode() if payload else None)
|
||
|
|
try:
|
||
|
|
with opener.open(req, timeout=20) as response:
|
||
|
|
return response.status, json.load(response)
|
||
|
|
except urllib.error.HTTPError as error:
|
||
|
|
return error.code, {}
|
||
|
|
|
||
|
|
checks = []
|
||
|
|
status, auth = request("/api/v1/auth/login", payload={key: credentials[key] for key in ("email", "password")})
|
||
|
|
passed = status == 200 and auth.get("active_tenant", {}).get("id") == TARGET_TENANT
|
||
|
|
checks.append({"check": "admin_login_in_acceptance_tenant", "passed": passed})
|
||
|
|
if not passed:
|
||
|
|
return checks
|
||
|
|
status, data = request(f"/api/v1/tenants/{TARGET_TENANT}/members?page=1&page_size=100", auth["token"])
|
||
|
|
members = {m["user_id"]: m for m in data.get("data", {}).get("members", [])}
|
||
|
|
checks.append({"check": "admin_member_list_available", "passed": status == 200})
|
||
|
|
for item in manifest["users"]:
|
||
|
|
member = members.get(item["target_id"], {})
|
||
|
|
expected_status = "active" if item["action"] == "map_existing_admin" else "suspended"
|
||
|
|
checks.append({"check": f"api_source_user_{item['source_user']['id']}_membership",
|
||
|
|
"passed": member.get("role") == item["target_membership_role"] and member.get("status") == expected_status})
|
||
|
|
if item["action"] == "create_suspended":
|
||
|
|
status, response = request("/api/v1/auth/login", payload={"email": item["target_email"], "password": "etl-disabled-login-probe"})
|
||
|
|
checks.append({"check": f"api_source_user_{item['source_user']['id']}_login_denied",
|
||
|
|
"passed": status in (401, 403) and not response.get("token")})
|
||
|
|
status, _ = request(f"/api/v1/tenants/{TARGET_TENANT}/members")
|
||
|
|
checks.append({"check": "anonymous_membership_list_denied", "passed": status in (401, 403)})
|
||
|
|
return checks
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("--manifest", type=Path, required=True)
|
||
|
|
parser.add_argument("--out", type=Path, required=True)
|
||
|
|
parser.add_argument("--credentials", type=Path)
|
||
|
|
args = parser.parse_args()
|
||
|
|
if args.out.exists():
|
||
|
|
parser.error("Output already exists")
|
||
|
|
manifest = json.loads(args.manifest.read_text())
|
||
|
|
before = target_snapshot()
|
||
|
|
output = sql_run(regression_sql(manifest))
|
||
|
|
checks = [json.loads(line) for line in output.splitlines() if line.startswith("{")]
|
||
|
|
after = target_snapshot()
|
||
|
|
checks.append({"check": "all_test_changes_rolled_back", "passed": before == after})
|
||
|
|
if args.credentials:
|
||
|
|
checks.extend(api_checks(manifest, args.credentials))
|
||
|
|
result = {"passed": all(c["passed"] for c in checks), "committed": False, "checks": checks}
|
||
|
|
save_json(args.out, result)
|
||
|
|
print(json.dumps(result))
|
||
|
|
return 0 if result["passed"] else 1
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|