#!/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) class TestComparaison(unittest.TestCase): """La comparaison seule (fonctions de l'outil, sans serveur).""" @classmethod def setUpClass(cls): from importlib.machinery import SourceFileLoader cls.outil = SourceFileLoader('trace_replay', OUTIL).load_module() def test_action_state_comparee_sur_son_etat_seul(self): # Brief 4 T7 : le plan du rule-based porte la puissance ESTIMÉE d'une action `state`, # l'optimiseur de référence ne la donne pas (le §6 ne la définit pas) — ce n'est pas un écart. o = self.outil maintenant = o.instant('2026-06-08T13:00:00Z') creneau = {'from': '2026-06-08T13:00:00Z', 'to': '2026-06-08T13:01:00Z'} regles = {'result': {'slots': [dict(creneau, actions=[ {'loadId': 'pac', 'kind': 'state', 'state': 3, 'powerW': 1500}, {'loadId': 'ecs', 'kind': 'setpoint', 'powerW': 1000}])]}} optimiseur = {'result': {'slots': [dict(creneau, actions=[ {'loadId': 'pac', 'kind': 'state', 'state': 3}, {'loadId': 'ecs', 'kind': 'setpoint', 'powerW': 1000.0}])]}} a = o.actions_courantes(regles, maintenant) self.assertEqual(o.differences(a, o.actions_courantes(optimiseur, maintenant), 0.5), []) optimiseur['result']['slots'][0]['actions'][0]['state'] = 4 self.assertEqual(o.differences(a, o.actions_courantes(optimiseur, maintenant), 0.5), ['pac : état 3 → 4']) optimiseur['result']['slots'][0]['actions'][0]['state'] = 3 optimiseur['result']['slots'][0]['actions'][1]['powerW'] = 1200 self.assertEqual(o.differences(a, o.actions_courantes(optimiseur, maintenant), 0.5), ['ecs : 1000 W → 1200 W']) if __name__ == '__main__': unittest.main(verbosity=2)