#!/usr/bin/env python3
"""Suppression de compte (decision de Sebastien, 18/10/2026). Garde-fous du point d'entree (session, pas « en tant que », pas un compte d'equipe, adresse retapee, mot de passe, debit),
suppression de TOUT le contenu liste dans le registre (chaque table « delete » est remplie avant et verifiee vide apres), conservation des pieces comptables / de moderation / des messages,
contenus des AUTRES membres intacts, fichiers du dossier du createur, ordre et reprise, et un garde-fou de COMPLETUDE : une table qui porte un identifiant de membre sans etre classee fait echouer ce test."""
import json, os, re, subprocess, sys, uuid
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from lib import ROOT, call, mk_user, ok, reset_rate_limits, sql, summary

def section(t): print('\n■ ' + t); reset_rate_limits()
SINK = os.environ.get('MAIL_TEST_SINK', '')
def sent_mails(): return [json.loads(l) for l in open(SINK, encoding='utf-8').read().split('\n') if l.strip()] if SINK and os.path.exists(SINK) else []
def php(code): return subprocess.run(['php', '-d', 'display_errors=0', '-r', '$_SERVER["REQUEST_METHOD"]="GET"; require "api/_account_delete.php"; ' + code], capture_output=True, text=True, cwd=ROOT).stdout.strip()
def q(v): return "'" + str(v).replace('\\', '\\\\').replace("'", "\\'") + "'"
def cnt(q_): return int(sql(q_))
def table_rows(t, col, uid): return cnt(f"SELECT COUNT(*) FROM {t} WHERE {col} = '{uid}'")
POLICY = json.loads(php('echo json_encode(ACCOUNT_DELETE_POLICY);'))
N = [0]
def uniq_n(): N[0] += 1; return N[0]
def uniq(n=10): N[0] += 1; return (uuid.uuid4().hex[:6] + str(N[0]))[:n]
def fill(t, **vals):
    """Insere une ligne dans t : valeurs données + valeurs plausibles pour les autres colonnes obligatoires ; les parents (clés étrangères) sont créés au besoin. → dict des valeurs."""
    cols = [r.split('\t') for r in sql(f"SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT IS NOT NULL, EXTRA, COLUMN_TYPE, IFNULL(CHARACTER_MAXIMUM_LENGTH, 0) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{t}' ORDER BY ORDINAL_POSITION").split('\n')]
    fks = {r.split('\t')[0]: r.split('\t')[1:] for r in sql(f"SELECT COLUMN_NAME, REFERENCED_TABLE_NAME, REFERENCED_COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{t}' AND REFERENCED_TABLE_NAME IS NOT NULL").split('\n') if r.strip()}
    out = dict(vals)
    for name, typ, nullable, has_def, extra, ctype, maxlen in cols:
        if name in out or 'auto_increment' in extra or 'GENERATED' in extra.upper(): continue
        if name in fks:
            ref, rcol = fks[name]
            if ref == 'users': out[name] = 'ad_b'
            else: out[name] = fill(ref)[rcol]
            continue
        if nullable == 'YES' or has_def == '1': continue
        ml = max(1, min(int(maxlen or 12), 12))
        if name.endswith('json') or name.endswith('_config'): out[name] = '{}'
        elif typ in ('varchar', 'char', 'text', 'tinytext', 'mediumtext', 'longtext'): out[name] = uniq(ml)
        elif typ in ('int', 'bigint', 'smallint', 'mediumint', 'tinyint', 'decimal', 'float', 'double'): out[name] = 1 if typ == 'tinyint' else 1000 + uniq_n()
        elif typ in ('datetime', 'timestamp'): out[name] = '2026-06-01 10:00:00'
        elif typ == 'date': out[name] = '2026-06-01'
        elif typ == 'enum': out[name] = re.findall(r"'([^']*)'", ctype)[0]
        elif typ in ('json', 'longblob', 'blob'): out[name] = '{}'
        else: out[name] = uniq(ml)
    names = ', '.join(out); vals_sql = ', '.join(q(v) for v in out.values())
    last = sql(f"INSERT INTO {t} ({names}) VALUES ({vals_sql}); SELECT LAST_INSERT_ID()").split()[-1]
    for name, typ, nullable, has_def, extra, ctype, maxlen in cols:
        if 'auto_increment' in extra and name not in out: out[name] = last
    return out
def wipe():
    p = "'ad\\_%'"
    for t, (cols, pol, why) in POLICY.items():
        if pol == 'delete' or pol == 'special':
            try: sql("DELETE FROM " + t + " WHERE " + ' OR '.join(f"{c} LIKE {p}" for c in cols))
            except Exception: pass
    sql(f"DELETE FROM refresh_tokens WHERE user_id LIKE {p}; DELETE FROM candidatures WHERE email LIKE 'ad\\_%'; DELETE FROM users WHERE id LIKE {p}")
    for t in ('invoices', 'payments', 'parcours_credits', 'reports', 'user_subscriptions', 'moderation_audit', 'messages'):
        try: sql(f"DELETE FROM {t} WHERE " + ' OR '.join(f"{c} LIKE {p}" for c in POLICY[t][0]))
        except Exception: pass
    try: sql("DELETE FROM conversations WHERE created_by LIKE 'ad\\_%'")
    except Exception: pass
    subprocess.run(['rm', '-rf', os.path.join(ROOT, 'uploads/artisans/ad_a'), os.path.join(ROOT, 'uploads/artisans/ad_b')])

wipe()
for uid, role in (('ad_a', 'client_essentiel'), ('ad_b', 'client_essentiel'), ('ad_c', 'client_essentiel'), ('ad_pw', 'client_essentiel'), ('ad_staff', 'admin'), ('ad_resp', 'responsable')): mk_user(uid, role, 'Prenom', uid.upper())
sql("UPDATE users SET created_at = NOW() - INTERVAL 60 DAY WHERE id LIKE 'ad\\_%'")
sql("UPDATE users SET lang = 'en', firstname = '<b>Joe</b>' WHERE id = 'ad_pw'")
HASH = php('echo password_hash("secret-pass-123", PASSWORD_DEFAULT);'); sql(f"UPDATE users SET password_hash = '{HASH}' WHERE id = 'ad_pw'")
THEME = int(sql("SELECT id FROM post_themes WHERE active = 1 AND system_key IS NULL ORDER BY id LIMIT 1"))
DEL = lambda who, body=None, headers=None: call('POST', who, '/account-delete', body if body is not None else {}, headers=headers)
exists = lambda uid: cnt(f"SELECT COUNT(*) FROM users WHERE id = '{uid}'") == 1

section("Garde-fou de complétude")
ok(json.loads(php('echo json_encode(account_delete_uncovered_columns(db()));')) == [], f'TOUTE table qui porte un identifiant de membre est classée (« delete » ou « keep », avec sa raison) ou nettoyée par une clé étrangère : {len(POLICY)} tables classées')
sql("CREATE TABLE zz_acct_probe (id INT AUTO_INCREMENT PRIMARY KEY, user_id VARCHAR(20))"); found = json.loads(php('echo json_encode(account_delete_uncovered_columns(db()));')); sql("DROP TABLE zz_acct_probe")
ok(found == ['zz_acct_probe.user_id'], 'contrôle positif : une NOUVELLE table avec une colonne user_id, non classée, est détectée (le test échouerait)')
ok(all(len(v) == 3 and v[1] in ('delete', 'keep', 'special') and v[2] for v in POLICY.values()) and sum(1 for v in POLICY.values() if v[1] == 'keep' and 'juriste' in v[2]) >= 4, 'chaque table classée a une politique valide ET une raison écrite ; les conservations comptables ou de messages sont signalées « à faire valider par un juriste »')

ok(json.loads(php('echo json_encode(account_delete_numeric_columns(db()));')) == [], 'AUCUNE colonne à supprimer n\'est numérique : comparer un nombre à l\'identifiant texte d\'un membre y lirait « = 0 » et détruirait des lignes étrangères')
legacy = [t for t, (c, pol, why) in POLICY.items() if 'ENTIÈRE héritée' in why]; ok(len(legacy) == 5, f'les {len(legacy)} tables à colonne entière héritée sont classées « keep » avec cette raison ({", ".join(legacy)})')
section("Point d'entrée : garde-fous")
MAILS0 = len(sent_mails()); ok(bool(SINK), 'la sortie de test des e-mails (MAIL_TEST_SINK) est configurée : lancer par tests/run.sh')
ok(call('POST', None, '/account-delete', {'email': 'ad_a@test.test'})[0] == 401 and call('GET', 'ad_a', '/account-delete')[0] == 405, 'sans session : 401 ; méthode GET : 405')
st, d, _ = DEL('ad_staff', {'email': 'ad_staff@test.test'}); ok(st == 403 and d['code'] == 'staff_account' and exists('ad_staff'), 'un administrateur ne peut pas supprimer son compte ici (403 « staff_account ») : il existe toujours')
st, d, _ = DEL('ad_resp', {'email': 'ad_resp@test.test'}); ok(st == 403 and d['code'] == 'staff_account' and exists('ad_resp'), 'ni un responsable')
imp = php('echo jwt_issue(["id" => "ad_a", "role" => "client_essentiel", "imp" => true], 3600);'); st, d, _ = call('POST', None, '/account-delete', {'email': 'ad_a@test.test'}, headers={'Authorization': 'Bearer ' + imp})
ok(st == 403 and d['code'] == 'impersonated' and exists('ad_a'), 'une connexion « en tant que » faite par un administrateur ne peut PAS supprimer le compte du membre (403 « impersonated »)')
for body, label in (({}, 'sans adresse'), ({'email': ''}, 'adresse vide'), ({'email': 'ad_b@test.test'}, 'l\'adresse D\'UN AUTRE membre'), ({'email': 'autre@exemple.fr'}, 'une adresse quelconque')):
    st, d, _ = DEL('ad_a', body); ok(st == 422 and d['code'] == 'email_mismatch' and exists('ad_a'), f'{label} : 422 « email_mismatch », rien n\'est supprimé')
reset_rate_limits(); st, d, _ = DEL('ad_pw', {'email': 'ad_pw@test.test'}); ok(st == 422 and d['code'] == 'password_required' and exists('ad_pw'), 'un compte AVEC mot de passe : le mot de passe est exigé (422 « password_required »)')
st, d, _ = DEL('ad_pw', {'email': 'ad_pw@test.test', 'password': 'mauvais'}); ok(st == 403 and d['code'] == 'password_wrong' and exists('ad_pw'), 'mauvais mot de passe : 403 « password_wrong », rien n\'est supprimé')
ok(len(sent_mails()) == MAILS0, 'AUCUN e-mail pour les tentatives refusées (401, 403, 422)')
reset_rate_limits(); codes = [DEL('ad_c', {'email': 'x@x.fr'})[0] for _ in range(7)]; ok(codes[:5] == [422] * 5 and codes[5] == 429, f'5 essais par heure et par adresse IP, puis 429 ({codes})'); reset_rate_limits()

section("Scénario complet : un créateur avec des données partout")
UID = 'ad_a'; seeded = {}
# contenu réel créé par l'API
pa1 = call('POST', UID, '/posts', {'body': 'Ma publication 1', 'theme_id': THEME, 'visibility': 'all'})[1]['post']['id']; pa2 = call('POST', UID, '/posts', {'body': 'Ma publication 2', 'theme_id': THEME, 'visibility': 'all'})[1]['post']['id']
pb = call('POST', 'ad_b', '/posts', {'body': 'Publication de Bruno', 'theme_id': THEME, 'visibility': 'all'})[1]['post']['id']; pb2 = call('POST', 'ad_b', '/posts', {'body': 'Autre de Bruno', 'theme_id': THEME, 'visibility': 'all'})[1]['post']['id']
c_on_b = call('POST', UID, f'/posts/{pb}/comments', {'body': 'Mon commentaire chez Bruno'})[1]['comment']['id']; call('POST', 'ad_b', f'/posts/{pb}/comments', {'body': 'Réponse de Bruno à ma remarque', 'parent_id': c_on_b})
c_on_a = call('POST', 'ad_b', f'/posts/{pa1}/comments', {'body': 'Bruno commente chez moi'})[1]['comment']['id']; bc = call('POST', 'ad_b', f'/posts/{pb2}/comments', {'body': 'Commentaire de Bruno chez lui'})[1]['comment']['id']
call('POST', UID, f'/posts/{pb}/like', {}); call('POST', UID, f'/posts/{pb2}/save', {}); call('POST', 'ad_b', f'/posts/{pa1}/like', {}); call('POST', 'ad_b', f'/posts/{pa2}/save', {})
sql(f"INSERT INTO posts (user_id, theme_id, body, visibility, kind, auto_event, auto_day, created_at, published_at) VALUES ('{UID}', {THEME}, '', 'all', 'auto', 'action_started', CURDATE(), NOW(), NOW())"); auto_id = int(sql(f"SELECT MAX(id) FROM posts WHERE user_id = '{UID}'"))
# toutes les autres tables « delete » : une ligne par colonne de membre
generic, skipped = 0, []
for t, (cols, pol, why) in POLICY.items():
    if pol != 'delete': continue
    for col in cols:
        try: fill(t, **{col: UID}); generic += 1
        except Exception as e: skipped.append(f'{t}.{col}')
ok(not skipped, f'une ligne a été créée pour {generic} colonnes de membre dans toutes les tables « delete »' + (f' — NON remplies : {skipped}' if skipped else ''))
# tables « keep »
kept = {}
for t, (cols, pol, why) in POLICY.items():
    if pol != 'keep' or t in ('conversations', 'audits', 'messaging_settings', 'messaging_settings_log', 'mobile_nav_config', 'threads_tokens', 'file_shares'): continue
    try: fill(t, **{cols[0]: UID}); kept[t] = table_rows(t, cols[0], UID)
    except Exception as e: kept[t] = None
cand = fill('candidatures', user_id=UID, email='ad_a_cand@test.test')
conv = fill('conversations', created_by=UID)['id']; fill('conversation_members', conversation_id=conv, user_id=UID); fill('conversation_members', conversation_id=conv, user_id='ad_b'); msg = fill('messages', conversation_id=conv, user_id=UID, body='Message conservé')['id']
for sub in ('gallery', 'designs'): os.makedirs(os.path.join(ROOT, f'uploads/artisans/ad_a/{sub}'), exist_ok=True); open(os.path.join(ROOT, f'uploads/artisans/ad_a/{sub}/f.webp'), 'wb').write(b'x')
os.makedirs(os.path.join(ROOT, 'uploads/artisans/ad_b/gallery'), exist_ok=True); open(os.path.join(ROOT, 'uploads/artisans/ad_b/gallery/f.webp'), 'wb').write(b'x')
legacy_rows = {}
for tl in legacy:
    try: fill(tl, user_id=0); legacy_rows[tl] = cnt(f"SELECT COUNT(*) FROM {tl} WHERE user_id = 0")
    except Exception: pass
ok(len(legacy_rows) >= 3, f'lignes héritées « user_id = 0 » créées dans {len(legacy_rows)} tables pour vérifier qu\'elles survivent')
others = {'b_posts': table_rows('posts', 'user_id', 'ad_b'), 'b_users': cnt("SELECT COUNT(*) FROM users WHERE id IN ('ad_b', 'ad_c', 'ad_pw')")}
before_keep = {t: kept[t] for t in kept}
token_before = call('GET', UID, '/auth/me')[0]
st, d, h = DEL(UID, {'email': '  AD_A@test.test '}); ok(st == 200 and d == {'ok': True}, 'suppression avec l\'adresse retapée (casse et espaces ignorés) : 200 { ok: true }')
M0 = len(sent_mails()); ok(token_before == 200 and not exists(UID) and call('GET', UID, '/auth/me')[0] == 401, 'le compte n\'existe plus et son ancien jeton ne marche plus (401) : la session est coupée immédiatement')
ok(any(k.lower() == 'set-cookie' for k in h), 'la réponse efface les cookies de session')
mails = sent_mails()[M0 - 1:] if False else sent_mails()[MAILS0:]
ok(len(mails) == 1 and mails[0]['to'] == 'ad_a@test.test' and mails[0]['subject'] == 'Ton compte Uluana a été supprimé', 'UN e-mail de confirmation part à l\'adresse du compte, en français (langue du compte)')
body = mails[0]['html'] if mails else ''; ok('Ce qui a été supprimé' in body and 'Ce qui est conservé' in body and 'Ancien membre' in body and 'factures' in body and 'contact@uluana.art' in body and 'tout de suite' in body, 'il dit ce qui est supprimé et ce qui est conservé, et donne l\'adresse de contact si le membre n\'est pas à l\'origine de la demande')
ok(not re.search(r'ad_a|https?://|token|password|mot de passe|<script', body) and body.count('href=') == 1, 'il ne contient AUCUN lien vers le compte (il n\'existe plus), aucun identifiant, aucun jeton : un seul lien « mailto: » de contact')
left = {t: sum(table_rows(t, c, UID) for c in cols) for t, (cols, pol, why) in POLICY.items() if pol in ('delete', 'special')}
ok(all(v == 0 for v in left.values()), f'AUCUNE ligne ne subsiste dans les {len(left)} tables à supprimer' + (' — restent : ' + str({t: v for t, v in left.items() if v}) if any(left.values()) else ''))
ok(cnt(f"SELECT COUNT(*) FROM posts WHERE id IN ({pa1}, {pa2}, {auto_id})") == 0 and cnt(f"SELECT COUNT(*) FROM comments WHERE id IN ({c_on_b}, {c_on_a})") == 0, 'ses publications (écrites ET automatique) et leurs commentaires ont disparu, ainsi que son commentaire chez Bruno')
ok(cnt(f"SELECT COUNT(*) FROM comments WHERE parent_id = {c_on_b}") == 0 and cnt(f"SELECT COUNT(*) FROM comments WHERE post_id = {pa1}") == 0, 'la réponse que Bruno avait faite à SON commentaire part avec lui (le fil de discussion n\'a plus de racine) ; et les commentaires reçus sur ses publications aussi')
ok(cnt(f"SELECT COUNT(*) FROM post_likes WHERE post_id IN ({pa1}, {pa2})") == 0 and cnt(f"SELECT COUNT(*) FROM post_saves WHERE post_id IN ({pa1}, {pa2})") == 0 and cnt(f"SELECT COUNT(*) FROM post_likes WHERE user_id = '{UID}'") == 0, 'les j\'aime et enregistrements, reçus ET donnés, ont disparu')
ok(table_rows('posts', 'user_id', 'ad_b') == others['b_posts'] and cnt(f"SELECT COUNT(*) FROM posts WHERE id IN ({pb}, {pb2})") == 2 and cnt(f"SELECT COUNT(*) FROM comments WHERE id = {bc}") == 1 and cnt("SELECT COUNT(*) FROM users WHERE id IN ('ad_b', 'ad_c', 'ad_pw')") == others['b_users'], 'les contenus et comptes des AUTRES membres sont INTACTS (publications, commentaire de Bruno chez lui)')
now_keep = {t: sum(table_rows(t, c, UID) for c in POLICY[t][0]) for t in kept}; ok(all(now_keep[t] >= 1 for t in kept if kept[t]) and sum(1 for t in kept if kept[t]) >= 6, f'conservé : {", ".join(t for t in kept if kept[t])} (factures, paiements, signalements, abonnements, crédits, audit de modération…)')
ok(all(cnt(f"SELECT COUNT(*) FROM {tl} WHERE user_id = 0") == n for tl, n in legacy_rows.items()), 'les lignes héritées « user_id = 0 » des tables à colonne entière sont INTACTES : la suppression d\'un compte ne détruit rien d\'étranger')
ok(cnt(f"SELECT COUNT(*) FROM candidatures WHERE email = 'ad_a_cand@test.test' AND user_id IS NULL") == 1, 'la candidature réglée est conservée, détachée du compte (user_id nul, clé étrangère SET NULL)')
ok(cnt(f"SELECT COUNT(*) FROM messages WHERE id = {msg}") == 1 and cnt(f"SELECT COUNT(*) FROM users WHERE id = '{UID}'") == 0 and table_rows('conversation_members', 'user_id', UID) == 0 and table_rows('conversation_members', 'user_id', 'ad_b') >= 1, 'ses messages privés restent (leur auteur n\'existe plus : « Ancien membre »), il a quitté ses conversations, Bruno garde la sienne')
ok(not os.path.exists(os.path.join(ROOT, 'uploads/artisans/ad_a')) and os.path.exists(os.path.join(ROOT, 'uploads/artisans/ad_b/gallery/f.webp')), 'son dossier de fichiers est supprimé (galerie, visuels) ; celui d\'un autre membre est intact')

section("Fonction et fichiers : cas limites")
ok(json.loads(php('echo json_encode(account_delete(db(), "ad_a"));'))['reason'] == 'not_found', 'supprimer un compte déjà supprimé : « not_found », sans erreur (reprise sûre)')
os.makedirs(os.path.join(ROOT, 'uploads/artisans/ad_b/x'), exist_ok=True)
for bad in ('../ad_b', '..', '', 'ad_b/../ad_b', '/etc', 'a b', 'ad_b\\x00', 'é'): ok(php(f'echo json_encode(account_delete_dir({json.dumps(bad)}));') == 'false', f'identifiant dangereux {bad!r} : refusé, aucun fichier touché') if False else None
res = [php(f'echo json_encode(account_delete_dir({json.dumps(b)}));') for b in ('../ad_b', '..', '', 'ad_b/../ad_b', '/etc', 'a b', 'é')]; ok(all(r == 'false' for r in res) and os.path.exists(os.path.join(ROOT, 'uploads/artisans/ad_b/gallery/f.webp')), f'identifiants dangereux (../, vide, chemin absolu, espace, accent) : refusés, aucun fichier touché ({res})')
ok(php('echo json_encode(account_delete_dir("ad_inexistant"));') == 'false', 'un dossier qui n\'existe pas : rien à supprimer, pas d\'erreur')
st, d, _ = DEL('ad_pw', {'email': 'AD_PW@test.test', 'password': 'secret-pass-123'}); ok(st == 200 and not exists('ad_pw'), 'avec le BON mot de passe, la suppression d\'un compte à mot de passe réussit')
en = [m for m in sent_mails() if m['to'] == 'ad_pw@test.test']; ok(len(en) == 1 and en[0]['subject'] == 'Your Uluana account has been deleted' and 'What was deleted' in en[0]['html'] and 'Former member' in en[0]['html'], 'compte en anglais : e-mail en anglais')
ok('&lt;b&gt;Joe&lt;/b&gt;' in en[0]['html'] and '<b>Joe</b>' not in en[0]['html'], 'le prénom est ÉCHAPPÉ (un prénom piégé « <b>… » ne s\'injecte pas dans l\'e-mail)')
ok(php('echo json_encode(account_delete_send_mail(["email" => "x@x.fr", "firstname" => "A", "lang" => "fr"], function ($to, $s, $h) { throw new RuntimeException("SMTP en panne"); }));') == 'false', 'un envoi qui lève une exception : la fonction rend false SANS exception (la suppression a déjà eu lieu, le membre ne voit rien)')
ok(php('echo json_encode([account_delete_send_mail(["email" => "pas-un-email"], fn() => true), account_delete_send_mail(["email" => ""], fn() => true), account_delete_send_mail([], fn() => true)]);') == '[false,false,false]', 'adresse invalide, vide ou absente : rien n\'est envoyé')
ok(php('echo json_encode(account_delete_send_mail(["email" => "x@x.fr", "firstname" => "A"], fn($to, $s, $h) => $to === "x@x.fr" && str_contains($s, "supprimé")));') == 'true' and php('[$s] = account_delete_mail([]); echo $s;') == 'Ton compte Uluana a été supprimé' and php('[$s, $h] = account_delete_mail(["lang" => "es"]); echo $s;') == 'Ton compte Uluana a été supprimé', 'sans langue connue (ou langue inconnue) : français ; sans prénom : « Bonjour, »')

section("Registre : conformité au code")
src = open(os.path.join(ROOT, 'api', '_account_delete.php'), encoding='utf-8').read(); ok('DELETE FROM users WHERE id' in src and src.index('posts_delete_post') < src.index('DELETE FROM users WHERE id'), 'dans le code, le contenu part AVANT la ligne users (un échec laisse le compte, la suppression se rejoue)')
ok(not re.search(r"password|token", json.dumps(call('POST', 'ad_b', '/account-delete', {'email': 'x'})[1]).lower().replace('password_', '')), 'la réponse d\'erreur ne divulgue rien d\'autre que le code et le message')
wipe(); summary()