#!/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.
"""

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'

# 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'),
}


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': '1.1', '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


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)


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