09043fb7c7
Contrat (docs/etm-powersync-optimizer-protocol.openrpc.json) :
- composants InvalidContextData, ContextProblem, ContextProblemCode : error.data du -32602 de
GetPlan = {"problems":[{path, problem}]}, 1 à 50 problèmes, truncated au-delà, chemin depuis
params (motif), nature dans une liste fermée (missing, type, range, format, enum, constraint),
formes fermées : aucune valeur reçue ne peut s'y glisser ;
- x-etm-data-schema et x-etm-data-required-from (1.3) sur cette erreur ;
- Handshake : token faux et autre majeure dans le même handshake → -32001, le token d'abord ;
- tolérance : error.data jamais lu par le moteur.
Prose (docs/OPTIMIZER_PROTOCOL.md) : v1.3, changelog §0 avec la compatibilité 1.2 ↔ 1.3, §2
(un lot reçu par le moteur reste sans réponse, à dessein), §3 (le token d'abord), §7 (forme
d'error.data, et ce que le moteur en fait : rien), annexes A, D, E, G. Blocs générés régénérés.
Générateur : un `|` dans une cellule (alternative d'un motif) est échappé ; une forme de
l'optimiseur fermée à dessein dit pourquoi (x-etm-closed-why) au lieu de « émise par le moteur ».
Test ajouté, vérifié échouant avec l'ancien générateur.
Moteur : version annoncée au handshake 1.3 (OptimizerProtocol::ProtocolVersion,
OptimizerClient::kProtocolVersion). Rien d'autre ne change dans le client. Tests qui citent la
version mis à jour (optimizerclient, socketscheduler, test du validateur).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
244 lines
12 KiB
Python
244 lines
12 KiB
Python
#!/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'
|
|
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'),
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
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)
|