#!/usr/bin/env python3
"""Notifications push (CDC n° 13) : configuration, abonnement par appareil (et refus des adresses détournées), préférences, envoi de bout en bout
(message chiffré DÉCHIFFRÉ par une référence indépendante en Node.js, signature VAPID vérifiée), heures calmes, appareils disparus, conformité RFC 8291.
Nécessite les variables VAPID_PUBLIC, VAPID_PRIVATE, VAPID_SUBJECT et PUSH_TEST_SINK : tests/run.sh les fournit."""
import base64, datetime, json, os, subprocess, sys, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from lib import ROOT, call, mk_user, ok, reset_rate_limits, sql, summary

SINK = os.environ.get('PUSH_TEST_SINK', ''); PUB = os.environ.get('VAPID_PUBLIC', '')
if not SINK or not PUB or not os.environ.get('VAPID_PRIVATE'): print("  KO  VAPID_PUBLIC / VAPID_PRIVATE / PUSH_TEST_SINK absents : lancer par tests/run.sh"); sys.exit(1)
REF = os.path.join(ROOT, 'tests', 'webpush_ref.js')
def ref(q): return json.loads(subprocess.run(['node', REF], input=json.dumps(q), capture_output=True, text=True).stdout)
def device(): return ref({'mode': 'keygen'})
def sub_body(dev, endpoint): return {'endpoint': endpoint, 'keys': {'p256dh': dev['public'], 'auth': dev['auth']}}
def fcm(tag=None): return 'https://fcm.googleapis.com/fcm/send/' + (tag or base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip('='))
def clear_sink(): open(SINK, 'w').close()
def sink(): return [json.loads(l) for l in open(SINK, encoding='utf-8').read().split('\n') if l.strip()]
def wait_sink(n, timeout=5):
    t0 = time.time()
    while time.time() - t0 < timeout:
        if len(sink()) >= n: return sink()
        time.sleep(0.1)
    return sink()
def nothing(wait=1.2): time.sleep(wait); return sink() == []
def message(line, dev): r = ref({'mode': 'decrypt', 'body': line['body'], 'private': dev['private'], 'public': dev['public'], 'auth': dev['auth']}); return json.loads(r['plaintext']) if r.get('ok') else None
def php(code, env=None):
    e = dict(os.environ); e.update(env or {})
    r = subprocess.run(['php', '-d', 'display_errors=0', '-d', 'error_reporting=0', '-r', '$_SERVER["REQUEST_METHOD"]="GET"; require "api/_push.php"; ' + code], capture_output=True, text=True, cwd=ROOT, env=e)
    return r.stdout.strip()
def section(t): print('\n■ ' + t); reset_rate_limits()
def setting(k, v): sql(f"INSERT INTO messaging_settings (setting_key, setting_value) VALUES ('{k}', '{json.dumps(v)}') ON DUPLICATE KEY UPDATE setting_value = '{json.dumps(v)}'")
def n_subs(uid): return int(sql(f"SELECT COUNT(*) FROM push_subscriptions WHERE user_id = '{uid}'"))
def prefs(who, body): return call('PUT', who, '/notification-prefs', body)
def publish(who, body='Un texte'): st, d, _ = call('POST', who, '/posts', {'body': body, 'theme_id': THEME, 'visibility': 'all'}); assert st == 201, (st, d); return d['post']['id']
def nrows(who): return len(call('GET', who, '/notifications')[1]['items'])
def wipe():
    p = "'p\\_%'"
    sql(f"DELETE FROM push_subscriptions WHERE user_id LIKE {p}")
    sql(f"DELETE cl FROM comment_likes cl JOIN comments c ON c.id = cl.comment_id WHERE c.user_id LIKE {p} OR c.post_id IN (SELECT id FROM posts WHERE user_id LIKE {p})")
    sql(f"DELETE FROM mentions WHERE member_id LIKE {p} OR (subject_type = 'post' AND subject_id IN (SELECT id FROM posts WHERE user_id LIKE {p}))")
    sql(f"DELETE FROM comments WHERE user_id LIKE {p} OR post_id IN (SELECT id FROM posts WHERE user_id LIKE {p})")
    for t in ('post_likes', 'post_saves', 'post_media'): sql(f"DELETE FROM {t} WHERE post_id IN (SELECT id FROM posts WHERE user_id LIKE {p})")
    sql(f"DELETE FROM post_likes WHERE user_id LIKE {p}; DELETE FROM posts WHERE user_id LIKE {p}; DELETE FROM notifications WHERE user_id LIKE {p} OR actor_id LIKE {p}; DELETE FROM notification_prefs WHERE user_id LIKE {p}; DELETE FROM notification_settings WHERE user_id LIKE {p}; DELETE FROM member_follows WHERE follower_id LIKE {p} OR followed_id LIKE {p}")

wipe(); sql("DELETE FROM users WHERE id LIKE 'p\\_%'")
for uid, first, last in [('p_a', 'Anne', 'Adam'), ('p_b', 'Bruno', 'Blanc'), ('p_c', 'Chloé', 'Carré')]: mk_user(uid, 'client_essentiel', first, last)
sql("UPDATE users SET lang = 'en' WHERE id = 'p_c'; UPDATE users SET lang = 'fr' WHERE id IN ('p_a', 'p_b')")
THEME = int(sql("SELECT id FROM post_themes WHERE active = 1 AND system_key IS NULL ORDER BY id LIMIT 1").strip()); clear_sink()
DA, DB, DC = device(), device(), device()

section("Configuration publique")
st, d, h = call('GET', None, '/push-config'); raw = json.dumps(d)
ok(st == 200 and d == {'enabled': True, 'public_key': PUB} and h.get('Cache-Control') == 'public, max-age=300', 'GET /push-config sans connexion : { enabled: true, public_key } (la clé VAPID du serveur), cache de 5 minutes')
pk = base64.urlsafe_b64decode(PUB + '=' * (-len(PUB) % 4)); ok(len(pk) == 65 and pk[0] == 4, 'la clé publique est un point P-256 non compressé (65 octets, préfixe 0x04)')
ok(os.environ['VAPID_PRIVATE'] not in raw and 'private' not in raw.lower(), 'la clé PRIVÉE n\'apparaît jamais dans la réponse')
ok(call('POST', None, '/push-config', {})[0] == 405 and call('DELETE', 'p_a', '/push-config')[0] == 405, 'autre méthode : 405')

section("Validation de la configuration du serveur (PHP en ligne de commande)")
good = {'VAPID_PUBLIC': PUB, 'VAPID_PRIVATE': os.environ['VAPID_PRIVATE'], 'VAPID_SUBJECT': 'mailto:contact@uluana.art'}
cfg = lambda **o: php('echo webpush_config() === null ? "null" : "ok";', {**good, **o})
ok(cfg() == 'ok' and cfg(VAPID_SUBJECT='https://uluana.art/contact') == 'ok', 'clés valides : push disponible (sujet « mailto: » ou « https:// »)')
ok(cfg(VAPID_SUBJECT='contact@uluana.art') == 'null' and cfg(VAPID_SUBJECT='') == 'null' and cfg(VAPID_SUBJECT='http://x.fr') == 'null', 'sujet sans « mailto: » ni « https:// » : push désactivé')
ok(cfg(VAPID_PUBLIC='abc') == 'null' and cfg(VAPID_PRIVATE='abc') == 'null' and cfg(VAPID_PUBLIC=PUB[:-4]) == 'null', 'clé trop courte ou tronquée : push désactivé')
ok(php('echo json_encode(push_enabled());', {**good, 'VAPID_PUBLIC': '', 'VAPID_PRIVATE': ''}) == 'false', 'sans clés : push désactivé (aucune erreur)')

section("Adresses d'abonnement : seulement de vrais services de push (protection contre le détournement du serveur)")
tab = [('https://fcm.googleapis.com/fcm/send/abc', 1), ('https://updates.push.services.mozilla.com/wpush/v2/abc', 1), ('https://web.push.apple.com/QWxs', 1), ('https://wns2-par02p.notify.windows.com/w/?token=abc', 1), ('https://fcm.googleapis.com:443/fcm/send/abc', 1),
       ('http://fcm.googleapis.com/fcm/send/abc', 0), ('https://evil.example/fcm.googleapis.com', 0), ('https://fcm.googleapis.com.evil.com/x', 0), ('https://evilfcm.googleapis.com.example/x', 0), ('https://notfcm.googleapis.com/x', 0),
       ('https://localhost/x', 0), ('https://127.0.0.1/x', 0), ('https://169.254.169.254/latest/meta-data', 0), ('https://[::1]/x', 0), ('https://10.0.0.5/x', 0), ('https://user@fcm.googleapis.com/x', 0), ('https://user:pw@fcm.googleapis.com/x', 0),
       ('https://fcm.googleapis.com:8443/x', 0), ('https://fcm.googleapis.com/x#frag', 0), ('https://fcm.googleapis.com/x y', 0), ('https://fcm.googleapis.com/' + 'a' * 600, 0), ('ftp://fcm.googleapis.com/x', 0), ('//fcm.googleapis.com/x', 0), ('', 0), ('javascript:alert(1)', 0)]
code = 'foreach (json_decode(stream_get_contents(STDIN), true) as $u) echo webpush_endpoint_ok($u) ? "1" : "0";'
out = subprocess.run(['php', '-d', 'display_errors=0', '-r', '$_SERVER["REQUEST_METHOD"]="GET"; require "api/_push.php"; ' + code], input=json.dumps([t[0] for t in tab]), capture_output=True, text=True, cwd=ROOT).stdout
bad = [(t[0][:50], t[1], o) for t, o in zip(tab, out) if str(t[1]) != o]
ok(len(out) == len(tab) and not bad, f'{len(tab)} adresses testées : les services de push (FCM, Mozilla, Apple, Windows) passent ; hôte interne, IP, localhost, métadonnées cloud, identifiants, port, fragment, http, adresse trop longue, hôte déguisé : refusés' + (f' — écarts : {bad[:3]}' if bad else ''))
ok(php('echo json_encode([webpush_endpoint_ok(null), webpush_endpoint_ok(12), webpush_endpoint_ok(["x"]), webpush_endpoint_ok(false)]);') == '[false,false,false,false]', 'une valeur qui n\'est pas du texte : refusée, sans erreur')

section("S'abonner")
ok(call('GET', None, '/push-subscriptions')[0] == 401 and call('POST', None, '/push-subscriptions', sub_body(DA, fcm()))[0] == 401 and call('DELETE', None, '/push-subscriptions', {'endpoint': 'x'})[0] == 401 and call('POST', None, '/push-subscriptions/test', {})[0] == 401, 'sans connexion : 401 partout')
ok(call('PATCH', 'p_a', '/push-subscriptions')[0] == 404 and call('GET', 'p_a', '/push-subscriptions/inconnu')[0] == 404, 'méthode ou route inconnue : 404')
def refuse(body, key, why):
    st, d, _ = call('POST', 'p_a', '/push-subscriptions', body); ok(st == 422 and key in d['errors'], f'refusé (422, « {key} ») : {why}')
refuse({'endpoint': 'https://evil.example/x', 'keys': {'p256dh': DA['public'], 'auth': DA['auth']}}, 'endpoint', 'adresse qui n\'est pas un service de push')
refuse({'endpoint': 'http://169.254.169.254/', 'keys': {'p256dh': DA['public'], 'auth': DA['auth']}}, 'endpoint', 'adresse interne (métadonnées du serveur)')
refuse({'endpoint': fcm()}, 'keys', 'clés absentes'); refuse({'keys': {'p256dh': DA['public'], 'auth': DA['auth']}}, 'endpoint', 'adresse absente')
refuse({'endpoint': fcm(), 'keys': {'p256dh': 'abc', 'auth': DA['auth']}}, 'keys.p256dh', 'clé publique trop courte')
refuse({'endpoint': fcm(), 'keys': {'p256dh': base64.urlsafe_b64encode(b'\x04' + os.urandom(64)).decode().rstrip('='), 'auth': DA['auth']}}, 'keys.p256dh', 'point qui n\'est PAS sur la courbe P-256')
refuse({'endpoint': fcm(), 'keys': {'p256dh': base64.urlsafe_b64encode(b'\x03' + os.urandom(64)).decode().rstrip('='), 'auth': DA['auth']}}, 'keys.p256dh', 'mauvais préfixe')
refuse({'endpoint': fcm(), 'keys': {'p256dh': DA['public'], 'auth': 'abc'}}, 'keys.auth', 'secret d\'authentification trop court')
refuse({'endpoint': fcm(), 'keys': {'p256dh': DA['public'], 'auth': base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip('=')}}, 'keys.auth', 'secret d\'authentification trop long')
refuse({'endpoint': fcm(), 'keys': {'p256dh': 5, 'auth': 7}}, 'keys.p256dh', 'clés qui ne sont pas du texte'); refuse({'endpoint': fcm(), 'keys': 'x'}, 'keys', 'clés qui ne sont pas un objet'); refuse({'endpoint': ['x'], 'keys': {}}, 'endpoint', 'adresse qui n\'est pas du texte')
ok(n_subs('p_a') == 0, 'après tous ces refus : aucun abonnement enregistré')
EA, EB, EC = fcm(), fcm(), fcm()
st, d, _ = call('POST', 'p_a', '/push-subscriptions', sub_body(DA, EA)); ok(st == 201 and d == {'ok': True, 'devices': 1} and n_subs('p_a') == 1, 'abonner un appareil : 201, 1 appareil')
st, d, _ = call('POST', 'p_a', '/push-subscriptions', sub_body(DA, EA)); ok(st == 201 and d['devices'] == 1 and n_subs('p_a') == 1, 'le même appareil une seconde fois : toujours UN appareil (mis à jour, pas dupliqué)')
call('POST', 'p_b', '/push-subscriptions', sub_body(DB, EB)); call('POST', 'p_c', '/push-subscriptions', sub_body(DC, EC))
ok(sql("SELECT COUNT(*) FROM push_subscriptions WHERE endpoint LIKE 'https://fcm.googleapis.com/%'") != '0' and sql(f"SELECT endpoint_hash FROM push_subscriptions WHERE user_id = 'p_a'") != '' and len(sql("SELECT endpoint_hash FROM push_subscriptions WHERE user_id = 'p_a'")) == 64, 'l\'identifiant d\'un appareil est l\'empreinte SHA-256 de son adresse')
st, d, _ = call('GET', 'p_a', '/push-subscriptions'); raw = json.dumps(d)
ok(st == 200 and d['enabled'] is True and d['max'] == 10 and len(d['devices']) == 1 and set(d['devices'][0]) == {'id', 'device', 'created_at', 'last_used_at'} and EA not in raw and DA['public'] not in raw and DA['auth'] not in raw, 'la liste des appareils ne renvoie JAMAIS l\'adresse ni les clés : { id, nom lisible, créé, dernier envoi }')
call('POST', 'p_c', '/push-subscriptions', sub_body(DC, fcm()), ) ; sql("UPDATE push_subscriptions SET user_agent = 'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit Safari/605' WHERE user_id = 'p_a'")
ok(call('GET', 'p_a', '/push-subscriptions')[1]['devices'][0]['device'] == 'iPhone · Safari', 'nom lisible déduit du navigateur : « iPhone · Safari »')
labels = {'Mozilla/5.0 (Linux; Android 14) AppleWebKit Chrome/120 Mobile Safari/537': 'Android · Chrome', 'Mozilla/5.0 (Windows NT 10.0) Gecko Firefox/121': 'Windows · Firefox', 'Mozilla/5.0 (Macintosh; Intel Mac OS X) Version/17 Safari/605': 'Mac · Safari', 'Mozilla/5.0 (Windows NT 10.0) Chrome/120 Safari/537 Edg/120': 'Windows · Edge', '': 'Appareil', 'curl/8': 'Appareil'}
got = {ua: php('echo push_device_label(json_decode(' + json.dumps(json.dumps(ua)) + '));') for ua in labels}; ok(got == labels, 'noms lisibles : Android · Chrome, Windows · Firefox, Mac · Safari, Windows · Edge, « Appareil » sans information')
st, d, _ = call('POST', 'p_b', '/push-subscriptions', sub_body(DA, EA)); ok(st == 201 and n_subs('p_a') == 0 and n_subs('p_b') == 2, 'un appareil qui change de COMPTE (même adresse) est réaffecté au nouveau : l\'ancien n\'en reçoit plus'); sql("DELETE FROM push_subscriptions WHERE user_id = 'p_b' AND endpoint = '" + EA + "'"); call('POST', 'p_a', '/push-subscriptions', sub_body(DA, EA))
for i in range(12): call('POST', 'p_c', '/push-subscriptions', sub_body(device(), fcm()))
ok(n_subs('p_c') == 10 and call('GET', 'p_c', '/push-subscriptions')[1]['devices'][-1]['id'] > call('GET', 'p_c', '/push-subscriptions')[1]['devices'][0]['id'], '12 appareils ajoutés : 10 au plus, les PLUS ANCIENS sont retirés'); sql("DELETE FROM push_subscriptions WHERE user_id = 'p_c'"); reset_rate_limits(); st, d, _ = call('POST', 'p_c', '/push-subscriptions', sub_body(DC, EC)); assert st == 201, st
st, d, _ = call('DELETE', 'p_a', '/push-subscriptions', {'endpoint': EB}); ok(st == 200 and d['removed'] is False and n_subs('p_b') == 1, 'désabonner l\'appareil de QUELQU\'UN D\'AUTRE : sans effet')
st, d, _ = call('DELETE', 'p_a', '/push-subscriptions', {'endpoint': EA}); ok(st == 200 and d['removed'] is True and n_subs('p_a') == 0, 'désabonner son appareil : supprimé')
ok(call('DELETE', 'p_a', '/push-subscriptions', {'endpoint': EA})[1]['removed'] is False and call('DELETE', 'p_a', '/push-subscriptions', {})[0] == 422 and call('DELETE', 'p_a', '/push-subscriptions', {'endpoint': 5})[0] == 422, 'désabonner deux fois : rien ; sans adresse ou adresse invalide : 422')
reset_rate_limits(); st, d, _ = call('POST', 'p_a', '/push-subscriptions', sub_body(DA, EA)); assert st == 201, st

section("Préférences : le canal « push »")
st, d, _ = call('GET', 'p_b', '/notification-prefs'); T = {t['type']: {c['channel']: c['enabled'] for c in t['channels']} for t in d['types']}
ok(T['message'] == {'in_app': True, 'push': True, 'email': True} and T['mention'] == {'in_app': True, 'push': True} and T['comment'] == {'in_app': True, 'push': True} and T['collaboration'] == {'in_app': True, 'email': True, 'push': True}, 'message, mention, commentaire, collaboration : push activé par défaut (et l\'email « non lu » pour les messages)')
ok(T['like'] == {'in_app': True, 'push': False}, '« j\'aime » : push DÉSACTIVÉ par défaut (trop fréquent)'); ok('moderation' not in T or 'push' not in T.get('moderation', {}), 'modération : aucun canal push (non réglable, toujours dans l\'application)')
st, d, _ = prefs('p_b', {'types': {'moderation': {'push': True}}}); ok(st == 422, 'activer le push pour la modération : refusé')
st, d, _ = prefs('p_b', {'types': {'comment': {'push': False}}}); ok(st == 200 and [c['enabled'] for t in d['types'] if t['type'] == 'comment' for c in t['channels'] if c['channel'] == 'push'] == [False], 'désactiver le push des commentaires : enregistré'); prefs('p_b', {'types': {'comment': {'push': True}}})

section("Un événement envoie un push, chiffré, signé, lisible par l'appareil seulement")
clear_sink(); PB = publish('p_b', 'Une jolie pièce en grès qui ne doit JAMAIS apparaître dans le push'); n0 = nrows('p_b')
call('POST', 'p_a', f'/posts/{PB}/like', {}); ok(nrows('p_b') == n0 + 1 and nothing(), '« j\'aime » : la notification est dans l\'application mais AUCUN push (désactivé par défaut)')
prefs('p_b', {'types': {'like': {'push': True}}}); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); call('POST', 'p_a', f'/posts/{PB}/like', {'like': False}); clear_sink(); call('POST', 'p_a', f'/posts/{PB}/like', {})
L = wait_sink(1); ok(len(L) == 1 and L[0]['endpoint'] == EB, 'avec le push « j\'aime » activé : UN envoi, vers l\'appareil de Bruno')
m = message(L[0], DB) if L else None
ok(m == {'title': 'Uluana', 'body': 'Anne A. a aimé votre publication', 'url': f'/fil?post={PB}', 'tag': f'uluana-like-post-{PB}', 'type': 'like'}, 'DÉCHIFFRÉ par la référence Node avec la clé de l\'appareil : « Anne A. a aimé votre publication », lien vers la publication, étiquette qui remplace les doublons')
ok(m and 'grès' not in json.dumps(m, ensure_ascii=False) and 'jolie' not in json.dumps(m, ensure_ascii=False), 'AUCUN extrait du contenu dans le message (il s\'affiche sur un écran verrouillé)')
ok(message(L[0], DA) is None and message(L[0], DC) is None, 'un autre appareil ne peut PAS le déchiffrer (chiffré pour une seule clé)')
H = {x.split(':', 1)[0]: x.split(':', 1)[1].strip() for x in L[0]['headers'] if not x.startswith('Authorization')}
ok(H['Content-Encoding'] == 'aes128gcm' and H['Content-Type'] == 'application/octet-stream' and H['TTL'] == '86400' and H['Urgency'] == 'normal', 'en-têtes du service de push : aes128gcm, TTL 24 h, urgence normale')
v = ref({'mode': 'vapid', 'authorization': next(x for x in L[0]['headers'] if x.startswith('Authorization')).split(': ', 1)[1], 'public': PUB}); now = time.time()
ok(v['valid'] is True and v['k'] == PUB and v['header'] == {'typ': 'JWT', 'alg': 'ES256'} and v['claims']['aud'] == 'https://fcm.googleapis.com' and v['claims']['sub'] == 'mailto:contact@uluana.art' and now < v['claims']['exp'] <= now + 13 * 3600 and v['sigLen'] == 64, 'signature VAPID vérifiée par la référence : ES256 de 64 octets, audience = le service de push, sujet du serveur, expire dans ≤ 12 h')
raw = base64.b64decode(L[0]['body']); ok(len(raw) > 86 and int.from_bytes(raw[16:20], 'big') == 4096 and raw[20] == 65 and raw[21] == 4, 'structure du corps : sel (16) + taille d\'enregistrement 4096 + clé de 65 octets, puis le message chiffré')
tam = dict(L[0]); b = bytearray(raw); b[-1] ^= 1; tam['body'] = base64.b64encode(bytes(b)).decode(); ok(message(tam, DB) is None, 'un message ALTÉRÉ d\'un seul bit est rejeté par la référence (authentification GCM)')
clear_sink(); call('POST', 'p_a', f'/posts/{PB}/comments', {'body': 'Superbe !'}); L = wait_sink(1); m = message(L[0], DB) if L else None
ok(m and m['body'] == 'Anne A. a commenté votre publication' and m['url'] == f'/fil?post={PB}' and m['type'] == 'comment', 'commentaire sur sa publication : « Anne A. a commenté votre publication »')
clear_sink(); call('POST', 'p_a', f'/posts/{PB}/comments', {'body': 'Et encore !'}); ok(nothing(), 'une seconde notification non lue du même membre sur la même publication : doublon, AUCUN second push')
cA = int(sql(f"SELECT id FROM comments WHERE post_id = {PB} ORDER BY id LIMIT 1")); clear_sink(); call('POST', 'p_b', f'/posts/{PB}/comments', {'body': 'Merci', 'parent_id': cA}); L = wait_sink(1); m = message(L[0], DA) if L else None
ok(m and m['body'] == 'Bruno B. a répondu à votre commentaire' and L[0]['endpoint'] == EA, 'réponse à un commentaire : « a répondu à votre commentaire », vers l\'appareil d\'Anne')
sql("INSERT IGNORE INTO member_follows (follower_id, followed_id) VALUES ('p_a', 'p_b')"); clear_sink(); st, d, _ = call('POST', 'p_a', '/posts', {'body': 'Salut @Bruno B. !', 'theme_id': THEME, 'visibility': 'all', 'mentions': ['p_b']}); L = wait_sink(1); m = message(L[0], DB) if L else None
ok(st == 201 and m and m['body'] == 'Anne A. vous a mentionné dans une publication' and m['url'] == f"/fil?post={d['post']['id']}" and m['type'] == 'mention', 'mention dans une publication : « Anne A. vous a mentionné dans une publication »')
reset_rate_limits(); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink(); call('POST', 'p_a', f"/posts/{PB}/comments", {'body': 'Pour @Bruno B. aussi', 'mentions': ['p_b']}); L = wait_sink(1); m = message(L[0], DB) if L else None
ok(len(L) == 1 and m and m['body'] == 'Anne A. vous a mentionné dans un commentaire', 'mention dans un commentaire : UN seul push, et c\'est la mention (pas aussi « a commenté votre publication »)')
st, d, _ = call('GET', 'p_c', '/notification-prefs'); sql("DELETE FROM notifications WHERE user_id = 'p_c'"); PC = publish('p_c', 'Post de Chloé'); clear_sink(); call('POST', 'p_a', f'/posts/{PC}/comments', {'body': 'Hello'}); L = wait_sink(1); m = message(L[0], DC) if L else None
ok(m and m['body'] == 'Anne A. commented on your post' and m['title'] == 'Uluana', 'membre dont la langue est l\'ANGLAIS : « Anne A. commented on your post »')
sql("DELETE FROM push_subscriptions WHERE user_id = 'p_b'"); call('POST', 'p_b', '/push-subscriptions', sub_body(DB, EB)); DB2 = device(); EB2 = fcm(); call('POST', 'p_b', '/push-subscriptions', sub_body(DB2, EB2)); reset_rate_limits(); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink(); call('POST', 'p_c', f'/posts/{PB}/comments', {'body': 'Deux appareils'}); L = wait_sink(2)
ok(len(L) == 2 and {x['endpoint'] for x in L} == {EB, EB2} and all(message(x, DB if x['endpoint'] == EB else DB2) for x in L) and L[0]['body'] != L[1]['body'], 'deux appareils : UN envoi chacun, chacun lisible avec SA clé ; les deux corps sont différents (sel et clé éphémère distincts)')

section("Quand le push NE part PAS")
sql("DELETE FROM push_subscriptions WHERE user_id = 'p_b' AND endpoint = '" + EB2 + "'"); mk_user('p_d', 'client_essentiel', 'Dan', 'Dupont'); reset_rate_limits(); sql("DELETE FROM notifications WHERE user_id = 'p_b'")
clear_sink(); call('POST', 'p_b', f'/posts/{PB}/comments', {'body': 'Je me réponds'}); ok(nothing(), 'sa PROPRE action : aucun push')
call('POST', 'p_b', '/blocks', {'user_id': 'p_d', 'block': True}); clear_sink(); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'Je suis bloqué'}); ok(nothing(), 'un membre BLOQUÉ (liste noire) : aucun push'); call('POST', 'p_b', '/blocks', {'user_id': 'p_d', 'block': False})
sql("DELETE FROM notifications WHERE user_id = 'p_b'"); prefs('p_b', {'types': {'comment': {'push': False}}}); clear_sink(); n1 = nrows('p_b'); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'Push coupé'}); ok(nrows('p_b') == n1 + 1 and nothing(), 'push désactivé pour ce type : la notification est dans l\'application, aucun push'); prefs('p_b', {'types': {'comment': {'push': True}}})
sql("DELETE FROM notifications WHERE user_id = 'p_b'"); prefs('p_b', {'types': {'comment': {'in_app': False, 'push': True}}}); clear_sink(); n1 = nrows('p_b'); call('POST', 'p_c', f'/posts/{PB}/comments', {'body': 'Sans cloche'}); L = wait_sink(1)
ok(nrows('p_b') == n1 and len(L) == 1, 'cloche désactivée mais push activé : AUCUNE notification dans l\'application, le push part quand même')
prefs('p_b', {'types': {'comment': {'in_app': False, 'push': False}}}); clear_sink(); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'Rien du tout'}); ok(nothing(), 'cloche ET push désactivés : rien'); prefs('p_b', {'types': {'comment': {'in_app': True, 'push': True}}})
utc = datetime.datetime.now(datetime.timezone.utc); fmt = lambda d: d.strftime('%H:%M'); sql("DELETE FROM notifications WHERE user_id = 'p_b'")
prefs('p_b', {'quiet': {'enabled': True, 'start': fmt(utc - datetime.timedelta(minutes=30)), 'end': fmt(utc + datetime.timedelta(minutes=30)), 'timezone': 'UTC'}}); clear_sink(); n1 = nrows('p_b'); reset_rate_limits(); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'En pleine nuit'}); ok(nrows('p_b') == n1 + 1 and nothing(), 'PENDANT les heures calmes : la notification est dans l\'application, AUCUN push')
prefs('p_b', {'quiet': {'enabled': True, 'start': fmt(utc + datetime.timedelta(hours=2)), 'end': fmt(utc + datetime.timedelta(hours=3)), 'timezone': 'UTC'}}); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink(); call('POST', 'p_c', f'/posts/{PB}/comments', {'body': 'En journée'}); ok(len(wait_sink(1)) == 1, 'heures calmes configurées mais pas en cours : le push part'); prefs('p_b', {'quiet': {'enabled': False}})
sql("DELETE FROM push_subscriptions WHERE user_id = 'p_b'"); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink(); n1 = nrows('p_b'); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'Sans appareil'}); ok(nrows('p_b') == n1 + 1 and nothing(), 'aucun appareil abonné : la notification est dans l\'application, rien à envoyer, aucune erreur')
call('POST', 'p_b', '/push-subscriptions', sub_body(DB, EB))

section("Appareils disparus ou en échec")
EG, EF = 'https://fcm.googleapis.com/gone/abc', 'https://fcm.googleapis.com/fail/abc'; DG, DF = device(), device(); call('POST', 'p_b', '/push-subscriptions', sub_body(DG, EG)); call('POST', 'p_b', '/push-subscriptions', sub_body(DF, EF)); reset_rate_limits(); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink()
call('POST', 'p_c', f'/posts/{PB}/comments', {'body': 'Trois appareils'}); L = wait_sink(3); time.sleep(0.6)
ok(len(L) == 3 and n_subs('p_b') == 2 and sql(f"SELECT COUNT(*) FROM push_subscriptions WHERE endpoint = '{EG}'") == '0', 'un service qui répond 410 (appareil désinscrit) : l\'abonnement est SUPPRIMÉ aussitôt')
ok(sql(f"SELECT failures FROM push_subscriptions WHERE endpoint = '{EF}'") == '1' and sql(f"SELECT failures FROM push_subscriptions WHERE endpoint = '{EB}'") == '0' and sql(f"SELECT last_used_at IS NOT NULL FROM push_subscriptions WHERE endpoint = '{EB}'") == '1', 'un service qui répond 500 : l\'abonnement reste, avec 1 échec ; un envoi réussi remet à zéro et date le dernier envoi')
sql(f"UPDATE push_subscriptions SET failures = 4 WHERE endpoint = '{EF}'"); sql("DELETE FROM notifications WHERE user_id = 'p_b'"); clear_sink(); call('POST', 'p_d', f'/posts/{PB}/comments', {'body': 'Cinquième échec'}); wait_sink(2); time.sleep(0.6)
ok(sql(f"SELECT COUNT(*) FROM push_subscriptions WHERE endpoint = '{EF}'") == '0', 'au 5e échec consécutif : l\'abonnement est supprimé'); 

section("Message de test")
clear_sink(); st, d, _ = call('POST', 'p_b', '/push-subscriptions/test', {}); L = wait_sink(1); m = message(L[0], DB) if L else None
ok(st == 200 and d == {'sent': 1, 'removed': 0, 'failed': 0} and m == {'title': 'Uluana', 'body': 'Notification de test : tout fonctionne.', 'url': '/notifications', 'tag': 'uluana-test', 'type': 'test'}, 'message de test : { sent: 1 }, texte français, lisible par l\'appareil')
clear_sink(); st, d, _ = call('POST', 'p_c', '/push-subscriptions/test', {}); L = wait_sink(1); ok(st == 200 and message(L[0], DC)['body'] == 'Test notification: everything works.', 'en anglais pour un membre anglophone')
sql("DELETE FROM push_subscriptions WHERE user_id = 'p_a'"); st, d, _ = call('POST', 'p_a', '/push-subscriptions/test', {}); ok(st == 409 and 'appareil' in d['error'], 'sans appareil abonné : 409 « Aucun appareil abonné »')
call('POST', 'p_b', '/push-subscriptions', sub_body(DG, EG)); clear_sink(); st, d, _ = call('POST', 'p_b', '/push-subscriptions/test', {}); wait_sink(2); ok(st == 200 and d['sent'] == 1 and d['removed'] == 1, 'le test lui-même retire un appareil disparu : { sent: 1, removed: 1 }')

section("Conformité à la norme (RFC 8291) et limites")
rfc = php('$as = ["private" => wp_b64u_decode("yfWPiYE-n46HLnH0KqZOF1fJJU3MYrct3AELtAQ-oRw"), "public" => wp_b64u_decode("BP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A8")]; echo wp_b64u_encode(webpush_encrypt("When I grow up, I want to be a watermelon", wp_b64u_decode("BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4"), wp_b64u_decode("BTBZMqHH6r4Tts7J_aSIgg"), $as, wp_b64u_decode("DGv6ra1nlYgDCS1FRnbzlw")));')
ok(rfc == 'DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A_yl95bQpu6cVPTpK4Mqgkf1CXztLVBSt2Ks3oZwbuwXPXLWyouBWLVWGNWQexSgSxsj_Qulcy4a-fN', 'EXEMPLE OFFICIEL de la RFC 8291 (annexe A) : le corps chiffré est identique, octet pour octet, à celui de la norme')
d0 = device(); texts = ['Héhé « ça marche » — 日本語 🎉', 'x', 'a' * 2999]
res = []
for t in texts:
    b = php('echo base64_encode(webpush_encrypt(json_decode(' + json.dumps(json.dumps(t)) + '), wp_b64u_decode("' + d0['public'] + '"), wp_b64u_decode("' + d0['auth'] + '")));'); r = ref({'mode': 'decrypt', 'body': b, 'private': d0['private'], 'public': d0['public'], 'auth': d0['auth']}); res.append(r.get('plaintext') == t)
ok(all(res), 'unicode (accents, guillemets, japonais, émoji), 1 caractère et 2 999 octets : déchiffrés à l\'identique par la référence indépendante')
ok(php('try { webpush_encrypt(str_repeat("a", 3001), wp_b64u_decode("' + d0['public'] + '"), wp_b64u_decode("' + d0['auth'] + '")); echo "pas d erreur"; } catch (InvalidArgumentException $e) { echo "refuse"; }') == 'refuse', 'un message de plus de 3 000 octets est refusé (les services en acceptent 4 078)')
ok(php('try { webpush_encrypt("x", "short", "short"); echo "pas d erreur"; } catch (InvalidArgumentException $e) { echo "refuse"; }') == 'refuse', 'clés d\'abonné invalides : refusées avant tout chiffrement')
bodies = {php('echo base64_encode(webpush_encrypt("même texte", wp_b64u_decode("' + d0['public'] + '"), wp_b64u_decode("' + d0['auth'] + '")));') for _ in range(3)}; ok(len(bodies) == 3, 'le même texte chiffré trois fois donne trois corps différents (sel et clé éphémère aléatoires)')
ok(php('echo webpush_send(["endpoint" => "https://evil.example/x", "p256dh" => "a", "auth" => "b"], "{}")["status"];') == '0' and php('echo json_encode(webpush_send(["endpoint" => "https://evil.example/x", "p256dh" => "a", "auth" => "b"], "{}")["gone"]);') == 'true', 'webpush_send refuse AUSSI une adresse détournée (défense en profondeur : une adresse déjà en base ne sera jamais appelée) et la signale à supprimer')
ok(php('echo json_encode(push_text("fr", "collaboration", "reply", "Anne A."));') == '{"title":"Uluana","body":"Anne A. a r\\u00e9pondu \\u00e0 votre projet"}' and json.loads(php('echo json_encode(push_text("en", "message", "", "Bob B."), JSON_UNESCAPED_UNICODE);'))['body'] == 'Bob B. sent you a message', 'textes de la collaboration et des messages (lot 2) prêts, fr et en')
ok(json.loads(php('echo json_encode(push_text("fr", "like", "", ""), JSON_UNESCAPED_UNICODE);'))['body'] == 'Nouvelle notification' or json.loads(php('echo json_encode(push_text("fr", "inconnu", "x", ""), JSON_UNESCAPED_UNICODE);'))['body'] == 'Nouvelle notification', 'type ou sorte inconnus : texte générique, jamais d\'erreur')
ok(php('echo push_url("collaboration", "project", 12, ["kind" => "reply"]);') == '/collaboration?project=12&mine=1' and php('echo push_url("message", "conversation", 3, []);') == '/messagerie?conversation=3' and php('echo push_url("message", "autre", 3, []);') == '/notifications' and php('echo push_url("like", "comment", 55, ["post_id" => 31]);') == '/fil?post=31', 'adresses : projet de collaboration, message → sa conversation, notifications par défaut, « j\'aime » d\'un commentaire → sa publication')

wipe(); sql("DELETE FROM users WHERE id LIKE 'p\\_%'"); clear_sink(); summary()