#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
# Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
"""Test de tools/etm-powersync-schema-check et du contrat OpenRPC (lot « contrat OpenRPC »).

    python3 -I tests/tools/etm-powersync-schema-check-test.py

Trois choses :
1. ce que le MOTEUR émet est conforme : les fichiers de référence du sérialiseur `GetPlan`
   (tests/auto/loadmodel/data/optimizer/, eux-mêmes épinglés au sérialiseur par la suite
   `loadmodel`) et les traces RÉELLES du moteur (tests/tools/data/) ;
2. l'OPTIMISEUR DE RÉFÉRENCE, mode par mode, sur une vraie session socket capturée : les modes
   « bons » sont valides, chaque mode « mauvais » est rejeté par le schéma OU par une règle
   sémantique nommée — et laquelle (`TABLE_MODES`, recopiée au rapport) ;
3. le validateur lui-même ne peut pas se taire : un mot-clé non géré, une règle orpheline, un
   champ ajouté, un mauvais type le font échouer ;
4. `error.data` du -32602 de GetPlan (protocole 1.3) : forme exigée d'un optimiseur 1.3, avertie
   d'un optimiseur 1.2, recopie d'une valeur reçue refusée des deux.
"""

import copy
import glob
import json
import os
import select
import socket
import subprocess
import sys
import tempfile
import time
import unittest

ICI = os.path.dirname(os.path.abspath(__file__))
RACINE = os.path.join(ICI, '..', '..')
OUTIL = os.path.join(RACINE, 'tools', 'etm-powersync-schema-check')
REF = os.path.join(RACINE, 'tools', 'etm-powersync-ref-optimizer', 'etm-powersync-ref-optimizer')
SCHEMA = os.path.join(RACINE, 'docs', 'etm-powersync-optimizer-protocol.openrpc.json')
GOLDEN = sorted(glob.glob(os.path.join(RACINE, 'tests', 'auto', 'loadmodel', 'data', 'optimizer', '*.json')))
TRACES = sorted(glob.glob(os.path.join(ICI, 'data', '*.ndjson')))
REQUETE = os.path.join(RACINE, 'tests', 'auto', 'loadmodel', 'data', 'optimizer', 'site-pv-ecs-borne-sgready.json')
TOKEN = 'jeton-de-test-0123456789'
with open(SCHEMA, encoding='utf-8') as _f:
    # La version du MOTEUR, celle que le contrat impose à ses Handshake (const) — jamais recopiée.
    VERSION_MOTEUR = json.load(_f)['components']['schemas']['ProtocolVersion']['const']

# mode → (qui rejette, verdict sémantique attendu). « schéma:<mot-clé> » : écart de forme ;
# « règle » : la forme passe, une règle x-etm-semantic-rules refuse ; « — » : rien ne rejette.
# Le verdict est le code que le MOTEUR publie (fallbackCode / lastErrorCode, suite refoptimizer).
TABLE_MODES = {
    'normal':         ('—', 'valide'),
    'abstain':        ('—', 'abstain'),
    'slow':           ('—', 'valide'),
    'protocol-1.0':   ('—', 'valide'),
    'deviate':        ('—', 'valide'),
    'echo-token':     ('—', 'valide'),
    'wrong-type':     ('schéma:type', 'valide'),
    'invalid':        ('schéma:type', 'malformed-plan'),
    'expired':        ('règle', 'expired'),
    'wrong-major':    ('règle', 'major-mismatch'),
    'drop':           ('transport', None),
    'unknown-load':   ('règle', 'unknown-load'),
    'forbidden-kind': ('règle', 'kind-mismatch'),
    'missing-load':   ('règle', 'incomplete-slot'),
    # Modes 1.3 : sur un contexte VALIDE et le bon token, ils se comportent comme `normal`.
    'old-data':       ('—', 'valide'),
    'echo-value':     ('—', 'valide'),
    'major-first':    ('—', 'valide'),
}

# Une valeur que nul chemin conforme ne peut contenir (tiret, deux-points) : si elle apparaît dans
# error.data, c'est une recopie.
MARQUEUR = 'valeur-sonde:7Qz9'


def verifier(*args):
    p = subprocess.run([sys.executable, '-I', OUTIL, *args], capture_output=True, text=True, timeout=60)
    resume = None
    for ligne in p.stdout.splitlines():
        if ligne.startswith('RÉSUMÉ '):
            resume = json.loads(ligne[len('RÉSUMÉ '):])
    return p.returncode, resume, p.stdout + p.stderr


def lire_ligne(sock, tampon, delai=5.0):
    fin = time.monotonic() + delai
    while b'\n' not in tampon[0]:
        reste = fin - time.monotonic()
        if reste <= 0 or not select.select([sock], [], [], reste)[0]:
            return None
        bloc = sock.recv(65536)
        if not bloc:
            return None
        tampon[0] += bloc
    ligne, tampon[0] = tampon[0].split(b'\n', 1)
    return json.loads(ligne)


def session(mode, dossier):
    """Une vraie session socket avec l'optimiseur de référence ; rend le fichier capturé."""
    sock_path = os.path.join(dossier, 'opt.sock')
    capture = os.path.join(dossier, f'{mode}.ndjson')
    jeton = os.path.join(dossier, 'jeton')
    with open(jeton, 'w') as f:
        f.write(TOKEN)
    proc = subprocess.Popen([sys.executable, '-I', '-B', REF, '--listen', f'unix://{sock_path}', '--mode', mode,
                             '--capture', capture, '--slow-s', '0.2', '--token-file', jeton],
                            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
    try:
        assert proc.stdout.readline().startswith('ÉCOUTE'), 'optimiseur de référence muet'
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.connect(sock_path)
        tampon = [b'']
        with open(REQUETE, encoding='utf-8') as f:
            params = json.load(f)
        s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'Handshake',
                              'params': {'protocolVersion': VERSION_MOTEUR, 'engine': 'etm-powersync-energy-plugin-etm/test',
                                         'token': TOKEN}}).encode() + b'\n')
        hs = lire_ligne(s, tampon)
        if hs and str(hs.get('result', {}).get('protocolVersion', '')).startswith('1'):
            s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 2, 'method': 'GetPlan', 'params': params}).encode() + b'\n')
            lire_ligne(s, tampon)
        s.close()
    finally:
        proc.terminate()
        proc.wait(5)
        proc.stdout.close()
    return capture


def session_contexte_invalide(dossier, mode, annonce, params):
    """Handshake accepté puis UN GetPlan de contexte invalide ; rend le fichier capturé."""
    sock_path = os.path.join(dossier, 'opt.sock')
    capture = os.path.join(dossier, f'invalide-{mode}-{annonce}.ndjson')
    proc = subprocess.Popen([sys.executable, '-I', '-B', REF, '--listen', f'unix://{sock_path}', '--mode', mode,
                             '--announce', annonce, '--capture', capture],
                            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
    try:
        assert proc.stdout.readline().startswith('ÉCOUTE'), 'optimiseur de référence muet'
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.connect(sock_path)
        tampon = [b'']
        s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'Handshake',
                              'params': {'protocolVersion': VERSION_MOTEUR,
                                         'engine': 'etm-powersync-energy-plugin-etm/test'}}).encode() + b'\n')
        assert lire_ligne(s, tampon)['result']['protocolVersion'] == annonce
        s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 2, 'method': 'GetPlan', 'params': params}).encode() + b'\n')
        reponse = lire_ligne(s, tampon)
        assert reponse['error']['code'] == -32602, reponse
        s.close()
    finally:
        proc.terminate()
        proc.wait(5)
        proc.stdout.close()
    return capture


class TestCeQueLeMoteurEmet(unittest.TestCase):
    def test_les_references_du_serialiseur_sont_conformes(self):
        self.assertGreaterEqual(len(GOLDEN), 4, 'fichiers de référence du sérialiseur introuvables')
        for f in GOLDEN:
            code, r, sortie = verifier('--method', 'GetPlan', '--params', f)
            self.assertEqual(code, 0, f'{os.path.basename(f)} :\n{sortie}')
            self.assertEqual(r['comptes'], {'GetPlan.params': 1})

    def test_les_traces_reelles_du_moteur_sont_conformes(self):
        self.assertTrue(TRACES, 'aucune trace réelle sous tests/tools/data')
        requetes = reponses = 0
        for f in TRACES:
            code, r, sortie = verifier('--trace', f, '--semantic')
            self.assertEqual(code, 0, f'{os.path.basename(f)} :\n{sortie}')
            requetes += r['comptes'].get('GetPlan.params', 0)
            reponses += r['comptes'].get('GetPlan.result', 0)
        # Le contrôle ne doit pas pouvoir se vider : des requêtes ET des réponses d'optimiseur jugées.
        self.assertGreater(requetes, 0)
        self.assertGreater(reponses, 0)


class TestOptimiseurDeReference(unittest.TestCase):
    def test_chaque_mode_et_qui_le_rejette(self):
        with tempfile.TemporaryDirectory() as d:
            obtenu = {}
            for mode in TABLE_MODES:
                capture = session(mode, d)
                code, r, sortie = verifier('--ndjson', capture, '--semantic')
                with open(capture, encoding='utf-8') as f:
                    self.assertNotIn(TOKEN, f.read(), 'le token est écrit en clair dans la capture')
                motifs = sorted({l.rsplit('[', 1)[1].rstrip(']') for l in sortie.splitlines()
                                 if l.startswith('ligne ') and l.endswith(']')})
                verdicts = {k: v for k, v in r['verdicts'].items() if k != 'valide'} if r else {}
                if not r['comptes'].get('GetPlan.result') and not verdicts and mode != 'wrong-major':
                    qui, verdict = 'transport', None
                else:
                    verdict = next(iter(verdicts), 'valide')
                    # `expired` ne compte pas comme un échec (§7), mais le plan n'est pas appliqué :
                    # c'est bien une règle qui l'écarte. `abstain` est une réponse permise.
                    qui = ('schéma:' + ','.join(motifs)) if motifs else \
                        ('règle' if verdict not in ('valide', 'abstain') else '—')
                obtenu[mode] = (qui, verdict)
                print(f'  {mode:15s} → {qui:12s} {verdict}', file=sys.stderr)
            self.assertEqual(obtenu, TABLE_MODES)


class TestLeValidateurNeSeTaitPas(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.TemporaryDirectory()
        with open(SCHEMA, encoding='utf-8') as f:
            self.schema = json.load(f)
        with open(REQUETE, encoding='utf-8') as f:
            self.requete = json.load(f)

    def tearDown(self):
        self.d.cleanup()

    def ecrire(self, nom, contenu):
        chemin = os.path.join(self.d.name, nom)
        with open(chemin, 'w', encoding='utf-8') as f:
            json.dump(contenu, f, ensure_ascii=False)
        return chemin

    def test_un_mot_cle_non_gere_est_une_erreur_de_schema(self):
        s = copy.deepcopy(self.schema)
        s['components']['schemas']['Meter']['uniqueItems'] = True
        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan',
                                   '--params', REQUETE)
        self.assertEqual(code, 2, sortie)
        self.assertIn('« uniqueItems » non géré', sortie)

    def test_une_regle_semantique_sans_implantation_est_une_erreur(self):
        s = copy.deepcopy(self.schema)
        s['x-etm-semantic-rules']['rules'].append(dict(s['x-etm-semantic-rules']['rules'][0], name='nouvelle-regle'))
        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan', '--params', REQUETE)
        self.assertEqual(code, 2, sortie)
        self.assertIn('nouvelle-regle', sortie)
        s = copy.deepcopy(self.schema)
        s['x-etm-semantic-rules']['rules'] = [r for r in s['x-etm-semantic-rules']['rules'] if r['name'] != 'unknown-load']
        code, _, sortie = verifier('--schema', self.ecrire('s.json', s), '--method', 'GetPlan', '--params', REQUETE)
        self.assertEqual(code, 2, sortie)

    def test_un_champ_ajoute_par_le_moteur_est_refuse(self):
        q = copy.deepcopy(self.requete)
        q['loads'][2]['telemetry']['champNouveau'] = 1
        code, r, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
        self.assertEqual(code, 1, sortie)
        self.assertIn('/loads/2/telemetry/champNouveau', sortie)
        q = copy.deepcopy(self.requete)
        q['nouveauParam'] = True
        code, _, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
        self.assertEqual(code, 1, sortie)

    def test_un_mauvais_type_est_refuse(self):
        q = copy.deepcopy(self.requete)
        q['loads'][0]['priority'] = 1.5
        code, _, sortie = verifier('--method', 'GetPlan', '--params', self.ecrire('q.json', q))
        self.assertEqual(code, 1, sortie)
        self.assertIn('[type]', sortie)

    def test_une_cle_de_motif_non_documentee_est_refusee(self):
        plan = {'slots': [{'from': '2026-10-12T13:42:00+02:00', 'to': '2026-10-12T13:47:00+02:00', 'actions': [
            {'loadId': lc['id'], 'kind': 'state' if lc['adapter'] == 'sg-ready' else 'setpoint',
             **({'state': 2} if lc['adapter'] == 'sg-ready' else {'powerW': 0}),
             'reason': 'test', 'reasonCode': 'SURPLUS_SETPOINT', 'reasonParams': {'budgetW': 1}}
            for lc in self.requete['loads']]}]}
        code, r, sortie = verifier('--method', 'GetPlan', '--result', self.ecrire('p.json', plan),
                                   '--request', REQUETE, '--semantic')
        self.assertEqual(code, 0, sortie)
        self.assertEqual(r['verdicts'], {'valide': 1})
        plan['slots'][0]['actions'][0]['reasonParams']['texteLibre'] = 'x'
        code, _, sortie = verifier('--method', 'GetPlan', '--result', self.ecrire('p.json', plan))
        self.assertEqual(code, 1, sortie)
        self.assertIn('[x-etm-reason-params]', sortie)

    def test_le_sous_ensemble_se_dit(self):
        code, _, sortie = verifier('--subset')
        self.assertEqual(code, 0)
        self.assertIn('oneOf', sortie)
        self.assertNotIn('uniqueItems', sortie)


class TestDonneesDErreur(unittest.TestCase):
    """error.data du -32602 de GetPlan (protocole 1.3), sur de vraies sessions de l'optimiseur de
    référence. La session porte un GetPlan de contexte invalide : le validateur la juge SANS
    --semantic (sinon la réponse error est un refus, quel que soit son data)."""

    # (mode, version annoncée) → (code de sortie, écarts > 0, avertissements > 0)
    ATTENDU = {
        ('normal', '1.3'):     (0, False, False),
        ('normal', '1.2'):     (0, False, False),
        ('old-data', '1.3'):   (1, True, False),    # forme d'avant 1.3, annoncée 1.3 : refusée
        ('old-data', '1.2'):   (0, False, True),    # la même, annoncée 1.2 : avertie
        ('echo-value', '1.3'): (1, True, False),
        ('echo-value', '1.2'): (1, True, True),     # forme avertie, RECOPIE refusée (règle de 1.2)
    }

    def test_forme_exigee_en_1_3_avertie_en_1_2_recopie_refusee(self):
        params = copy.deepcopy(json.load(open(REQUETE, encoding='utf-8')))
        params['timestamp'] = MARQUEUR      # sans décalage, illisible : `format`
        obtenu = {}
        with tempfile.TemporaryDirectory() as d:
            for (mode, annonce) in self.ATTENDU:
                capture = session_contexte_invalide(d, mode, annonce, params)
                code, r, sortie = verifier('--ndjson', capture, '--responses-only')
                self.assertEqual(r['comptes'].get('GetPlan.error'), 1, sortie)
                self.assertNotIn('GetPlan.params', r['comptes'])
                self.assertEqual(r['comptes'].get('GetPlan.error.data'), 1, sortie)
                obtenu[(mode, annonce)] = (code, r['ecarts'] > 0, r['avertissements'] > 0)
                print(f'  {mode:11s} {annonce} → code {code}, écarts {r["ecarts"]}, avertissements '
                      f'{r["avertissements"]}', file=sys.stderr)
                if mode == 'echo-value':
                    self.assertIn('recopie la valeur reçue en params.timestamp', sortie)
                    self.assertNotIn(MARQUEUR, sortie, 'le validateur recopie lui-même la valeur')
                if (mode, annonce) == ('old-data', '1.2'):
                    self.assertIn('AVERTISSEMENT', sortie)
                    self.assertIn('à aligner avant de s\'annoncer 1.3', sortie)
        self.assertEqual(obtenu, self.ATTENDU)

    def test_la_forme_vient_du_contrat(self):
        # Contre-épreuve : sans x-etm-data-schema, la forme d'avant 1.3 annoncée 1.3 passe — c'est
        # bien le contrat qui juge, pas une règle cachée dans l'outil.
        with open(SCHEMA, encoding='utf-8') as f:
            s = json.load(f)
        for e in next(m for m in s['methods'] if m['name'] == 'GetPlan')['errors']:
            e.pop('x-etm-data-schema', None)
            e.pop('x-etm-data-required-from', None)
        with tempfile.TemporaryDirectory() as d:
            schema = os.path.join(d, 's.json')
            with open(schema, 'w', encoding='utf-8') as f:
                json.dump(s, f, ensure_ascii=False)
            capture = session_contexte_invalide(d, 'old-data', '1.3', {'timestamp': MARQUEUR})
            self.assertEqual(verifier('--ndjson', capture, '--responses-only')[0], 1)
            code, r, sortie = verifier('--schema', schema, '--ndjson', capture, '--responses-only')
            self.assertEqual(code, 0, sortie)
            self.assertNotIn('GetPlan.error.data', r['comptes'])

    def ecrire(self, d, nom, contenu):
        chemin = os.path.join(d, nom)
        with open(chemin, 'w', encoding='utf-8') as f:
            json.dump(contenu, f, ensure_ascii=False)
        return chemin

    def juger(self, d, data, requete=None, version='1.3'):
        args = ['--method', 'GetPlan', '--error',
                self.ecrire(d, 'e.json', {'code': -32602, 'message': 'contexte invalide', 'data': data}),
                '--handshake', self.ecrire(d, 'h.json', {'protocolVersion': version})]
        if requete is not None:
            args += ['--request', self.ecrire(d, 'q.json', requete)]
        return verifier(*args)

    def test_la_forme_ligne_a_ligne(self):
        p = {'path': 'params.loads[0].id', 'problem': 'missing'}
        cas = [
            ({'problems': [p]}, 0, None),
            ({'problems': [{'path': 'params', 'problem': 'type'}]}, 0, None),
            ({'problems': [{'path': 'params.loads[0].declared.estimatedPowerW.3', 'problem': 'range'}]}, 0, None),
            ({'problems': [p] * 50, 'truncated': True}, 0, None),
            ({'problems': []}, 1, '[minItems]'),
            ({'problems': [p] * 51}, 1, '[maxItems]'),
            ({'problems': [p] * 50, 'truncated': False}, 1, '[const]'),
            ({'problems': [p] * 3, 'truncated': True}, 1, '[x-etm-truncated]'),
            ({'problems': [dict(p, problem='absent')]}, 1, '[enum]'),
            ({'problems': [dict(p, path='/loads/0/id')]}, 1, '[pattern]'),
            ({'problems': [dict(p, path='loads[0].id')]}, 1, '[pattern]'),
            ({'problems': [dict(p, path='params.loads.0.id ')]}, 1, '[pattern]'),
            ({'problems': [dict(p, value=3)]}, 1, '[additionalProperties]'),
            ({'problems': [p], 'detail': 'x'}, 1, '[additionalProperties]'),
            ({'problems': [{'path': 'params.timestamp'}]}, 1, '[required]'),
            ({'problemes': ['params.timestamp : format']}, 1, '[required]'),
            ('contexte invalide', 1, '[type]'),
        ]
        with tempfile.TemporaryDirectory() as d:
            for data, code_attendu, mot in cas:
                code, r, sortie = self.juger(d, data)
                self.assertEqual(code, code_attendu, f'{data!r}\n{sortie}')
                if mot:
                    self.assertIn(mot, sortie, data)
            # data absent : écart en 1.3
            chemin = self.ecrire(d, 'e.json', {'code': -32602, 'message': 'contexte invalide'})
            code, _, sortie = verifier('--method', 'GetPlan', '--error', chemin)
            self.assertEqual(code, 1, sortie)
            self.assertIn('« data » manquant', sortie)
            # Un autre code d'erreur n'a pas de forme normée : rien à juger.
            chemin = self.ecrire(d, 'e.json', {'code': -32603, 'message': 'erreur interne', 'data': [1, 'x']})
            code, r, sortie = verifier('--method', 'GetPlan', '--error', chemin)
            self.assertEqual((code, r['ecarts']), (0, 0), sortie)

    def test_la_recopie_ne_peut_pas_accuser_un_chemin_conforme(self):
        # Valeurs de la requête qui sont AUSSI des noms de champ (domain « battery », trigger
        # « forecast », meter.source « meter ») : un chemin conforme les contient, ce n'est pas une
        # recopie. Une valeur à tiret ou à deux-points recopiée, elle, l'est.
        requete = copy.deepcopy(json.load(open(REQUETE, encoding='utf-8')))
        requete['trigger'] = 'forecast'
        requete['loads'][0]['domain'] = 'battery'
        data = {'problems': [{'path': 'params.battery.socPercent', 'problem': 'range'},
                             {'path': 'params.forecast.pv[3].from', 'problem': 'format'},
                             {'path': 'params.meter.source', 'problem': 'enum'}]}
        with tempfile.TemporaryDirectory() as d:
            code, r, sortie = self.juger(d, data, requete)
            self.assertEqual(code, 0, sortie)
            ident = requete['loads'][0]['id']
            self.assertRegex(ident, '[^A-Za-z0-9_]')
            code, r, sortie = self.juger(d, {'problemes': [f'charge {ident} : priorité hors bornes']}, requete, '1.2')
            self.assertEqual(code, 1, sortie)
            self.assertIn('recopie la valeur reçue en params.loads[0].id', sortie)
            self.assertNotIn(ident, sortie.split('RÉSUMÉ')[0])
            # Avant 1.2, la règle n'existait pas : avertie seulement.
            code, r, sortie = self.juger(d, {'problemes': [f'charge {ident}']}, requete, '1.1')
            self.assertEqual((code, r['ecarts'], r['avertissements'] > 0), (0, 0, True), sortie)


if __name__ == '__main__':
    unittest.main(verbosity=2)