64a864bfaa
Python 3, bibliothèque standard seule. unix:// par défaut (unix:///run/powersync/optimizer.sock), tcp:// accepté ; token lu dans un fichier (--token-file), jamais en argument ni affiché. Réponses écrites à côté de la trace, jamais d'écrasement ; résumé : requêtes, taux d'abstention, computeMs médian et max, cycles non comparables, cycles dont les actions du créneau courant diffèrent du plan enregistré. --fail-on-divergence : code 1. Test (unittest) contre un serveur Unix minimal qui renvoie les réponses enregistrées (ce n'est pas l'optimiseur de référence) : trace synthétique et trois traces réelles du moteur → zéro divergence ; réponse modifiée → divergence vue, code 1 ; tcp://, refus d'écraser, endpoint absent. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
233 lines
11 KiB
Python
233 lines
11 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-trace-replay (brief 4 T6).
|
|
|
|
python3 -I tests/tools/etm-powersync-trace-replay-test.py
|
|
|
|
Le serveur ci-dessous n'est PAS l'optimiseur de référence (T7) : il ne planifie rien, il RENVOIE
|
|
la réponse enregistrée dans la trace pour la requête reçue (reconnue par son `timestamp`), et
|
|
une abstention quand la réponse enregistrée est une erreur. Rejouer une trace contre lui doit
|
|
donc donner zéro divergence ; une réponse modifiée doit être vue.
|
|
|
|
Les traces rejouées : une trace synthétique (toutes les formes de ligne), et les traces RÉELLES
|
|
du moteur sous tests/tools/data/ (exportées par la suite `socketscheduler`,
|
|
ETM_POWERSYNC_TRACE_EXPORT).
|
|
"""
|
|
|
|
import glob
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
|
|
ICI = os.path.dirname(os.path.abspath(__file__))
|
|
OUTIL = os.path.join(ICI, '..', '..', 'tools', 'etm-powersync-trace-replay')
|
|
DONNEES = os.path.join(ICI, 'data')
|
|
SECRET = 'T6-secret"de\\test-0123456789'
|
|
|
|
|
|
class ServeurEnregistre:
|
|
"""Serveur NDJSON / JSON-RPC 2.0 minimal : handshake, puis GetPlan → réponse enregistrée."""
|
|
|
|
def __init__(self, trace, adresse, modifier=None):
|
|
self.reponses = {}
|
|
with open(trace, encoding='utf-8') as f:
|
|
for l in f:
|
|
if not l.strip():
|
|
continue
|
|
d = json.loads(l)
|
|
req = d.get('request')
|
|
if isinstance(req, dict):
|
|
self.reponses[req.get('timestamp')] = d.get('response')
|
|
self.modifier = modifier # timestamp dont la réponse est altérée
|
|
self.tokens = []
|
|
self.requetes = 0
|
|
if isinstance(adresse, str):
|
|
self.ecoute = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
else:
|
|
self.ecoute = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
self.ecoute.bind(adresse)
|
|
self.ecoute.listen(1)
|
|
self.adresse = self.ecoute.getsockname()
|
|
self.fil = threading.Thread(target=self.servir, daemon=True)
|
|
self.fil.start()
|
|
|
|
def repondre(self, req):
|
|
enregistree = self.reponses.get(req.get('timestamp'))
|
|
if not isinstance(enregistree, dict) or 'result' not in enregistree:
|
|
return {'abstain': True, 'reason': 'aucune réponse enregistrée'}
|
|
res = json.loads(json.dumps(enregistree['result']))
|
|
if req.get('timestamp') == self.modifier:
|
|
for c in res.get('slots') or []:
|
|
for a in c.get('actions') or []:
|
|
a['powerW'] = float(a.get('powerW') or 0) + 250
|
|
return res
|
|
|
|
def servir(self):
|
|
conn, _ = self.ecoute.accept()
|
|
tampon = b''
|
|
with conn:
|
|
while True:
|
|
morceau = conn.recv(65536)
|
|
if not morceau:
|
|
return
|
|
tampon += morceau
|
|
while b'\n' in tampon:
|
|
ligne, _, tampon = tampon.partition(b'\n')
|
|
m = json.loads(ligne)
|
|
if m.get('method') == 'Handshake':
|
|
self.tokens.append(m['params'].get('token'))
|
|
res = {'protocolVersion': '1.1', 'name': 'serveur-de-test', 'planning': False}
|
|
else:
|
|
self.requetes += 1
|
|
res = self.repondre(m.get('params') or {})
|
|
conn.sendall(json.dumps({'jsonrpc': '2.0', 'id': m['id'], 'result': res}).encode() + b'\n')
|
|
|
|
def fermer(self):
|
|
self.ecoute.close()
|
|
|
|
|
|
def trace_synthetique(chemin):
|
|
"""Une ligne de chaque forme : rule-based, optimiseur (2 créneaux), abstention, erreur
|
|
enregistrée (non comparable), L2 (sans requête)."""
|
|
def req(ts):
|
|
return {'timestamp': ts, 'trigger': 'cycle', 'loads': [{'id': 'ecs'}, {'id': 'pac'}]}
|
|
|
|
def creneau(de, a, actions):
|
|
return {'from': de, 'to': a, 'actions': actions}
|
|
|
|
lignes = [
|
|
{'t': '2026-06-08T15:00:00+02:00', 'request': req('2026-06-08T13:00:00Z'),
|
|
'response': {'result': {'strategy': 'rules', 'slots': [creneau(
|
|
'2026-06-08T13:00:00Z', '2026-06-08T13:01:00Z',
|
|
[{'loadId': 'ecs', 'kind': 'setpoint', 'powerW': 1000, 'reasonCode': 'SURPLUS_SETPOINT'},
|
|
{'loadId': 'pac', 'kind': 'state', 'powerW': 0, 'state': 2, 'reasonCode': 'SGREADY_NORMAL'}])]}},
|
|
'executed': [], 'strategy': 'rules'},
|
|
{'t': '2026-06-08T15:01:00+02:00', 'request': req('2026-06-08T13:01:00Z'),
|
|
'response': {'result': {'planId': 'p', 'strategy': 'x', 'diagnostics': {'computeMs': 12},
|
|
'slots': [creneau('2026-06-08T12:00:00Z', '2026-06-08T13:00:00Z', []),
|
|
creneau('2026-06-08T13:00:00Z', '2026-06-08T14:00:00Z',
|
|
[{'loadId': 'ecs', 'kind': 'setpoint', 'powerW': 1500,
|
|
'reason': 'texte libre'}])]}},
|
|
'executed': [], 'strategy': 'optimizer', 'latencyMs': 15},
|
|
{'t': '2026-06-08T15:02:00+02:00', 'request': req('2026-06-08T13:02:00Z'),
|
|
'response': {'result': {'abstain': True, 'reason': 'recalcul', 'diagnostics': {'computeMs': 40}}},
|
|
'executed': [], 'strategy': 'rules', 'latencyMs': 41},
|
|
{'t': '2026-06-08T15:03:00+02:00', 'request': req('2026-06-08T13:03:00Z'),
|
|
'response': {'error': {'code': 'no-response'}}, 'executed': [], 'strategy': 'rules'},
|
|
{'t': '2026-06-08T15:04:00+02:00', 'executed': [], 'strategy': 'none'},
|
|
]
|
|
with open(chemin, 'w', encoding='utf-8') as f:
|
|
for l in lignes:
|
|
f.write(json.dumps(l, separators=(',', ':')) + '\n')
|
|
|
|
|
|
class TestTraceReplay(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.dossier = tempfile.TemporaryDirectory()
|
|
self.d = self.dossier.name
|
|
self.trace = os.path.join(self.d, 'etm-powersync-optimizer-trace-2026-06-08.ndjson')
|
|
trace_synthetique(self.trace)
|
|
self.token = os.path.join(self.d, 'token')
|
|
with open(self.token, 'w') as f:
|
|
f.write(SECRET + '\n')
|
|
os.chmod(self.token, 0o600)
|
|
self.serveurs = []
|
|
|
|
def tearDown(self):
|
|
for s in self.serveurs:
|
|
s.fermer()
|
|
self.dossier.cleanup()
|
|
|
|
def serveur(self, trace, modifier=None, tcp=False):
|
|
adresse = ('127.0.0.1', 0) if tcp else os.path.join(self.d, f'opt-{len(self.serveurs)}.sock')
|
|
s = ServeurEnregistre(trace, adresse, modifier)
|
|
self.serveurs.append(s)
|
|
endpoint = f'tcp://127.0.0.1:{s.adresse[1]}' if tcp else f'unix://{adresse}'
|
|
return s, endpoint
|
|
|
|
def rejouer(self, *args):
|
|
p = subprocess.run([sys.executable, '-I', OUTIL, *args], capture_output=True, text=True, timeout=60)
|
|
return p.returncode, p.stdout, p.stderr
|
|
|
|
def test_zero_divergence_sur_unix(self):
|
|
s, endpoint = self.serveur(self.trace)
|
|
code, out, err = self.rejouer(self.trace, '--endpoint', endpoint, '--token-file', self.token,
|
|
'--fail-on-divergence')
|
|
self.assertEqual(code, 0, out + err)
|
|
self.assertIn('requêtes rejouées : 4', out)
|
|
self.assertIn('abstentions : 2 (50.0 %)', out) # l'abstention, et la réponse à l'erreur
|
|
self.assertIn('computeMs : médian 26, max 40', out)
|
|
self.assertIn('cycles non comparables (réponse enregistrée en erreur, ou rejouée en erreur) : 1', out)
|
|
self.assertIn('cycles dont les actions diffèrent du plan enregistré : 0', out)
|
|
self.assertEqual(s.requetes, 4) # la ligne L2 n'est pas rejouée
|
|
# Le token voyage dans le handshake, et nulle part ailleurs.
|
|
self.assertEqual(s.tokens, [SECRET])
|
|
sorties = glob.glob(os.path.join(self.d, '*.replay-*'))
|
|
self.assertEqual(len(sorties), 2, sorties) # réponses + résumé, à côté de la trace
|
|
for chemin in sorties:
|
|
with open(chemin, encoding='utf-8') as f:
|
|
self.assertNotIn('0123456789', f.read())
|
|
self.assertNotIn('0123456789', out + err)
|
|
with open([x for x in sorties if x.endswith('.ndjson')][0], encoding='utf-8') as f:
|
|
reponses = [json.loads(l) for l in f]
|
|
self.assertEqual([r['line'] for r in reponses], [1, 2, 3, 4])
|
|
self.assertFalse(any(r['divergent'] for r in reponses))
|
|
|
|
def test_divergence_detectee_et_code_non_nul(self):
|
|
s, endpoint = self.serveur(self.trace, modifier='2026-06-08T13:01:00Z')
|
|
code, out, err = self.rejouer(self.trace, '--endpoint', endpoint, '--fail-on-divergence')
|
|
self.assertEqual(code, 1, out + err)
|
|
self.assertIn('cycles dont les actions diffèrent du plan enregistré : 1', out)
|
|
self.assertIn('ligne 2 (2026-06-08T15:01:00+02:00) : ecs : 1500 W → 1750.0 W', out)
|
|
# Sans l'option : la divergence est dite, le code reste 0.
|
|
s2, endpoint2 = self.serveur(self.trace, modifier='2026-06-08T13:01:00Z')
|
|
code, out, _ = self.rejouer(self.trace, '--endpoint', endpoint2)
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('diffèrent du plan enregistré : 1', out)
|
|
|
|
def test_tcp_accepte(self):
|
|
_, endpoint = self.serveur(self.trace, tcp=True)
|
|
code, out, err = self.rejouer(self.trace, '--endpoint', endpoint, '--fail-on-divergence')
|
|
self.assertEqual(code, 0, out + err)
|
|
self.assertIn('diffèrent du plan enregistré : 0', out)
|
|
|
|
def test_jamais_d_ecrasement(self):
|
|
_, endpoint = self.serveur(self.trace)
|
|
sortie = os.path.join(self.d, 'deja-la.ndjson')
|
|
with open(sortie, 'w') as f:
|
|
f.write('à garder\n')
|
|
code, _, err = self.rejouer(self.trace, '--endpoint', endpoint, '--out', sortie)
|
|
self.assertEqual(code, 2)
|
|
self.assertIn('refus d\'écraser', err)
|
|
with open(sortie) as f:
|
|
self.assertEqual(f.read(), 'à garder\n')
|
|
|
|
def test_endpoint_absent(self):
|
|
code, _, err = self.rejouer(self.trace, '--endpoint', 'unix://' + os.path.join(self.d, 'absent.sock'))
|
|
self.assertEqual(code, 2)
|
|
self.assertIn('connexion impossible', err)
|
|
|
|
def test_traces_reelles_du_moteur(self):
|
|
fichiers = sorted(glob.glob(os.path.join(DONNEES, 'etm-powersync-trace-*.ndjson')))
|
|
self.assertTrue(fichiers, 'aucune trace réelle sous tests/tools/data')
|
|
for trace in fichiers:
|
|
with self.subTest(trace=os.path.basename(trace)):
|
|
_, endpoint = self.serveur(trace)
|
|
out_dir = tempfile.mkdtemp(dir=self.d)
|
|
code, out, err = self.rejouer(trace, '--endpoint', endpoint, '--fail-on-divergence',
|
|
'--out', os.path.join(out_dir, 'r.ndjson'))
|
|
self.assertEqual(code, 0, out + err)
|
|
self.assertIn('diffèrent du plan enregistré : 0', out)
|
|
self.assertNotIn('requêtes rejouées : 0', out)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|