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>
152 lines
7.1 KiB
Python
152 lines
7.1 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-gen-protocol-doc (lot « contrat OpenRPC »).
|
|
|
|
python3 -I tests/tools/etm-powersync-gen-protocol-doc-test.py
|
|
|
|
Le dépôt est à jour ; une prose non régénérée, un bloc édité à la main, une liste du code changée
|
|
font échouer `--check` ; les balises cassées et un dump du code incomplet sont des erreurs.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ICI = os.path.dirname(os.path.abspath(__file__))
|
|
RACINE = os.path.join(ICI, '..', '..')
|
|
OUTIL = os.path.join(RACINE, 'tools', 'etm-powersync-gen-protocol-doc')
|
|
SCHEMA = os.path.join(RACINE, 'docs', 'etm-powersync-optimizer-protocol.openrpc.json')
|
|
PROSE = os.path.join(RACINE, 'docs', 'OPTIMIZER_PROTOCOL.md')
|
|
|
|
|
|
def lancer(*args):
|
|
p = subprocess.run([sys.executable, '-I', OUTIL, *args], capture_output=True, text=True, timeout=60)
|
|
return p.returncode, p.stdout + p.stderr
|
|
|
|
|
|
class TestGenProtocolDoc(unittest.TestCase):
|
|
def setUp(self):
|
|
self.d = tempfile.mkdtemp()
|
|
self.schema = os.path.join(self.d, 's.json')
|
|
self.prose = os.path.join(self.d, 'p.md')
|
|
shutil.copy(SCHEMA, self.schema)
|
|
shutil.copy(PROSE, self.prose)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.d)
|
|
|
|
def args(self, *x):
|
|
return ('--schema', self.schema, '--prose', self.prose) + x
|
|
|
|
def lire(self, f):
|
|
with open(f, encoding='utf-8') as h:
|
|
return h.read()
|
|
|
|
def ecrire(self, f, texte):
|
|
with open(f, 'w', encoding='utf-8') as h:
|
|
h.write(texte)
|
|
|
|
def test_le_depot_est_a_jour(self):
|
|
code, sortie = lancer('--check')
|
|
self.assertEqual(code, 0, sortie)
|
|
self.assertRegex(sortie, r'✓ contrat canonique, [0-9]+ bloc')
|
|
|
|
def test_un_contrat_change_sans_regenerer_la_prose_echoue(self):
|
|
doc = json.loads(self.lire(self.schema))
|
|
doc['components']['schemas']['Meter']['properties']['exportW']['description'] = 'changée'
|
|
self.ecrire(self.schema, json.dumps(doc, ensure_ascii=False, indent=2) + '\n')
|
|
code, sortie = lancer(*self.args('--check'))
|
|
self.assertEqual(code, 1, sortie)
|
|
self.assertIn('blocs générés pas à jour', sortie)
|
|
code, _ = lancer(*self.args())
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('| changée |', self.lire(self.prose))
|
|
self.assertEqual(lancer(*self.args('--check'))[0], 0)
|
|
|
|
def test_un_bloc_edite_a_la_main_echoue(self):
|
|
p = self.lire(self.prose).replace('| `timestamp` | string | oui |', '| `timestamp` | string | non |', 1)
|
|
self.ecrire(self.prose, p)
|
|
self.assertEqual(lancer(*self.args('--check'))[0], 1)
|
|
|
|
def test_un_contrat_non_canonique_echoue(self):
|
|
self.ecrire(self.schema, json.dumps(json.loads(self.lire(self.schema)), ensure_ascii=False) + '\n')
|
|
code, sortie = lancer(*self.args('--check'))
|
|
self.assertEqual(code, 1, sortie)
|
|
self.assertIn('pas à jour (forme canonique', sortie)
|
|
|
|
def test_une_liste_du_code_changee_se_regenere(self):
|
|
doc = json.loads(self.lire(self.schema))
|
|
listes = {}
|
|
for it in doc['x-etm-generated']['items']:
|
|
noeud = doc
|
|
for m in it['pointer'][1:].split('/'):
|
|
noeud = noeud[m]
|
|
listes[it['from']] = noeud
|
|
listes['DecisionCode::optimizerEmittableCodes()'] = listes['DecisionCode::optimizerEmittableCodes()'] + ['NOUVEAU_CODE']
|
|
dump = os.path.join(self.d, 'listes.json')
|
|
self.ecrire(dump, json.dumps(listes))
|
|
code, sortie = lancer(*self.args('--enums-from', dump, '--check'))
|
|
self.assertEqual(code, 1, sortie) # le contrat commité n'a pas la liste du code
|
|
self.assertEqual(lancer(*self.args('--enums-from', dump))[0], 0)
|
|
self.assertIn('NOUVEAU_CODE', self.lire(self.schema))
|
|
self.assertIn('`NOUVEAU_CODE`', self.lire(self.prose)) # la table des codes suit
|
|
del listes['TariffConfig::includes()']
|
|
self.ecrire(dump, json.dumps(listes))
|
|
code, sortie = lancer(*self.args('--enums-from', dump))
|
|
self.assertEqual(code, 2, sortie)
|
|
listes['TariffConfig::includes()'] = ['energy']
|
|
listes['Liste::orpheline()'] = []
|
|
self.ecrire(dump, json.dumps(listes))
|
|
code, sortie = lancer(*self.args('--enums-from', dump))
|
|
self.assertEqual(code, 2, sortie)
|
|
self.assertIn('orpheline', sortie)
|
|
|
|
def test_le_tableau_des_erreurs_suit_le_contrat(self):
|
|
# Protocole 1.2 : methods[].errors puis x-etm-message-errors, et ce que fait le moteur.
|
|
p = self.lire(self.prose)
|
|
self.assertIn('| `-32602` | Handshake | majeure incompatible | fermée |', p)
|
|
self.assertIn('| `-32600` | GetPlan | handshake requis | conservée |', p)
|
|
self.assertIn('| `-32600` | toute | lot interdit | conservée |', p)
|
|
self.assertIn('- **GetPlan** : Toute réponse error à GetPlan', p)
|
|
doc = json.loads(self.lire(self.schema))
|
|
doc['x-etm-message-errors']['items'].append({'code': -32099, 'message': 'erreur ajoutée',
|
|
'data': 'contre-épreuve', 'x-etm-connection': 'conservée'})
|
|
self.ecrire(self.schema, json.dumps(doc, ensure_ascii=False, indent=2) + '\n')
|
|
code, sortie = lancer(*self.args('--check'))
|
|
self.assertEqual(code, 1, sortie)
|
|
self.assertEqual(lancer(*self.args())[0], 0)
|
|
self.assertIn('| `-32099` | toute | erreur ajoutée | conservée | contre-épreuve |', self.lire(self.prose))
|
|
|
|
def test_la_forme_d_error_data_se_lit_dans_sa_table(self):
|
|
# Protocole 1.3 : le motif du chemin porte une alternative — un `|` nu fermerait la colonne,
|
|
# et la ligne aurait six cellules au lieu de cinq. Une forme de l'OPTIMISEUR fermée à dessein
|
|
# ne se dit pas « émise par le moteur ».
|
|
p = self.lire(self.prose)
|
|
ligne = next(l for l in p.splitlines() if l.startswith('| `path` | string |'))
|
|
self.assertEqual(len(ligne.replace('\\|', '').split('|')), 7, ligne)
|
|
self.assertIn('motif `^params(\\.[A-Za-z0-9_]+\\|\\[[0-9]+\\])*$`', ligne)
|
|
debut = p.index('<!-- BEGIN GENERATED table:InvalidContextData -->')
|
|
bloc = p[debut:p.index('<!-- END GENERATED -->', debut)]
|
|
self.assertIn('**refusés** (forme de l\'optimiseur fermée à dessein', bloc)
|
|
self.assertNotIn('émise par le moteur', bloc)
|
|
|
|
def test_des_balises_cassees_sont_une_erreur(self):
|
|
p = self.lire(self.prose)
|
|
i = p.index('<!-- END GENERATED -->')
|
|
self.ecrire(self.prose, p[:i] + p[i + len('<!-- END GENERATED -->'):])
|
|
self.assertEqual(lancer(*self.args('--check'))[0], 2)
|
|
self.ecrire(self.prose, p.replace('BEGIN GENERATED rules', 'BEGIN GENERATED inconnu', 1))
|
|
code, sortie = lancer(*self.args('--check'))
|
|
self.assertEqual(code, 2, sortie)
|
|
self.ecrire(self.prose, 'aucune balise\n')
|
|
self.assertEqual(lancer(*self.args('--check'))[0], 2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|