ead662e044
Tableaux de messages (champs, types, obligatoire, valeurs, sens), règles sémantiques et codes de motif, entre balises BEGIN/END GENERATED de OPTIMIZER_PROTOCOL.md. --check (code 1 si la prose ou la forme canonique du contrat n'est pas à jour) branché en 5/5 de tools/ci-quality.sh. --enums-from pose les listes fermées du contrat depuis le dump du code. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
123 lines
5.1 KiB
Python
123 lines
5.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_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)
|