ETM-Schurig/etm-powersync-energy-plugin-etm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
etm-powersync-energy-plugin.../tests/tools/etm-powersync-schema-check-test.py
Patrick Schurig fb581218c8 feat(outils): etm-powersync-schema-check, validateur du contrat OpenRPC 
Python, bibliothèque standard seule. Sous-ensemble de JSON Schema déclaré ;
un mot-clé non géré est une erreur de schéma, jamais ignoré. Valide un
message, une session NDJSON ou des traces ; --semantic applique
x-etm-semantic-rules dans l'ordre du validateur du moteur (une règle sans
implantation, ou l'inverse, est une erreur).

Test : références du sérialiseur et traces réelles du moteur conformes ;
optimiseur de référence mode par mode, et qui rejette chaque mode mauvais.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 21:15:51 +02:00

241 lines
12 KiB
Python
Raw Blame History

#!/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)
Reference in New Issue View Git Blame Copy Permalink