f20b08a3a2
- testOverloadLeavesTheChargerItsPhases (3 cas) et testAnUnassignedCarTakesItsPhasesAway : une borne triphasée sans voiture, ou dont la voiture monophasée est désassignée, comptée sur une phase — 3 680 W alloués et attendus pour 16 A × 3. - testOverloadReactsToTheTotalWhenPhasesLag : des courants par phase en retard sur le total retardent la protection. - testPhaseSafetyMargin : la marge de 10 % (décision du 2026-10-27). - testTheDrawCapHoldsAgainstAnOptimizerAtMaximum (refoptimizer) : sous un optimiseur qui demande le maximum, ce que la protection a retiré revient au cycle suivant. - Témoins verts sur etm76 : testSheddingOpensTheRelaysUnderMinOn (le délestage franchit minOn et ouvre les relais) et testCommandDivergenceSeesAPhaseCountError (×3 vu en régime établi). - Optimiseur de référence : mode over-limit (toute consigne en watts à 50 kW). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
413 lines
22 KiB
Python
413 lines
22 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 ;
|
|
4. `error.data` du -32602 de GetPlan (protocole 1.3) : forme exigée d'un optimiseur 1.3, avertie
|
|
d'un optimiseur 1.2, recopie d'une valeur reçue refusée des deux.
|
|
"""
|
|
|
|
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'),
|
|
# Modes 1.3 : sur un contexte VALIDE et le bon token, ils se comportent comme `normal`.
|
|
'old-data': ('—', 'valide'),
|
|
'echo-value': ('—', 'valide'),
|
|
'major-first': ('—', 'valide'),
|
|
# Une consigne hors des bornes est un plan VALIDE : c'est au moteur de l'écrêter, pas au contrat.
|
|
'over-limit': ('—', 'valide'),
|
|
}
|
|
|
|
# Une valeur que nul chemin conforme ne peut contenir (tiret, deux-points) : si elle apparaît dans
|
|
# error.data, c'est une recopie.
|
|
MARQUEUR = 'valeur-sonde:7Qz9'
|
|
|
|
|
|
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
|
|
|
|
|
|
def session_contexte_invalide(dossier, mode, annonce, params):
|
|
"""Handshake accepté puis UN GetPlan de contexte invalide ; rend le fichier capturé."""
|
|
sock_path = os.path.join(dossier, 'opt.sock')
|
|
capture = os.path.join(dossier, f'invalide-{mode}-{annonce}.ndjson')
|
|
proc = subprocess.Popen([sys.executable, '-I', '-B', REF, '--listen', f'unix://{sock_path}', '--mode', mode,
|
|
'--announce', annonce, '--capture', capture],
|
|
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'']
|
|
s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'Handshake',
|
|
'params': {'protocolVersion': VERSION_MOTEUR,
|
|
'engine': 'etm-powersync-energy-plugin-etm/test'}}).encode() + b'\n')
|
|
assert lire_ligne(s, tampon)['result']['protocolVersion'] == annonce
|
|
s.sendall(json.dumps({'jsonrpc': '2.0', 'id': 2, 'method': 'GetPlan', 'params': params}).encode() + b'\n')
|
|
reponse = lire_ligne(s, tampon)
|
|
assert reponse['error']['code'] == -32602, reponse
|
|
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)
|
|
|
|
|
|
class TestDonneesDErreur(unittest.TestCase):
|
|
"""error.data du -32602 de GetPlan (protocole 1.3), sur de vraies sessions de l'optimiseur de
|
|
référence. La session porte un GetPlan de contexte invalide : le validateur la juge SANS
|
|
--semantic (sinon la réponse error est un refus, quel que soit son data)."""
|
|
|
|
# (mode, version annoncée) → (code de sortie, écarts > 0, avertissements > 0)
|
|
ATTENDU = {
|
|
('normal', '1.3'): (0, False, False),
|
|
('normal', '1.2'): (0, False, False),
|
|
('old-data', '1.3'): (1, True, False), # forme d'avant 1.3, annoncée 1.3 : refusée
|
|
('old-data', '1.2'): (0, False, True), # la même, annoncée 1.2 : avertie
|
|
('echo-value', '1.3'): (1, True, False),
|
|
('echo-value', '1.2'): (1, True, True), # forme avertie, RECOPIE refusée (règle de 1.2)
|
|
}
|
|
|
|
def test_forme_exigee_en_1_3_avertie_en_1_2_recopie_refusee(self):
|
|
params = copy.deepcopy(json.load(open(REQUETE, encoding='utf-8')))
|
|
params['timestamp'] = MARQUEUR # sans décalage, illisible : `format`
|
|
obtenu = {}
|
|
with tempfile.TemporaryDirectory() as d:
|
|
for (mode, annonce) in self.ATTENDU:
|
|
capture = session_contexte_invalide(d, mode, annonce, params)
|
|
code, r, sortie = verifier('--ndjson', capture, '--responses-only')
|
|
self.assertEqual(r['comptes'].get('GetPlan.error'), 1, sortie)
|
|
self.assertNotIn('GetPlan.params', r['comptes'])
|
|
self.assertEqual(r['comptes'].get('GetPlan.error.data'), 1, sortie)
|
|
obtenu[(mode, annonce)] = (code, r['ecarts'] > 0, r['avertissements'] > 0)
|
|
print(f' {mode:11s} {annonce} → code {code}, écarts {r["ecarts"]}, avertissements '
|
|
f'{r["avertissements"]}', file=sys.stderr)
|
|
if mode == 'echo-value':
|
|
self.assertIn('recopie la valeur reçue en params.timestamp', sortie)
|
|
self.assertNotIn(MARQUEUR, sortie, 'le validateur recopie lui-même la valeur')
|
|
if (mode, annonce) == ('old-data', '1.2'):
|
|
self.assertIn('AVERTISSEMENT', sortie)
|
|
self.assertIn('à aligner avant de s\'annoncer 1.3', sortie)
|
|
self.assertEqual(obtenu, self.ATTENDU)
|
|
|
|
def test_la_forme_vient_du_contrat(self):
|
|
# Contre-épreuve : sans x-etm-data-schema, la forme d'avant 1.3 annoncée 1.3 passe — c'est
|
|
# bien le contrat qui juge, pas une règle cachée dans l'outil.
|
|
with open(SCHEMA, encoding='utf-8') as f:
|
|
s = json.load(f)
|
|
for e in next(m for m in s['methods'] if m['name'] == 'GetPlan')['errors']:
|
|
e.pop('x-etm-data-schema', None)
|
|
e.pop('x-etm-data-required-from', None)
|
|
with tempfile.TemporaryDirectory() as d:
|
|
schema = os.path.join(d, 's.json')
|
|
with open(schema, 'w', encoding='utf-8') as f:
|
|
json.dump(s, f, ensure_ascii=False)
|
|
capture = session_contexte_invalide(d, 'old-data', '1.3', {'timestamp': MARQUEUR})
|
|
self.assertEqual(verifier('--ndjson', capture, '--responses-only')[0], 1)
|
|
code, r, sortie = verifier('--schema', schema, '--ndjson', capture, '--responses-only')
|
|
self.assertEqual(code, 0, sortie)
|
|
self.assertNotIn('GetPlan.error.data', r['comptes'])
|
|
|
|
def ecrire(self, d, nom, contenu):
|
|
chemin = os.path.join(d, nom)
|
|
with open(chemin, 'w', encoding='utf-8') as f:
|
|
json.dump(contenu, f, ensure_ascii=False)
|
|
return chemin
|
|
|
|
def juger(self, d, data, requete=None, version='1.3'):
|
|
args = ['--method', 'GetPlan', '--error',
|
|
self.ecrire(d, 'e.json', {'code': -32602, 'message': 'contexte invalide', 'data': data}),
|
|
'--handshake', self.ecrire(d, 'h.json', {'protocolVersion': version})]
|
|
if requete is not None:
|
|
args += ['--request', self.ecrire(d, 'q.json', requete)]
|
|
return verifier(*args)
|
|
|
|
def test_la_forme_ligne_a_ligne(self):
|
|
p = {'path': 'params.loads[0].id', 'problem': 'missing'}
|
|
cas = [
|
|
({'problems': [p]}, 0, None),
|
|
({'problems': [{'path': 'params', 'problem': 'type'}]}, 0, None),
|
|
({'problems': [{'path': 'params.loads[0].declared.estimatedPowerW.3', 'problem': 'range'}]}, 0, None),
|
|
({'problems': [p] * 50, 'truncated': True}, 0, None),
|
|
({'problems': []}, 1, '[minItems]'),
|
|
({'problems': [p] * 51}, 1, '[maxItems]'),
|
|
({'problems': [p] * 50, 'truncated': False}, 1, '[const]'),
|
|
({'problems': [p] * 3, 'truncated': True}, 1, '[x-etm-truncated]'),
|
|
({'problems': [dict(p, problem='absent')]}, 1, '[enum]'),
|
|
({'problems': [dict(p, path='/loads/0/id')]}, 1, '[pattern]'),
|
|
({'problems': [dict(p, path='loads[0].id')]}, 1, '[pattern]'),
|
|
({'problems': [dict(p, path='params.loads.0.id ')]}, 1, '[pattern]'),
|
|
({'problems': [dict(p, value=3)]}, 1, '[additionalProperties]'),
|
|
({'problems': [p], 'detail': 'x'}, 1, '[additionalProperties]'),
|
|
({'problems': [{'path': 'params.timestamp'}]}, 1, '[required]'),
|
|
({'problemes': ['params.timestamp : format']}, 1, '[required]'),
|
|
('contexte invalide', 1, '[type]'),
|
|
]
|
|
with tempfile.TemporaryDirectory() as d:
|
|
for data, code_attendu, mot in cas:
|
|
code, r, sortie = self.juger(d, data)
|
|
self.assertEqual(code, code_attendu, f'{data!r}\n{sortie}')
|
|
if mot:
|
|
self.assertIn(mot, sortie, data)
|
|
# data absent : écart en 1.3
|
|
chemin = self.ecrire(d, 'e.json', {'code': -32602, 'message': 'contexte invalide'})
|
|
code, _, sortie = verifier('--method', 'GetPlan', '--error', chemin)
|
|
self.assertEqual(code, 1, sortie)
|
|
self.assertIn('« data » manquant', sortie)
|
|
# Un autre code d'erreur n'a pas de forme normée : rien à juger.
|
|
chemin = self.ecrire(d, 'e.json', {'code': -32603, 'message': 'erreur interne', 'data': [1, 'x']})
|
|
code, r, sortie = verifier('--method', 'GetPlan', '--error', chemin)
|
|
self.assertEqual((code, r['ecarts']), (0, 0), sortie)
|
|
|
|
def test_la_recopie_ne_peut_pas_accuser_un_chemin_conforme(self):
|
|
# Valeurs de la requête qui sont AUSSI des noms de champ (domain « battery », trigger
|
|
# « forecast », meter.source « meter ») : un chemin conforme les contient, ce n'est pas une
|
|
# recopie. Une valeur à tiret ou à deux-points recopiée, elle, l'est.
|
|
requete = copy.deepcopy(json.load(open(REQUETE, encoding='utf-8')))
|
|
requete['trigger'] = 'forecast'
|
|
requete['loads'][0]['domain'] = 'battery'
|
|
data = {'problems': [{'path': 'params.battery.socPercent', 'problem': 'range'},
|
|
{'path': 'params.forecast.pv[3].from', 'problem': 'format'},
|
|
{'path': 'params.meter.source', 'problem': 'enum'}]}
|
|
with tempfile.TemporaryDirectory() as d:
|
|
code, r, sortie = self.juger(d, data, requete)
|
|
self.assertEqual(code, 0, sortie)
|
|
ident = requete['loads'][0]['id']
|
|
self.assertRegex(ident, '[^A-Za-z0-9_]')
|
|
code, r, sortie = self.juger(d, {'problemes': [f'charge {ident} : priorité hors bornes']}, requete, '1.2')
|
|
self.assertEqual(code, 1, sortie)
|
|
self.assertIn('recopie la valeur reçue en params.loads[0].id', sortie)
|
|
self.assertNotIn(ident, sortie.split('RÉSUMÉ')[0])
|
|
# Avant 1.2, la règle n'existait pas : avertie seulement.
|
|
code, r, sortie = self.juger(d, {'problemes': [f'charge {ident}']}, requete, '1.1')
|
|
self.assertEqual((code, r['ecarts'], r['avertissements'] > 0), (0, 0, True), sortie)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|