ETM-Schurig/etm-powersync-energy-plugin-etm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

merge: brief 4 T2 (client de transport) dans feature/forecast-protocol-1.1

Browse Source 
Fusion de feature/forecast-protocol-1.1-t2 (0fdbede). Conflit résolu dans etm/etm.pri :
optimizerprotocol.* (T1) et optimizerclient.* (T2) sont tous deux listés.
Cible de test renommée etm-powersync-optimizerclient (règle de nommage du 2026-10-16).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Patrick Schurig 2026-10-16 19:02:38 +02:00
commit a280820138
6 changed files with 1900 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
energyplugin/etm/etm.pri
View File

@ -18,6 +18,7 @@ HEADERS += \
$$PWD/adapters/relayrouter.h \
$$PWD/scheduler/rulebasedscheduler.h \
$$PWD/scheduler/optimizerprotocol.h \
$$PWD/scheduler/optimizerclient.h \
$$PWD/ratios/energyratioscalculator.h \
$$PWD/energyarbitrator.h \
$$PWD/spotmarketbuild.h \
@ -36,5 +37,6 @@ SOURCES += \
$$PWD/adapters/relayrouter.cpp \
$$PWD/scheduler/rulebasedscheduler.cpp \
$$PWD/scheduler/optimizerprotocol.cpp \
$$PWD/scheduler/optimizerclient.cpp \
$$PWD/energyarbitrator.cpp \
$$PWD/spotmarketbuild.cpp \

655
energyplugin/etm/scheduler/optimizerclient.cpp Normal file
View File

@ -0,0 +1,655 @@
// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
#include "optimizerclient.h"
#include <QFile>
#include <QJsonDocument>
#include <QJsonObject>
#include <QJsonParseError>
#include <QLocalSocket>
#include <QLoggingCategory>
#include <QSaveFile>
#include <QTcpSocket>
#include <QTimer>
#include <algorithm>
#include <cmath>
#include <limits>
#include <nymeasettings.h>
Q_DECLARE_LOGGING_CATEGORY(dcNymeaEnergy)
namespace {
// Plus long token lu ou écrit. Un secret partagé n'a aucune raison d'approcher cette taille ; la
// borne évite qu'un fichier posé par erreur (une archive, un journal) parte au handshake.
constexpr int kMaxTokenBytes = 4096;
// Majeure d'une version semver « M.m » : ce qui précède le premier point.
QString majorOf(const QString &version)
{
return version.section(QLatin1Char('.'), 0, 0);
}
} // namespace
// ─── OptimizerEndpoint ──────────────────────────────────────────────────────────────────────
OptimizerEndpoint OptimizerEndpoint::parse(const QString &text)
{
OptimizerEndpoint e;
static const QString unixScheme = QStringLiteral("unix://");
static const QString tcpScheme = QStringLiteral("tcp://");
if (text.startsWith(unixScheme)) {
const QString path = text.mid(unixScheme.size());
if (path.size() < 2 || !path.startsWith(QLatin1Char('/')))
return e;
e.kind = Unix;
e.path = path;
return e;
}
if (!text.startsWith(tcpScheme))
return e;
const QString rest = text.mid(tcpScheme.size());
QString host;
QString portText;
if (rest.startsWith(QLatin1Char('['))) {
const int close = rest.indexOf(QLatin1Char(']'));
if (close < 0 || rest.mid(close + 1, 1) != QLatin1String(":"))
return e;
host = rest.mid(1, close - 1);
portText = rest.mid(close + 2);
} else {
const int colon = rest.indexOf(QLatin1Char(':'));
if (colon < 0 || rest.indexOf(QLatin1Char(':'), colon + 1) >= 0)
return e; // IPv6 sans crochets : ambigu, refusé
host = rest.left(colon);
portText = rest.mid(colon + 1);
}
bool ok = false;
const uint port = portText.toUInt(&ok);
// toUInt accepte « +7400 » et les espaces de tête : on exige des chiffres seuls.
const bool digitsOnly = !portText.isEmpty()
&& std::all_of(portText.cbegin(), portText.cend(), [](QChar c) { return c.isDigit(); });
if (host.isEmpty() || !ok || !digitsOnly || port == 0 || port > 65535)
return e;
e.kind = Tcp;
e.host = host;
e.port = static_cast<quint16>(port);
return e;
}
QString OptimizerEndpoint::toString() const
{
switch (kind) {
case Unix:
return QStringLiteral("unix://") + path;
case Tcp:
if (host.contains(QLatin1Char(':')))
return QStringLiteral("tcp://[%1]:%2").arg(host).arg(port);
return QStringLiteral("tcp://%1:%2").arg(host).arg(port);
case Invalid:
break;
}
return QStringLiteral("(adresse invalide)");
}
// ─── OptimizerClientClock ───────────────────────────────────────────────────────────────────
OptimizerClientClock::OptimizerClientClock(QObject *parent)
: QObject(parent)
, m_timer(new QTimer(this))
{
m_elapsed.start();
m_timer->setSingleShot(true);
connect(m_timer, &QTimer::timeout, this, &OptimizerClientClock::wakeUp);
}
qint64 OptimizerClientClock::nowMs() const
{
return m_elapsed.elapsed();
}
void OptimizerClientClock::armWakeUp(qint64 delayMs)
{
m_timer->start(static_cast<int>(qBound(qint64(0), delayMs, qint64(std::numeric_limits<int>::max()))));
}
void OptimizerClientClock::disarmWakeUp()
{
m_timer->stop();
}
// ─── OptimizerClient ────────────────────────────────────────────────────────────────────────
OptimizerClient::OptimizerClient(const OptimizerEndpoint &endpoint, const QString &tokenPath,
const QString &engineVersion, OptimizerClientClock *clock,
QObject *parent)
: QObject(parent)
, m_endpoint(endpoint)
, m_tokenPath(tokenPath)
, m_engineVersion(engineVersion)
, m_clock(clock ? clock : new OptimizerClientClock(this))
{
connect(m_clock, &OptimizerClientClock::wakeUp, this, &OptimizerClient::onWakeUp);
}
OptimizerClient::~OptimizerClient()
{
// Pas de stop() : il émettrait stateChanged() vers des récepteurs peut-être déjà détruits.
disconnect(m_clock, nullptr, this, nullptr);
m_clock->disarmWakeUp();
dropSocket();
}
QString OptimizerClient::defaultTokenPath()
{
return NymeaSettings::settingsPath() + QLatin1Char('/') + QLatin1String(kTokenFileName);
}
bool OptimizerClient::writeToken(const QString &path, const QByteArray &token, QString *error)
{
auto fail = [error](const QString &why) {
if (error)
*error = why;
return false;
};
if (token.isEmpty()) {
if (QFile::exists(path) && !QFile::remove(path))
return fail(QStringLiteral("retrait du fichier de token impossible (%1)").arg(path));
return true;
}
// Le token est relu avec trimmed() : un blanc en bord ou un caractère de contrôle ne
// survivrait pas à l'aller-retour, et le handshake échouerait sans raison visible.
if (token.size() > kMaxTokenBytes)
return fail(QStringLiteral("token refusé : plus de %1 octets").arg(kMaxTokenBytes));
for (const char c : token) {
if (static_cast<unsigned char>(c) <= 0x20 || c == 0x7f)
return fail(QStringLiteral("token refusé : blanc ou caractère de contrôle"));
}
const QFileDevice::Permissions ownerOnly = QFileDevice::ReadOwner | QFileDevice::WriteOwner;
QSaveFile file(path);
if (!file.open(QIODevice::WriteOnly))
return fail(QStringLiteral("ouverture de %1 impossible : %2").arg(path, file.errorString()));
// Droits posés sur le temporaire AVANT d'y écrire le secret (fchmod sur le descripteur) :
// il n'existe aucun instant où le token est sur disque avec les droits de l'umask. Le
// renommage du commit les conserve. C'est aussi ce qui resserre un fichier PRÉEXISTANT :
// QSaveFile recopie ses droits sur le temporaire à l'ouverture, ils sont écrasés ici.
if (!file.setPermissions(ownerOnly)) {
file.cancelWriting();
return fail(QStringLiteral("droits 0600 impossibles sur %1 : %2").arg(path, file.errorString()));
}
if (file.write(token) != token.size()) {
file.cancelWriting();
return fail(QStringLiteral("écriture de %1 impossible : %2").arg(path, file.errorString()));
}
if (!file.commit())
return fail(QStringLiteral("validation de %1 impossible : %2").arg(path, file.errorString()));
return true;
}
void OptimizerClient::start()
{
if (m_state != Stopped)
return;
if (!m_endpoint.isValid()) {
m_lastErrorCode = QStringLiteral("invalid-endpoint");
qCWarning(dcNymeaEnergy()) << "[Optimiseur] adresse invalide — client non démarré, "
"stratégie de repli seule";
return;
}
qCInfo(dcNymeaEnergy()).noquote() << "[Optimiseur] démarrage du client vers" << m_endpoint.toString();
connectNow();
}
void OptimizerClient::stop()
{
if (m_state == Stopped)
return;
m_pendingId = 0;
m_handshakeDeadline = -1;
m_requestDeadline = -1;
m_reconnectAt = -1;
dropSocket();
m_clock->disarmWakeUp();
qCInfo(dcNymeaEnergy()).noquote() << "[Optimiseur] client arrêté (" << m_endpoint.toString() << ")";
setState(Stopped);
}
quint64 OptimizerClient::requestPlan(const QVariantMap &params)
{
if (m_state != Ready)
return 0;
if (m_pendingId != 0)
qCDebug(dcNymeaEnergy()) << "[Optimiseur] requête" << m_pendingId << "remplacée, sa réponse sera ignorée";
const quint64 id = ++m_lastId;
m_pendingId = id;
m_requestDeadline = m_clock->nowMs() + kRequestTimeoutMs;
send(QVariantMap{
{QStringLiteral("jsonrpc"), QStringLiteral("2.0")},
{QStringLiteral("id"), id},
{QStringLiteral("method"), QStringLiteral("GetPlan")},
{QStringLiteral("params"), params},
});
// Un échec d'écriture n'est pas traité ici : le socket signale sa perte (→ planFailed
// « disconnected »), et à défaut les 10 s bornent l'attente. Émettre planFailed() avant
// d'avoir rendu l'identifiant le rendrait inattribuable par l'appelant.
rearm();
return id;
}
void OptimizerClient::setState(State state)
{
if (m_state == state)
return;
m_state = state;
emit stateChanged(state);
}
void OptimizerClient::connectNow()
{
dropSocket();
m_reconnectAt = -1;
m_buffer.clear();
m_skippingLine = false;
m_errorsLoggedThisConnection.clear();
m_tokenSent = false;
m_handshakeDeadline = m_clock->nowMs() + kHandshakeTimeoutMs;
setState(Connecting);
rearm();
if (m_endpoint.kind == OptimizerEndpoint::Unix) {
auto *socket = new QLocalSocket(this);
m_socket = socket;
connect(socket, &QLocalSocket::connected, this, &OptimizerClient::onConnected);
connect(socket, &QLocalSocket::disconnected, this, &OptimizerClient::onSocketLost);
connect(socket, &QLocalSocket::errorOccurred, this, &OptimizerClient::onSocketLost);
connect(socket, &QLocalSocket::readyRead, this, &OptimizerClient::onReadyRead);
// Peut échouer DANS l'appel (chemin absent) : onSocketLost() s'exécute alors avant le
// retour et aura déjà planifié la tentative suivante. Rien ne doit suivre cette ligne.
socket->connectToServer(m_endpoint.path);
} else {
auto *socket = new QTcpSocket(this);
m_socket = socket;
connect(socket, &QTcpSocket::connected, this, &OptimizerClient::onConnected);
connect(socket, &QTcpSocket::disconnected, this, &OptimizerClient::onSocketLost);
connect(socket, &QTcpSocket::errorOccurred, this, &OptimizerClient::onSocketLost);
connect(socket, &QTcpSocket::readyRead, this, &OptimizerClient::onReadyRead);
socket->connectToHost(m_endpoint.host, m_endpoint.port);
}
}
void OptimizerClient::onConnected()
{
if (sender() != m_socket || m_state != Connecting)
return;
setState(Handshaking);
QVariantMap params{
{QStringLiteral("protocolVersion"), QLatin1String(kProtocolVersion)},
{QStringLiteral("engine"), QStringLiteral("etm-powersync-energy-plugin-etm/") + m_engineVersion},
};
// Relu à CHAQUE connexion (décision Q2) : un token remplacé vaut dès la tentative suivante.
const QByteArray token = readToken();
m_tokenSent = !token.isEmpty();
if (m_tokenSent)
params.insert(QStringLiteral("token"), QString::fromUtf8(token));
m_handshakeId = ++m_lastId;
// Ce message porte le secret : il n'est JAMAIS journalisé, à aucun niveau.
send(QVariantMap{
{QStringLiteral("jsonrpc"), QStringLiteral("2.0")},
{QStringLiteral("id"), m_handshakeId},
{QStringLiteral("method"), QStringLiteral("Handshake")},
{QStringLiteral("params"), params},
});
qCDebug(dcNymeaEnergy()) << "[Optimiseur] connecté, handshake envoyé ( id" << m_handshakeId
<< ", token" << (m_tokenSent ? "configuré" : "absent") << ")";
}
void OptimizerClient::onSocketLost()
{
if (sender() != m_socket)
return;
QString detail = m_socket ? m_socket->errorString() : QString();
switch (m_state) {
case Connecting:
failConnection(QStringLiteral("connect-failed"), detail);
break;
case Handshaking:
if (m_tokenSent)
detail += QStringLiteral(" — un token a été envoyé : l'optimiseur attend-il celui-là ?");
failConnection(QStringLiteral("closed-during-handshake"), detail);
break;
case Ready:
failConnection(QStringLiteral("connection-lost"), detail);
break;
case Stopped:
case Waiting:
break;
}
}
void OptimizerClient::onReadyRead()
{
QIODevice *socket = m_socket;
if (sender() != socket || !socket)
return;
m_buffer.append(socket->readAll());
while (m_socket == socket) {
const int newline = m_buffer.indexOf('\n');
if (m_skippingLine) {
// Reste d'une ligne déjà déclarée trop longue : jeté sans être analysé.
if (newline < 0) {
m_buffer.clear();
return;
}
m_buffer.remove(0, newline + 1);
m_skippingLine = false;
continue;
}
if (newline < 0) {
if (m_buffer.size() > kMaxLineBytes) {
countProtocolError(QStringLiteral("line-too-long"),
QStringLiteral("plus de %1 octets sans fin de ligne").arg(kMaxLineBytes));
m_buffer.clear();
m_skippingLine = true;
}
return;
}
const QByteArray line = m_buffer.left(newline);
m_buffer.remove(0, newline + 1);
if (line.size() > kMaxLineBytes) {
countProtocolError(QStringLiteral("line-too-long"),
QStringLiteral("%1 octets").arg(line.size()));
continue;
}
handleLine(line); // peut fermer la connexion : la garde de boucle le voit
}
}
void OptimizerClient::handleLine(const QByteArray &line)
{
if (line.trimmed().isEmpty())
return;
QJsonParseError parseError;
const QJsonDocument doc = QJsonDocument::fromJson(line, &parseError);
if (parseError.error != QJsonParseError::NoError || !doc.isObject()) {
// Le contenu n'est pas journalisé : il peut être volumineux, et il vient d'ailleurs.
countProtocolError(QStringLiteral("invalid-json"), QStringLiteral("%1 octets").arg(line.size()));
return;
}
const QJsonObject message = doc.object();
const QJsonValue idValue = message.value(QStringLiteral("id"));
if (idValue.isUndefined() || idValue.isNull()) {
const QJsonValue method = message.value(QStringLiteral("method"));
if (method.isString()) {
// Notification JSON-RPC (§8 UpdateLearned) : reçue, annoncée, rien de plus ici.
qCDebug(dcNymeaEnergy()) << "[Optimiseur] notification" << method.toString() << "reçue";
emit notificationReceived(method.toString(),
message.value(QStringLiteral("params")).toObject().toVariantMap());
return;
}
countProtocolError(QStringLiteral("missing-id"), QStringLiteral("message sans identifiant"));
return;
}
const double raw = idValue.toDouble(-1);
if (!idValue.isDouble() || raw < 1 || raw > 9007199254740992.0 || std::floor(raw) != raw) {
countProtocolError(QStringLiteral("unknown-id"), QStringLiteral("identifiant non entier"));
return;
}
const quint64 id = static_cast<quint64>(raw);
if (m_state == Handshaking && id == m_handshakeId) {
handleHandshakeReply(message);
} else if (m_pendingId != 0 && id == m_pendingId) {
handlePlanReply(id, message);
} else if (id <= m_lastId) {
// Requête remplacée ou expirée : sa réponse arrive trop tard, elle ne commande rien.
++m_staleResponses;
qCDebug(dcNymeaEnergy()) << "[Optimiseur] réponse obsolète ignorée ( id" << id << ")";
} else {
countProtocolError(QStringLiteral("unknown-id"), QStringLiteral("identifiant %1 jamais émis").arg(id));
}
}
void OptimizerClient::handleHandshakeReply(const QJsonObject &message)
{
if (message.contains(QStringLiteral("error"))) {
// Seul le CODE est journalisé : un serveur peut recopier n'importe quoi dans son
// message, y compris le token qu'il vient de refuser.
const int code = message.value(QStringLiteral("error")).toObject().value(QStringLiteral("code")).toInt();
QString detail = QStringLiteral("erreur JSON-RPC %1").arg(code);
if (m_tokenSent)
detail += QStringLiteral(" — un token a été envoyé : l'optimiseur attend-il celui-là ?");
failConnection(QStringLiteral("handshake-refused"), detail);
return;
}
const QJsonValue resultValue = message.value(QStringLiteral("result"));
if (!resultValue.isObject()) {
failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("result absent ou non objet"));
return;
}
const QJsonObject result = resultValue.toObject();
const QJsonValue version = result.value(QStringLiteral("protocolVersion"));
if (!version.isString() || version.toString().isEmpty()) {
failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("protocolVersion absent"));
return;
}
if (majorOf(version.toString()) != majorOf(QLatin1String(kProtocolVersion))) {
failConnection(QStringLiteral("major-mismatch"),
QStringLiteral("optimiseur en %1, moteur en %2 — majeures différentes, refusé (§3)")
.arg(version.toString().left(32), QLatin1String(kProtocolVersion)));
return;
}
const QJsonValue capabilities = result.value(QStringLiteral("capabilities"));
if (!capabilities.isUndefined() && !capabilities.isObject()) {
// §3 : présent, il RESTREINT les kinds acceptés. Illisible, on ne peut pas l'appliquer.
failConnection(QStringLiteral("invalid-handshake"), QStringLiteral("capabilities n'est pas un objet"));
return;
}
m_hasCapabilities = capabilities.isObject();
m_capabilities = capabilities.toObject().toVariantMap();
m_serverName = result.value(QStringLiteral("name")).toString();
m_serverProtocolVersion = version.toString();
m_serverPlanning = result.value(QStringLiteral("planning")).toBool();
m_handshakeDeadline = -1;
m_backoffMs = kBackoffInitialMs;
const int failuresBefore = m_failureStreak;
m_failureStreak = 0;
m_failureStreakCode.clear();
setState(Ready);
rearm();
qCInfo(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] connecté à %1 : « %2 », protocole %3, capabilities %4%5")
.arg(m_endpoint.toString(), m_serverName.left(64), m_serverProtocolVersion.left(32),
m_hasCapabilities ? QStringLiteral("annoncées") : QStringLiteral("absentes"),
failuresBefore > 0 ? QStringLiteral(" (après %1 échec(s))").arg(failuresBefore) : QString());
emit ready();
}
void OptimizerClient::handlePlanReply(quint64 id, const QJsonObject &message)
{
m_pendingId = 0;
m_requestDeadline = -1;
rearm();
if (message.contains(QStringLiteral("error"))) {
const int code = message.value(QStringLiteral("error")).toObject().value(QStringLiteral("code")).toInt();
m_lastErrorCode = QStringLiteral("rpc-error");
qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": erreur JSON-RPC" << code;
emit planFailed(id, QStringLiteral("rpc-error"));
return;
}
const QJsonValue result = message.value(QStringLiteral("result"));
if (!result.isObject()) {
m_lastErrorCode = QStringLiteral("invalid-response");
qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": réponse sans result objet";
emit planFailed(id, QStringLiteral("invalid-response"));
return;
}
qCDebug(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": réponse reçue";
emit planReceived(id, result.toObject().toVariantMap());
}
void OptimizerClient::failConnection(const QString &code, const QString &detail)
{
const quint64 pending = m_pendingId;
m_pendingId = 0;
m_requestDeadline = -1;
m_handshakeDeadline = -1;
dropSocket();
m_lastErrorCode = code;
const qint64 delay = m_backoffMs;
// Optimiseur absent : une tentative par minute au plafond. La PREMIÈRE de chaque série est
// annoncée ; les suivantes de même nature sont comptées, et la série se clôt en qCInfo au
// retour (« après N échec(s) »).
++m_failureStreak;
if (code != m_failureStreakCode) {
m_failureStreakCode = code;
qCWarning(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] %1 (%2) : %3 — nouvelle tentative dans %4 s, repli rule-based")
.arg(m_endpoint.toString(), code, detail).arg(delay / 1000.0);
} else {
qCDebug(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] %1 (%2), échec n°%3 — nouvelle tentative dans %4 s")
.arg(m_endpoint.toString(), code).arg(m_failureStreak).arg(delay / 1000.0);
}
scheduleReconnect();
if (pending != 0)
emit planFailed(pending, QStringLiteral("disconnected"));
}
void OptimizerClient::dropSocket()
{
QIODevice *socket = m_socket;
if (!socket)
return;
m_socket = nullptr;
socket->disconnect(this);
if (auto *local = qobject_cast<QLocalSocket *>(socket))
local->abort();
else if (auto *tcp = qobject_cast<QTcpSocket *>(socket))
tcp->abort();
socket->deleteLater();
m_buffer.clear();
m_skippingLine = false;
}
void OptimizerClient::scheduleReconnect()
{
m_reconnectAt = m_clock->nowMs() + m_backoffMs;
m_backoffMs = qMin(m_backoffMs * 2, kBackoffMaxMs);
setState(Waiting);
rearm();
}
void OptimizerClient::onWakeUp()
{
const qint64 now = m_clock->nowMs();
if (m_handshakeDeadline >= 0 && now >= m_handshakeDeadline
&& (m_state == Connecting || m_state == Handshaking)) {
failConnection(QStringLiteral("handshake-timeout"),
QStringLiteral("pas de handshake valide en %1 s").arg(kHandshakeTimeoutMs / 1000));
}
if (m_requestDeadline >= 0 && now >= m_requestDeadline && m_pendingId != 0) {
const quint64 id = m_pendingId;
m_pendingId = 0;
m_requestDeadline = -1;
m_lastErrorCode = QStringLiteral("timeout");
// La connexion est conservée : une réponse tardive sera reconnue obsolète par son id.
qCWarning(dcNymeaEnergy()) << "[Optimiseur] GetPlan" << id << ": pas de réponse en"
<< kRequestTimeoutMs / 1000 << "s";
emit planFailed(id, QStringLiteral("timeout"));
}
if (m_reconnectAt >= 0 && now >= m_reconnectAt && m_state == Waiting)
connectNow();
rearm();
}
void OptimizerClient::rearm()
{
qint64 next = -1;
for (const qint64 deadline : {m_handshakeDeadline, m_requestDeadline, m_reconnectAt}) {
if (deadline >= 0 && (next < 0 || deadline < next))
next = deadline;
}
if (next < 0)
m_clock->disarmWakeUp();
else
m_clock->armWakeUp(qMax<qint64>(0, next - m_clock->nowMs()));
}
void OptimizerClient::countProtocolError(const QString &code, const QString &detail)
{
++m_protocolErrors;
m_lastErrorCode = code;
// Un serveur fautif peut en envoyer à chaque ligne : la première de chaque nature est
// journalisée par connexion, le compteur porte les autres.
if (!m_errorsLoggedThisConnection.contains(code)) {
m_errorsLoggedThisConnection.append(code);
qCWarning(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] ligne écartée (%1) : %2 — connexion conservée ; "
"les suivantes de même nature sont comptées sans être journalisées")
.arg(code, detail);
} else {
qCDebug(dcNymeaEnergy()).noquote() << QStringLiteral("[Optimiseur] ligne écartée (%1)").arg(code);
}
emit protocolError(code);
}
bool OptimizerClient::send(const QVariantMap &message)
{
if (!m_socket)
return false;
QByteArray line = QJsonDocument(QJsonObject::fromVariantMap(message)).toJson(QJsonDocument::Compact);
line.append('\n');
return m_socket->write(line) == line.size();
}
QByteArray OptimizerClient::readToken()
{
if (m_tokenPath.isEmpty())
return {};
QFile file(m_tokenPath);
if (!file.exists())
return {};
if (!file.open(QIODevice::ReadOnly)) {
qCWarning(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] fichier de token %1 illisible (%2) — handshake sans token")
.arg(m_tokenPath, file.errorString());
return {};
}
const QFileDevice::Permissions tooWide = QFileDevice::ReadGroup | QFileDevice::WriteGroup
| QFileDevice::ExeGroup | QFileDevice::ReadOther | QFileDevice::WriteOther | QFileDevice::ExeOther;
if (file.permissions() & tooWide) {
qCWarning(dcNymeaEnergy()).noquote()
<< QStringLiteral("[Optimiseur] fichier de token %1 lisible au-delà de son propriétaire "
"(attendu 0600) — utilisé quand même").arg(m_tokenPath);
}
return file.read(kMaxTokenBytes).trimmed();
}

312
energyplugin/etm/scheduler/optimizerclient.h Normal file
View File

@ -0,0 +1,312 @@
// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
#pragma once
#include <QByteArray>
#include <QElapsedTimer>
#include <QObject>
#include <QPointer>
#include <QString>
#include <QStringList>
#include <QVariantMap>
class QIODevice;
class QJsonObject;
class QTimer;
/*!
* \brief Adresse d'un optimiseur : \c unix:///chemin ou \c tcp://hôte:port (OPTIMIZER_PROTOCOL §2).
*
* \invariant Une adresse non reconnue est \c Invalid, jamais « à peu près » interprétée : un
* client construit dessus ne se connecte pas.
*/
struct OptimizerEndpoint
{
/*! \brief Famille de transport. */
enum Kind { Invalid, Unix, Tcp };
Kind kind = Invalid; //!< Famille ; \c Invalid si l'adresse n'a pas été reconnue.
QString path; //!< Chemin du socket local (\c Unix seulement).
QString host; //!< Hôte (\c Tcp seulement), sans crochets pour une IPv6.
quint16 port = 0; //!< Port (\c Tcp seulement), 1-65535.
/*!
* \brief Lit une adresse d'optimiseur.
* \param text \c unix:///run/powersync/optimizer.sock ou \c tcp://192.168.1.10:7400
* (IPv6 entre crochets : \c tcp://[fd00::1]:7400).
* \return L'adresse, ou \c kind == Invalid : schéma inconnu, chemin non absolu, hôte vide,
* port absent, nul ou hors bornes, ou tout reste après le port.
*/
static OptimizerEndpoint parse(const QString &text);
/*! \brief \return Vrai si l'adresse est utilisable. */
bool isValid() const { return kind != Invalid; }
/*! \brief \return La forme textuelle canonique, pour le journal (aucun secret n'y figure). */
QString toString() const;
};
/*!
* \brief Source de temps et réveil du client, injectable (\c tools/check-simulation-clock.py).
*
* Le client ne lit JAMAIS l'horloge murale : ses trois délais (handshake 5 s, \c GetPlan 10 s,
* backoff 1 → 60 s) sont des durées, mesurées sur cette horloge MONOTONE. Il n'arme qu'un réveil
* à la fois — l'échéance la plus proche — et réévalue toutes ses échéances quand il sonne.
*
* L'implémentation par défaut lit un \c QElapsedTimer et arme un \c QTimer à coup unique. Une
* suite de test la dérive pour avancer le temps à la main : aucun délai ne s'écoule alors tant
* que le test ne l'a pas décidé.
*
* \invariant \c nowMs() ne décroît jamais.
* \invariant Après \c armWakeUp(d), \c wakeUp() est émis au plus une fois, au plus tôt \c d ms
* plus tard ; un nouvel \c armWakeUp() ou \c disarmWakeUp() remplace le réveil en attente.
*/
class OptimizerClientClock : public QObject
{
Q_OBJECT
public:
/*! \brief Horloge réelle (monotone). \param parent Propriétaire Qt. */
explicit OptimizerClientClock(QObject *parent = nullptr);
/*! \brief \return Millisecondes monotones depuis une origine arbitraire. */
virtual qint64 nowMs() const;
/*! \brief Arme le réveil unique. \param delayMs Délai, en ms (0 = au prochain tour de boucle). */
virtual void armWakeUp(qint64 delayMs);
/*! \brief Désarme le réveil en attente, s'il y en a un. */
virtual void disarmWakeUp();
signals:
/*! \brief Le délai armé est écoulé. */
void wakeUp();
private:
QElapsedTimer m_elapsed;
QTimer *m_timer = nullptr;
};
/*!
* \brief Transport client vers un optimiseur externe (OPTIMIZER_PROTOCOL §2-§4) — brief 4 T2.
*
* Un socket \c unix:// (\c QLocalSocket) ou \c tcp:// (\c QTcpSocket), JSON-RPC 2.0, un message
* par ligne. Le client se connecte, fait le handshake §3, puis porte des \c GetPlan à la demande.
* Il ne sait rien du contenu d'un plan : il rend le \c result tel quel, la validation est à
* l'appelant (T3).
*
* **Machine d'états** :
* \code
* Stopped ──start()──► Connecting ──connecté──► Handshaking ──réponse valide──► Ready
* │ │ │
* └──── échec, 5 s ──────────┴── refus, majeure, 5 s ───┤ fermeture
* ▼ ▼
* Waiting (backoff 1, 2, 4 … 60 s) ──échéance──► Connecting
* \endcode
* Les 5 s du handshake courent depuis le DÉBUT de la tentative : une connexion TCP qui ne
* s'établit jamais (hôte filtré) est bornée par le même délai. Le backoff double à chaque échec
* et revient à 1 s après un handshake réussi.
*
* **Une requête en vol au plus.** \c requestPlan() pendant qu'une requête attend la remplace :
* l'ancienne n'est plus attendue, sa réponse sera ignorée à son arrivée (comptée dans
* \c staleResponses()), et elle ne produit ni \c planReceived() ni \c planFailed(). Les
* identifiants croissent sur toute la vie du client, à travers les reconnexions.
*
* **Erreurs de ligne** — ligne non JSON, message sans identifiant, identifiant jamais émis, ligne
* de plus de \c kMaxLineBytes : comptées (\c protocolErrors()), annoncées par
* \c protocolError(), et la connexion est CONSERVÉE. La requête en vol continue d'attendre.
*
* **Le token** est relu dans \c tokenPath à CHAQUE tentative de connexion : le remplacer prend
* effet à la suivante, sans redémarrage. Il ne figure dans aucune ligne de journal, ni en clair
* ni par sa longueur ; le message de handshake qui le porte n'est jamais journalisé.
*
* \invariant Aucune méthode n'attend : toute E/S est asynchrone. \c requestPlan() rend la main
* immédiatement ; la réponse arrive par signal, au plus tôt au tour de boucle suivant.
* \invariant Le client n'écoute aucun port (invariant 10) : il se connecte, c'est tout.
* \invariant Sans \c start(), le client n'ouvre rien, n'arme rien et ne journalise rien.
*/
class OptimizerClient : public QObject
{
Q_OBJECT
public:
/*! \brief État de la connexion — voir la machine d'états de la classe. */
enum State {
Stopped, //!< Pas démarré, ou arrêté par \c stop().
Waiting, //!< Déconnecté, prochaine tentative au backoff.
Connecting, //!< Connexion du socket en cours.
Handshaking, //!< Connecté, \c Handshake envoyé, réponse attendue.
Ready //!< Handshake accepté : \c requestPlan() est servi.
};
Q_ENUM(State)
/*! \brief Version du protocole annoncée au handshake. Seule la majeure est comparée. */
static constexpr const char *kProtocolVersion = "1.1";
static constexpr qint64 kHandshakeTimeoutMs = 5000; //!< §3, depuis le début de la tentative.
static constexpr qint64 kRequestTimeoutMs = 10000; //!< §4, par requête \c GetPlan.
static constexpr qint64 kBackoffInitialMs = 1000; //!< §2, premier délai de reconnexion.
static constexpr qint64 kBackoffMaxMs = 60000; //!< §2, plafond du backoff.
/*! \brief Plus longue ligne acceptée, octets hors \c '\\n' (1 Mio, proposition T0 ; écart E24). */
static constexpr int kMaxLineBytes = 1 << 20;
/*! \brief Nom du fichier de token, sous \c NymeaSettings::settingsPath() (décision Q2). */
static constexpr const char *kTokenFileName = "energy-optimizer.token";
/*!
* \brief Construit un client ARRÊTÉ : rien n'est ouvert avant \c start().
* \param endpoint Adresse de l'optimiseur ; invalide → \c start() refuse.
* \param tokenPath Fichier du token (\c defaultTokenPath() en production) ; vide = jamais de
* token. Un fichier absent vaut « pas de token configuré ».
* \param engineVersion Version du moteur, annoncée dans \c params.engine.
* \param clock Horloge injectable ; \c nullptr = horloge réelle, possédée par le client. Une
* horloge fournie n'est pas possédée et doit survivre au client.
* \param parent Propriétaire Qt.
*/
OptimizerClient(const OptimizerEndpoint &endpoint, const QString &tokenPath,
const QString &engineVersion, OptimizerClientClock *clock = nullptr,
QObject *parent = nullptr);
~OptimizerClient() override;
/*! \brief \return \c NymeaSettings::settingsPath() + \c "/energy-optimizer.token". */
static QString defaultTokenPath();
/*!
* \brief Pose le token d'optimiseur : écriture atomique (\c QSaveFile), droits \c 0600.
*
* Les droits sont posés sur le fichier temporaire AVANT d'y écrire le secret : il n'existe
* aucun instant où le token est sur disque avec les droits de l'umask. Un fichier préexistant
* aux droits plus larges est ainsi resserré à 0600.
*
* \param path Fichier cible (\c defaultTokenPath() en production). Jamais \c energy.conf.
* \param token Secret ; vide = retrait du fichier (plus de token configuré).
* \param[out] error Motif d'échec, en français, sans le secret.
* \return Vrai si le fichier est posé (ou retiré) avec succès.
* \note Pas de RPC dans ce lot : c'est la fonction qu'elle appellera.
*/
static bool writeToken(const QString &path, const QByteArray &token, QString *error = nullptr);
/*! \brief Démarre : première tentative immédiate. Sans effet si déjà démarré. */
void start();
/*! \brief Ferme la connexion et désarme tout. La requête en vol est abandonnée sans signal. */
void stop();
/*! \brief \return L'état courant. */
State state() const { return m_state; }
/*!
* \brief Envoie un \c GetPlan. Remplace la requête en vol s'il y en a une.
* \param params Les \c params de la requête, tels quels (\c SurplusContext sérialisé, T1).
* \return L'identifiant JSON-RPC de la requête, ou 0 si le client n'est pas \c Ready (rien
* n'est envoyé : l'appelant reste sur sa stratégie de repli).
*/
quint64 requestPlan(const QVariantMap &params);
/*! \brief \return Identifiant de la requête attendue, 0 si aucune. */
quint64 pendingRequestId() const { return m_pendingId; }
/*! \brief \return \c result.capabilities du dernier handshake accepté (vide si absent). */
QVariantMap capabilities() const { return m_capabilities; }
/*! \brief \return Vrai si le dernier handshake accepté portait \c capabilities (§3 : seul
* ce cas restreint les \c kind acceptés). */
bool hasCapabilities() const { return m_hasCapabilities; }
/*! \brief \return \c result.name du dernier handshake accepté. */
QString serverName() const { return m_serverName; }
/*! \brief \return \c result.protocolVersion du dernier handshake accepté. */
QString serverProtocolVersion() const { return m_serverProtocolVersion; }
/*! \brief \return \c result.planning du dernier handshake accepté. */
bool serverPlanning() const { return m_serverPlanning; }
/*! \brief \return Erreurs de ligne comptées depuis la construction (connexion conservée). */
int protocolErrors() const { return m_protocolErrors; }
/*! \brief \return Réponses reçues pour une requête remplacée ou expirée, ignorées. */
int staleResponses() const { return m_staleResponses; }
/*! \brief \return Code de la dernière erreur (transport, handshake, ligne ou requête). */
QString lastErrorCode() const { return m_lastErrorCode; }
/*! \brief \return Délai qui sera appliqué au prochain échec de connexion, en ms. */
qint64 nextBackoffMs() const { return m_backoffMs; }
signals:
/*! \brief L'état a changé. \param state Le nouvel état. */
void stateChanged(OptimizerClient::State state);
/*! \brief Handshake accepté : \c capabilities() & co. sont à jour. */
void ready();
/*! \brief Réponse à la requête attendue. \param id Son identifiant. \param result Le
* \c result JSON-RPC, non validé. */
void planReceived(quint64 id, const QVariantMap &result);
/*!
* \brief La requête attendue n'aura pas de réponse utilisable.
* \param id Son identifiant.
* \param code \c timeout (10 s), \c disconnected (connexion perdue), \c rpc-error (réponse
* \c error), \c invalid-response (ni \c result objet ni \c error).
* \note Une requête REMPLACÉE n'émet rien : son abandon est le fait de l'appelant.
*/
void planFailed(quint64 id, const QString &code);
/*! \brief Une ligne reçue a été écartée, la connexion est conservée.
* \param code \c line-too-long, \c invalid-json, \c missing-id, \c unknown-id. */
void protocolError(const QString &code);
/*! \brief Notification JSON-RPC (\c method sans \c id), ex. \c UpdateLearned (§8). Rien
* n'en est fait ici. \param method Son nom. \param params Ses paramètres. */
void notificationReceived(const QString &method, const QVariantMap &params);
private:
void setState(State state);
void connectNow();
void onConnected();
void onSocketLost();
void onReadyRead();
void onWakeUp();
void handleLine(const QByteArray &line);
void handleHandshakeReply(const QJsonObject &message);
void handlePlanReply(quint64 id, const QJsonObject &message);
void failConnection(const QString &code, const QString &detail);
void dropSocket();
void scheduleReconnect();
void rearm();
void countProtocolError(const QString &code, const QString &detail);
bool send(const QVariantMap &message);
QByteArray readToken();
OptimizerEndpoint m_endpoint;
QString m_tokenPath;
QString m_engineVersion;
OptimizerClientClock *m_clock = nullptr;
State m_state = Stopped;
QPointer<QIODevice> m_socket;
QByteArray m_buffer;
bool m_skippingLine = false; // reste d'une ligne trop longue à jeter jusqu'au '\n'
quint64 m_lastId = 0; // dernier identifiant émis
quint64 m_handshakeId = 0;
quint64 m_pendingId = 0;
bool m_tokenSent = false; // le handshake en cours porte un token (pour le diagnostic)
qint64 m_handshakeDeadline = -1; // -1 = pas d'échéance
qint64 m_requestDeadline = -1;
qint64 m_reconnectAt = -1;
qint64 m_backoffMs = kBackoffInitialMs;
QVariantMap m_capabilities;
bool m_hasCapabilities = false;
QString m_serverName;
QString m_serverProtocolVersion;
bool m_serverPlanning = false;
int m_protocolErrors = 0;
int m_staleResponses = 0;
QString m_lastErrorCode;
QString m_failureStreakCode; // échec répété : annoncé une fois, puis compté
int m_failureStreak = 0;
QStringList m_errorsLoggedThisConnection;
};

2
tests/auto/auto.pro
View File

@ -1,2 +1,2 @@
TEMPLATE = subdirs
SUBDIRS += charging decisions loadmodel simulation spotmarket
SUBDIRS += charging decisions loadmodel optimizerclient simulation spotmarket

21
tests/auto/optimizerclient/optimizerclient.pro Normal file
View File

@ -0,0 +1,21 @@
QT += testlib network
QT -= gui
CONFIG += testcase c++17 link_pkgconfig
PKGCONFIG += nymea
TARGET = etm-powersync-optimizerclient
# Cible VOLONTAIREMENT légère, comme loadmodel : aucun serveur nymea, aucun EnergyTestBase, et
# du moteur le seul client de transport (brief 4 T2). Les faux optimiseurs sont des
# QLocalServer/QTcpServer du processus de test, et le temps est SIMULÉ : aucun des délais
# (5 s, 10 s, backoff 1 → 60 s) ne s'écoule réellement.
INCLUDEPATH += $$PWD/../../../energyplugin
HEADERS += $$PWD/../../../energyplugin/etm/scheduler/optimizerclient.h
SOURCES += $$PWD/../../../energyplugin/etm/scheduler/optimizerclient.cpp \
testoptimizerclient.cpp
target.path = $$[QT_INSTALL_PREFIX]/bin
INSTALLS += target

909
tests/auto/optimizerclient/testoptimizerclient.cpp Normal file
View File

@ -0,0 +1,909 @@
// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync
//
// Transport client vers l'optimiseur (brief 4 T2). Faux optimiseurs dans le processus
// (QLocalServer / QTcpServer), temps SIMULÉ : la boucle d'événements ne sert qu'aux E/S réelles
// des sockets, jamais à laisser s'écouler un délai du client.
#include "etm/scheduler/optimizerclient.h"
#include <QCoreApplication>
#include <QDir>
#include <QElapsedTimer>
#include <QFile>
#include <QJsonArray>
#include <QJsonDocument>
#include <QJsonObject>
#include <QLocalServer>
#include <QLocalSocket>
#include <QLoggingCategory>
#include <QPointer>
#include <QSignalSpy>
#include <QTcpServer>
#include <QTcpSocket>
#include <QTemporaryDir>
#include <QTest>
#include <sys/stat.h>
// Le moteur la définit dans smartchargingmanager.cpp ; cette cible légère ne le lie pas.
Q_LOGGING_CATEGORY(dcNymeaEnergy, "NymeaEnergy")
namespace {
// Journal capturé : TOUS les niveaux, debug compris (règle posée en initTestCase).
QStringList s_log;
QtMessageHandler s_previousHandler = nullptr;
void captureMessage(QtMsgType type, const QMessageLogContext &context, const QString &message)
{
s_log.append(message);
if (s_previousHandler)
s_previousHandler(type, context, message);
}
int countLog(const QString &needle)
{
int n = 0;
for (const QString &line : std::as_const(s_log))
n += line.contains(needle) ? 1 : 0;
return n;
}
// Attente d'une condition d'E/S. L'horloge murale ne borne QUE l'attente du test, jamais un
// délai du client.
template <typename F>
bool waitFor(F condition, int timeoutMs = 5000)
{
QElapsedTimer t;
t.start();
while (!condition()) {
if (t.elapsed() > timeoutMs)
return false;
QCoreApplication::processEvents(QEventLoop::AllEvents, 10);
}
return true;
}
// Laisse passer les événements en attente sans rien attendre de précis : sert à constater
// qu'il NE se passe rien.
void settle(int ms = 50)
{
QElapsedTimer t;
t.start();
while (t.elapsed() < ms)
QCoreApplication::processEvents(QEventLoop::AllEvents, 5);
}
} // namespace
/*! Horloge simulée : le temps n'avance que par advance(), et les réveils échus sonnent dans
* l'ordre, chacun à son instant exact. */
class SimClock : public OptimizerClientClock
{
Q_OBJECT
public:
qint64 nowMs() const override { return m_now; }
void armWakeUp(qint64 delayMs) override { m_deadline = m_now + delayMs; }
void disarmWakeUp() override { m_deadline = -1; }
bool armed() const { return m_deadline >= 0; }
qint64 remaining() const { return m_deadline - m_now; }
void advance(qint64 ms)
{
const qint64 target = m_now + ms;
while (m_deadline >= 0 && m_deadline <= target) {
m_now = m_deadline;
m_deadline = -1;
emit wakeUp();
}
m_now = target;
}
private:
qint64 m_now = 1000000; // origine arbitraire non nulle
qint64 m_deadline = -1;
};
/*! Faux optimiseur : accepte une connexion à la fois, découpe ce qu'il reçoit en lignes JSON,
* et répond ce que le test lui dicte. */
class FakeOptimizer : public QObject
{
Q_OBJECT
public:
enum Transport { Unix, Tcp };
FakeOptimizer(Transport transport, const QString &dir) : m_transport(transport), m_dir(dir) {}
bool listen()
{
if (m_transport == Unix) {
m_local = new QLocalServer(this);
QLocalServer::removeServer(socketPath());
connect(m_local, &QLocalServer::newConnection, this, [this] {
while (QLocalSocket *s = m_local->nextPendingConnection())
adopt(s);
});
return m_local->listen(socketPath());
}
m_tcp = new QTcpServer(this);
connect(m_tcp, &QTcpServer::newConnection, this, [this] {
while (QTcpSocket *s = m_tcp->nextPendingConnection())
adopt(s);
});
// Port déjà choisi lors d'une écoute précédente : on le reprend (reconnexion).
return m_tcp->listen(QHostAddress::LocalHost, m_port) && (m_port = m_tcp->serverPort()) != 0;
}
QString socketPath() const { return m_dir + QStringLiteral("/optimizer.sock"); }
QString endpoint() const
{
if (m_transport == Unix)
return QStringLiteral("unix://") + socketPath();
return QStringLiteral("tcp://127.0.0.1:%1").arg(m_port);
}
bool waitMessage(QJsonObject *out, int timeoutMs = 5000)
{
if (!waitFor([this] { return !m_messages.isEmpty(); }, timeoutMs))
return false;
*out = m_messages.takeFirst();
return true;
}
void sendRaw(const QByteArray &bytes)
{
QVERIFY(m_client);
m_client->write(bytes);
if (auto *tcp = qobject_cast<QTcpSocket *>(m_client.data()))
tcp->flush();
else if (auto *local = qobject_cast<QLocalSocket *>(m_client.data()))
local->flush();
}
void sendJson(const QJsonObject &o) { sendRaw(QJsonDocument(o).toJson(QJsonDocument::Compact) + '\n'); }
void replyResult(qint64 id, const QJsonObject &result)
{
sendJson(QJsonObject{{"jsonrpc", "2.0"}, {"id", id}, {"result", result}});
}
void abortClient()
{
if (auto *tcp = qobject_cast<QTcpSocket *>(m_client.data()))
tcp->abort();
else if (auto *local = qobject_cast<QLocalSocket *>(m_client.data()))
local->abort();
}
int connections = 0;
int disconnections = 0;
private:
void adopt(QIODevice *s)
{
++connections;
m_client = s;
m_buffer.clear();
connect(s, &QIODevice::readyRead, this, [this, s] {
m_buffer.append(s->readAll());
int nl;
while ((nl = m_buffer.indexOf('\n')) >= 0) {
m_messages.append(QJsonDocument::fromJson(m_buffer.left(nl)).object());
m_buffer.remove(0, nl + 1);
}
});
if (auto *tcp = qobject_cast<QTcpSocket *>(s))
connect(tcp, &QTcpSocket::disconnected, this, [this] { ++disconnections; });
else if (auto *local = qobject_cast<QLocalSocket *>(s))
connect(local, &QLocalSocket::disconnected, this, [this] { ++disconnections; });
}
Transport m_transport;
QString m_dir;
quint16 m_port = 0;
QLocalServer *m_local = nullptr;
QTcpServer *m_tcp = nullptr;
QPointer<QIODevice> m_client;
QByteArray m_buffer;
QList<QJsonObject> m_messages;
};
static QJsonObject handshakeResult(const QString &version = QStringLiteral("1.1"), bool withCapabilities = true)
{
QJsonObject r{{"protocolVersion", version}, {"name", "faux-optimiseur"}, {"planning", true}};
if (withCapabilities)
r.insert("capabilities", QJsonObject{{"maxHorizonH", 48}, {"kinds", QJsonArray{"setpoint", "state"}}});
return r;
}
class TestOptimizerClient : public QObject
{
Q_OBJECT
private slots:
void initTestCase();
void init();
void testEndpointParsing_data();
void testEndpointParsing();
void testNotStartedOpensNothing();
void testInvalidEndpointIsNotStarted();
void testHandshakeAndPlan_data();
void testHandshakeAndPlan();
void testWrongMajorIsRefused();
void testInvalidHandshakeIsRefused_data();
void testInvalidHandshakeIsRefused();
void testWrongTokenThenFixedToken();
void testHandshakeTimeout();
void testPlanTimeoutKeepsConnection();
void testAbruptCloseThenReconnect();
void testBackoffDoublesToSixtySeconds();
void testStaleResponseIsIgnored();
void testLineLengthLimit();
void testBadLinesKeepConnection();
void testWriteToken();
void testTokenNeverInAnyLog();
private:
// Démarre le client et mène le handshake jusqu'à Ready.
void bringUp(OptimizerClient &client, FakeOptimizer &server, const QJsonObject &result = handshakeResult());
QTemporaryDir m_dir;
};
void TestOptimizerClient::initTestCase()
{
QLoggingCategory::setFilterRules(QStringLiteral("NymeaEnergy.debug=true"));
s_previousHandler = qInstallMessageHandler(captureMessage);
}
void TestOptimizerClient::init()
{
s_log.clear();
QVERIFY(m_dir.isValid());
}
void TestOptimizerClient::bringUp(OptimizerClient &client, FakeOptimizer &server, const QJsonObject &result)
{
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
QCOMPARE(hs.value("method").toString(), QStringLiteral("Handshake"));
server.replyResult(hs.value("id").toInteger(), result);
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
}
void TestOptimizerClient::testEndpointParsing_data()
{
QTest::addColumn<QString>("text");
QTest::addColumn<int>("kind");
QTest::addColumn<QString>("canonical");
QTest::newRow("unix") << "unix:///run/powersync/optimizer.sock" << int(OptimizerEndpoint::Unix)
<< "unix:///run/powersync/optimizer.sock";
QTest::newRow("tcp ipv4") << "tcp://192.168.1.10:7400" << int(OptimizerEndpoint::Tcp) << "tcp://192.168.1.10:7400";
QTest::newRow("tcp nom") << "tcp://nas.local:7400" << int(OptimizerEndpoint::Tcp) << "tcp://nas.local:7400";
QTest::newRow("tcp ipv6") << "tcp://[fd00::1]:7400" << int(OptimizerEndpoint::Tcp) << "tcp://[fd00::1]:7400";
QTest::newRow("unix relatif") << "unix://run/x.sock" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("unix vide") << "unix://" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp sans port") << "tcp://192.168.1.10" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp port 0") << "tcp://h:0" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp port trop grand") << "tcp://h:65536" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp port signé") << "tcp://h:+80" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp reste") << "tcp://h:80/x" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp ipv6 nu") << "tcp://fd00::1:7400" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("tcp hôte vide") << "tcp://:7400" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("http") << "http://h:80" << int(OptimizerEndpoint::Invalid) << "";
QTest::newRow("vide") << "" << int(OptimizerEndpoint::Invalid) << "";
}
void TestOptimizerClient::testEndpointParsing()
{
QFETCH(QString, text);
QFETCH(int, kind);
QFETCH(QString, canonical);
const OptimizerEndpoint e = OptimizerEndpoint::parse(text);
QCOMPARE(int(e.kind), kind);
if (e.isValid())
QCOMPARE(e.toString(), canonical);
}
void TestOptimizerClient::testNotStartedOpensNothing()
{
// Le moteur n'instancie pas encore le client ; le jour où il le fera, la construction seule
// ne doit rien ouvrir, rien armer, rien écrire au journal.
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
{
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
settle();
clock.advance(120000);
settle();
QCOMPARE(client.state(), OptimizerClient::Stopped);
QCOMPARE(client.requestPlan({}), quint64(0));
}
QCOMPARE(server.connections, 0);
QVERIFY(!clock.armed());
QCOMPARE(s_log, QStringList());
}
void TestOptimizerClient::testInvalidEndpointIsNotStarted()
{
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse("tcp://nulle-part"), QString(), "9.9", &clock);
client.start();
QCOMPARE(client.state(), OptimizerClient::Stopped);
QCOMPARE(client.lastErrorCode(), QStringLiteral("invalid-endpoint"));
QVERIFY(!clock.armed());
QCOMPARE(countLog("[Optimiseur] adresse invalide"), 1);
}
void TestOptimizerClient::testHandshakeAndPlan_data()
{
QTest::addColumn<int>("transport");
QTest::newRow("unix") << int(FakeOptimizer::Unix);
QTest::newRow("tcp") << int(FakeOptimizer::Tcp);
}
void TestOptimizerClient::testHandshakeAndPlan()
{
QFETCH(int, transport);
FakeOptimizer server(FakeOptimizer::Transport(transport), m_dir.path());
QVERIFY(server.listen());
SimClock clock;
// Fichier de token ABSENT : pas de clé token du tout (ni vide, ni nulle).
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), m_dir.filePath("absent.token"), "9.9", &clock);
QSignalSpy readySpy(&client, &OptimizerClient::ready);
QSignalSpy planSpy(&client, &OptimizerClient::planReceived);
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
QCOMPARE(client.state(), OptimizerClient::Handshaking);
QCOMPARE(hs.value("jsonrpc").toString(), QStringLiteral("2.0"));
QCOMPARE(hs.value("method").toString(), QStringLiteral("Handshake"));
QCOMPARE(hs.value("id").toInteger(), 1);
const QJsonObject params = hs.value("params").toObject();
QCOMPARE(params.value("protocolVersion").toString(), QStringLiteral("1.1"));
QCOMPARE(params.value("engine").toString(), QStringLiteral("etm-powersync-energy-plugin-etm/9.9"));
QVERIFY(!params.contains("token"));
// Seule l'échéance du handshake est armée, à 5 s depuis le début de la tentative.
QVERIFY(clock.armed());
QCOMPARE(clock.remaining(), qint64(5000));
server.replyResult(1, handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
QCOMPARE(readySpy.count(), 1);
QVERIFY(client.hasCapabilities());
QCOMPARE(client.capabilities().value("maxHorizonH").toInt(), 48);
QCOMPARE(client.capabilities().value("kinds").toStringList(), (QStringList{"setpoint", "state"}));
QCOMPARE(client.serverName(), QStringLiteral("faux-optimiseur"));
QCOMPARE(client.serverProtocolVersion(), QStringLiteral("1.1"));
QVERIFY(client.serverPlanning());
QVERIFY(!clock.armed()); // prêt et rien en vol : aucun réveil
QCOMPARE(countLog("[Optimiseur] connecté à " + server.endpoint()), 1);
// GetPlan : params transmis tels quels, identifiants croissants.
const QVariantMap context{{"timestamp", "2026-10-16T10:00:00Z"}, {"loads", QVariantList{}}};
const quint64 id = client.requestPlan(context);
QCOMPARE(id, quint64(2));
QCOMPARE(clock.remaining(), qint64(10000));
QJsonObject req;
QVERIFY(server.waitMessage(&req));
QCOMPARE(req.value("method").toString(), QStringLiteral("GetPlan"));
QCOMPARE(req.value("id").toInteger(), 2);
QCOMPARE(req.value("params").toObject().toVariantMap(), context);
server.replyResult(2, QJsonObject{{"planId", "p1"}, {"slots", QJsonArray{}}});
QVERIFY(waitFor([&] { return planSpy.count() == 1; }));
QCOMPARE(planSpy.at(0).at(0).value<quint64>(), quint64(2));
QCOMPARE(planSpy.at(0).at(1).toMap().value("planId").toString(), QStringLiteral("p1"));
QCOMPARE(client.pendingRequestId(), quint64(0));
QVERIFY(!clock.armed());
QCOMPARE(client.requestPlan(context), quint64(3));
// Sans capabilities : mémorisé comme ABSENT, pas comme vide (§3, seul le présent restreint).
client.stop();
QCOMPARE(client.state(), OptimizerClient::Stopped);
bringUp(client, server, handshakeResult("1.4", false));
QVERIFY(!client.hasCapabilities());
QCOMPARE(client.serverProtocolVersion(), QStringLiteral("1.4")); // même majeure : accepté
}
void TestOptimizerClient::testWrongMajorIsRefused()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
QSignalSpy readySpy(&client, &OptimizerClient::ready);
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
server.replyResult(1, handshakeResult("2.0"));
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QCOMPARE(client.lastErrorCode(), QStringLiteral("major-mismatch"));
QCOMPARE(readySpy.count(), 0);
QCOMPARE(client.requestPlan({}), quint64(0));
QVERIFY(waitFor([&] { return server.disconnections == 1; }));
QCOMPARE(countLog("optimiseur en 2.0, moteur en 1.1 — majeures différentes, refusé"), 1);
// Nouvelle tentative au backoff (§3), pas avant.
clock.advance(999);
settle();
QCOMPARE(server.connections, 1);
clock.advance(1);
QVERIFY(server.waitMessage(&hs));
QCOMPARE(server.connections, 2);
QCOMPARE(hs.value("id").toInteger(), 2); // les identifiants traversent les reconnexions
server.replyResult(2, handshakeResult("1.0"));
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
}
void TestOptimizerClient::testInvalidHandshakeIsRefused_data()
{
QTest::addColumn<QByteArray>("reply");
QTest::addColumn<QString>("code");
QTest::newRow("protocolVersion absent")
<< QByteArray(R"({"jsonrpc":"2.0","id":1,"result":{"name":"x"}})") << "invalid-handshake";
QTest::newRow("protocolVersion non chaîne")
<< QByteArray(R"({"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1.1}})") << "invalid-handshake";
QTest::newRow("result non objet")
<< QByteArray(R"({"jsonrpc":"2.0","id":1,"result":"1.1"})") << "invalid-handshake";
QTest::newRow("capabilities non objet")
<< QByteArray(R"({"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"1.1","capabilities":["setpoint"]}})")
<< "invalid-handshake";
QTest::newRow("erreur JSON-RPC")
<< QByteArray(R"({"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"inconnu"}})") << "handshake-refused";
}
void TestOptimizerClient::testInvalidHandshakeIsRefused()
{
QFETCH(QByteArray, reply);
QFETCH(QString, code);
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
QSignalSpy readySpy(&client, &OptimizerClient::ready);
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
server.sendRaw(reply + '\n');
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QCOMPARE(client.lastErrorCode(), code);
QCOMPARE(readySpy.count(), 0);
QCOMPARE(clock.remaining(), qint64(1000));
QCOMPARE(countLog("(" + code + ") :"), 1);
}
void TestOptimizerClient::testWrongTokenThenFixedToken()
{
const QString tokenPath = m_dir.filePath("energy-optimizer.token");
QVERIFY(OptimizerClient::writeToken(tokenPath, "jeton-faux-7f3a"));
FakeOptimizer server(FakeOptimizer::Tcp, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), tokenPath, "9.9", &clock);
// 1. Mauvais token : le serveur ferme la connexion (§3).
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
QCOMPARE(hs.value("params").toObject().value("token").toString(), QStringLiteral("jeton-faux-7f3a"));
server.abortClient();
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QCOMPARE(client.lastErrorCode(), QStringLiteral("closed-during-handshake"));
QCOMPARE(countLog("un token a été envoyé"), 1);
// 2. Le serveur refuse par une erreur JSON-RPC : même issue, autre code.
clock.advance(1000);
QVERIFY(server.waitMessage(&hs));
server.sendJson(QJsonObject{{"jsonrpc", "2.0"}, {"id", hs.value("id")},
{"error", QJsonObject{{"code", -32001}, {"message", "token refusé"}}}});
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QCOMPARE(client.lastErrorCode(), QStringLiteral("handshake-refused"));
// 3. Token corrigé sur disque, sans toucher au client : relu à la connexion suivante.
QVERIFY(OptimizerClient::writeToken(tokenPath, "jeton-juste-91c2"));
clock.advance(1999);
settle();
QCOMPARE(server.connections, 2);
clock.advance(1);
QVERIFY(server.waitMessage(&hs));
QCOMPARE(hs.value("params").toObject().value("token").toString(), QStringLiteral("jeton-juste-91c2"));
server.replyResult(hs.value("id").toInteger(), handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
QCOMPARE(countLog("(après 2 échec(s))"), 1);
// 4. Token retiré : la connexion suivante part SANS clé token.
QVERIFY(OptimizerClient::writeToken(tokenPath, QByteArray()));
QVERIFY(!QFile::exists(tokenPath));
server.abortClient();
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
clock.advance(1000);
QVERIFY(server.waitMessage(&hs));
QVERIFY(!hs.value("params").toObject().contains("token"));
}
void TestOptimizerClient::testHandshakeTimeout()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs)); // reçu, jamais répondu
clock.advance(4999);
settle();
QCOMPARE(client.state(), OptimizerClient::Handshaking);
clock.advance(1);
QCOMPARE(client.state(), OptimizerClient::Waiting);
QCOMPARE(client.lastErrorCode(), QStringLiteral("handshake-timeout"));
QVERIFY(waitFor([&] { return server.disconnections == 1; }));
QCOMPARE(clock.remaining(), qint64(1000)); // prochaine tentative au backoff
// Une réponse tardive au handshake abandonné ne ressuscite rien.
clock.advance(1000);
QVERIFY(server.waitMessage(&hs));
QCOMPARE(hs.value("id").toInteger(), 2);
server.replyResult(1, handshakeResult());
settle();
QCOMPARE(client.state(), OptimizerClient::Handshaking);
QCOMPARE(client.staleResponses(), 1);
server.replyResult(2, handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
}
void TestOptimizerClient::testPlanTimeoutKeepsConnection()
{
FakeOptimizer server(FakeOptimizer::Tcp, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
bringUp(client, server);
QSignalSpy failSpy(&client, &OptimizerClient::planFailed);
QSignalSpy planSpy(&client, &OptimizerClient::planReceived);
const quint64 id = client.requestPlan({{"n", 1}});
QJsonObject req;
QVERIFY(server.waitMessage(&req));
clock.advance(9999);
settle();
QCOMPARE(failSpy.count(), 0);
clock.advance(1);
QCOMPARE(failSpy.count(), 1);
QCOMPARE(failSpy.at(0).at(0).value<quint64>(), id);
QCOMPARE(failSpy.at(0).at(1).toString(), QStringLiteral("timeout"));
QCOMPARE(client.state(), OptimizerClient::Ready); // connexion conservée
QCOMPARE(server.disconnections, 0);
QVERIFY(!clock.armed());
// La réponse tardive est reconnue par son id et ne commande rien.
server.replyResult(qint64(id), QJsonObject{{"planId", "tardif"}});
QVERIFY(waitFor([&] { return client.staleResponses() == 1; }));
QCOMPARE(planSpy.count(), 0);
const quint64 next = client.requestPlan({{"n", 2}});
QVERIFY(server.waitMessage(&req));
server.replyResult(qint64(next), QJsonObject{{"planId", "frais"}});
QVERIFY(waitFor([&] { return planSpy.count() == 1; }));
QCOMPARE(planSpy.at(0).at(1).toMap().value("planId").toString(), QStringLiteral("frais"));
}
void TestOptimizerClient::testAbruptCloseThenReconnect()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
bringUp(client, server);
QSignalSpy failSpy(&client, &OptimizerClient::planFailed);
const quint64 id = client.requestPlan({});
QJsonObject req;
QVERIFY(server.waitMessage(&req));
server.abortClient();
QVERIFY(waitFor([&] { return failSpy.count() == 1; }));
QCOMPARE(failSpy.at(0).at(0).value<quint64>(), id);
QCOMPARE(failSpy.at(0).at(1).toString(), QStringLiteral("disconnected"));
QCOMPARE(client.state(), OptimizerClient::Waiting);
QCOMPARE(client.lastErrorCode(), QStringLiteral("connection-lost"));
QCOMPARE(client.requestPlan({}), quint64(0));
clock.advance(999);
settle();
QCOMPARE(server.connections, 1);
clock.advance(1);
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
server.replyResult(hs.value("id").toInteger(), handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
QCOMPARE(client.nextBackoffMs(), qint64(1000)); // remis à 1 s après succès
}
void TestOptimizerClient::testBackoffDoublesToSixtySeconds()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path()); // n'écoute pas encore
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
int attempts = 0;
connect(&client, &OptimizerClient::stateChanged, this, [&](OptimizerClient::State s) {
if (s == OptimizerClient::Connecting)
++attempts;
});
client.start();
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QCOMPARE(attempts, 1);
QCOMPARE(client.lastErrorCode(), QStringLiteral("connect-failed"));
const QList<qint64> delays{1000, 2000, 4000, 8000, 16000, 32000, 60000, 60000};
for (const qint64 d : delays) {
const int before = attempts;
QCOMPARE(clock.remaining(), d);
clock.advance(d - 1);
settle(10);
QCOMPARE(attempts, before);
clock.advance(1);
QVERIFY(waitFor([&] { return attempts == before + 1 && client.state() == OptimizerClient::Waiting; }));
}
// Une série d'échecs identiques : annoncée UNE fois, puis comptée.
QCOMPARE(countLog("(connect-failed) :"), 1);
QVERIFY(server.listen());
clock.advance(60000);
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
server.replyResult(hs.value("id").toInteger(), handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
QCOMPARE(countLog("(après 9 échec(s))"), 1);
QCOMPARE(client.nextBackoffMs(), qint64(1000));
}
void TestOptimizerClient::testStaleResponseIsIgnored()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
bringUp(client, server);
QSignalSpy planSpy(&client, &OptimizerClient::planReceived);
QSignalSpy failSpy(&client, &OptimizerClient::planFailed);
const quint64 first = client.requestPlan({{"trigger", "cycle"}});
clock.advance(3000);
const quint64 second = client.requestPlan({{"trigger", "load-config"}});
QCOMPARE(second, first + 1);
QCOMPARE(client.pendingRequestId(), second);
QCOMPARE(clock.remaining(), qint64(10000)); // l'attente repart pour la seconde
QJsonObject a, b;
QVERIFY(server.waitMessage(&a));
QVERIFY(server.waitMessage(&b));
server.replyResult(qint64(first), QJsonObject{{"planId", "ancien"}});
QVERIFY(waitFor([&] { return client.staleResponses() == 1; }));
QCOMPARE(planSpy.count(), 0);
QCOMPARE(client.pendingRequestId(), second);
server.replyResult(qint64(second), QJsonObject{{"planId", "récent"}});
QVERIFY(waitFor([&] { return planSpy.count() == 1; }));
QCOMPARE(planSpy.at(0).at(0).value<quint64>(), second);
QCOMPARE(planSpy.at(0).at(1).toMap().value("planId").toString(), QStringLiteral("récent"));
QCOMPARE(failSpy.count(), 0); // le remplacement est silencieux
QCOMPARE(client.protocolErrors(), 0);
}
void TestOptimizerClient::testLineLengthLimit()
{
FakeOptimizer server(FakeOptimizer::Tcp, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
bringUp(client, server);
QSignalSpy planSpy(&client, &OptimizerClient::planReceived);
QSignalSpy errorSpy(&client, &OptimizerClient::protocolError);
QJsonObject req;
// Exactement kMaxLineBytes : acceptée (des blancs complètent un JSON valide).
const quint64 first = client.requestPlan({});
QVERIFY(server.waitMessage(&req));
QByteArray atLimit = QJsonDocument(QJsonObject{{"jsonrpc", "2.0"}, {"id", qint64(first)},
{"result", QJsonObject{{"planId", "limite"}}}})
.toJson(QJsonDocument::Compact);
atLimit.append(QByteArray(OptimizerClient::kMaxLineBytes - atLimit.size(), ' '));
QCOMPARE(atLimit.size(), OptimizerClient::kMaxLineBytes);
server.sendRaw(atLimit + '\n');
QVERIFY(waitFor([&] { return planSpy.count() == 1; }, 10000));
QCOMPARE(client.protocolErrors(), 0);
// Un octet de plus : écartée, comptée, connexion conservée, requête toujours attendue.
const quint64 second = client.requestPlan({});
QVERIFY(server.waitMessage(&req));
QByteArray over = QJsonDocument(QJsonObject{{"jsonrpc", "2.0"}, {"id", qint64(second)},
{"result", QJsonObject{{"planId", "trop-long"}}}})
.toJson(QJsonDocument::Compact);
over.append(QByteArray(OptimizerClient::kMaxLineBytes + 1 - over.size(), ' '));
server.sendRaw(over + '\n');
QVERIFY(waitFor([&] { return errorSpy.count() == 1; }, 10000));
QCOMPARE(errorSpy.at(0).at(0).toString(), QStringLiteral("line-too-long"));
settle();
QCOMPARE(planSpy.count(), 1);
QCOMPARE(client.state(), OptimizerClient::Ready);
QCOMPARE(client.pendingRequestId(), second);
// 3 Mio sans aucune fin de ligne : écartés AVANT la fin (la mémoire reste bornée), une seule
// erreur ; tout jusqu'au '\n' est jeté, et la ligne d'APRÈS est lue normalement.
server.sendRaw(QByteArray(3 * OptimizerClient::kMaxLineBytes, 'x'));
QVERIFY(waitFor([&] { return errorSpy.count() == 2; }, 10000));
server.sendRaw("yyy\n"); // fin de la MÊME ligne trop longue : jetée avec elle
server.replyResult(qint64(second), QJsonObject{{"planId", "après"}});
QVERIFY(waitFor([&] { return planSpy.count() == 2; }, 10000));
QCOMPARE(planSpy.at(1).at(1).toMap().value("planId").toString(), QStringLiteral("après"));
QCOMPARE(client.protocolErrors(), 2);
QCOMPARE(server.disconnections, 0);
}
void TestOptimizerClient::testBadLinesKeepConnection()
{
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), QString(), "9.9", &clock);
bringUp(client, server);
QSignalSpy planSpy(&client, &OptimizerClient::planReceived);
QSignalSpy errorSpy(&client, &OptimizerClient::protocolError);
QSignalSpy notifSpy(&client, &OptimizerClient::notificationReceived);
const quint64 id = client.requestPlan({});
QJsonObject req;
QVERIFY(server.waitMessage(&req));
server.sendRaw("ceci n'est pas du JSON\n");
server.sendRaw("[1,2,3]\n");
server.sendRaw("{\"jsonrpc\":\"2.0\",\"result\":{}}\n");
server.sendRaw("{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{\"code\":-32700}}\n");
server.sendRaw("{\"jsonrpc\":\"2.0\",\"id\":999,\"result\":{}}\n");
server.sendRaw("{\"jsonrpc\":\"2.0\",\"id\":\"abc\",\"result\":{}}\n");
server.sendRaw("{\"jsonrpc\":\"2.0\",\"method\":\"UpdateLearned\",\"params\":{\"confidence\":0.4}}\n");
server.sendRaw("\n");
QVERIFY(waitFor([&] { return errorSpy.count() == 6 && notifSpy.count() == 1; }));
QStringList codes;
for (const QList<QVariant> &e : std::as_const(errorSpy))
codes << e.at(0).toString();
QCOMPARE(codes, (QStringList{"invalid-json", "invalid-json", "missing-id", "missing-id", "unknown-id", "unknown-id"}));
QCOMPARE(notifSpy.at(0).at(0).toString(), QStringLiteral("UpdateLearned"));
QCOMPARE(client.state(), OptimizerClient::Ready);
QCOMPARE(client.pendingRequestId(), id);
// Une ligne de journal par NATURE d'erreur et par connexion, pas une par ligne.
QCOMPARE(countLog("ligne écartée (invalid-json) :"), 1);
QCOMPARE(countLog("ligne écartée (missing-id) :"), 1);
QCOMPARE(countLog("ligne écartée (unknown-id) :"), 1);
server.replyResult(qint64(id), QJsonObject{{"planId", "ok"}});
QVERIFY(waitFor([&] { return planSpy.count() == 1; }));
QCOMPARE(client.protocolErrors(), 6);
QCOMPARE(server.disconnections, 0);
}
void TestOptimizerClient::testWriteToken()
{
const QString path = m_dir.filePath("tok/energy-optimizer.token");
QVERIFY(QDir().mkpath(m_dir.filePath("tok")));
const mode_t previousUmask = ::umask(022);
struct stat st;
// Nouveau fichier : 0600 malgré un umask permissif.
QString error;
QVERIFY2(OptimizerClient::writeToken(path, "secret-un", &error), qPrintable(error));
QCOMPARE(::stat(QFile::encodeName(path).constData(), &st), 0);
QCOMPARE(int(st.st_mode & 0777), 0600);
QFile f(path);
QVERIFY(f.open(QIODevice::ReadOnly));
QCOMPARE(f.readAll(), QByteArray("secret-un"));
f.close();
// Fichier préexistant en 0644 (posé à la main) : remplacé ET resserré.
QVERIFY(QFile::setPermissions(path, QFileDevice::ReadOwner | QFileDevice::WriteOwner
| QFileDevice::ReadGroup | QFileDevice::ReadOther));
QVERIFY(OptimizerClient::writeToken(path, "secret-deux"));
QCOMPARE(::stat(QFile::encodeName(path).constData(), &st), 0);
QCOMPARE(int(st.st_mode & 0777), 0600);
QVERIFY(f.open(QIODevice::ReadOnly));
QCOMPARE(f.readAll(), QByteArray("secret-deux"));
f.close();
// Refus : blanc ou contrôle (le trimmed() de la relecture ne les rendrait pas), trop long.
QVERIFY(!OptimizerClient::writeToken(path, "avec\nretour", &error));
QVERIFY(!error.contains("avec"));
QVERIFY(!OptimizerClient::writeToken(path, " bord"));
QVERIFY(!OptimizerClient::writeToken(path, QByteArray(4097, 'a')));
QVERIFY(f.open(QIODevice::ReadOnly));
QCOMPARE(f.readAll(), QByteArray("secret-deux")); // intact
f.close();
// Seul ce fichier est écrit à côté : aucun temporaire laissé, rien d'autre (energy.conf).
QCOMPARE(QDir(m_dir.filePath("tok")).entryList(QDir::Files | QDir::Hidden), QStringList{"energy-optimizer.token"});
// Vide : retrait (plus de token configuré) ; retirer un absent n'est pas une erreur.
QVERIFY(OptimizerClient::writeToken(path, QByteArray()));
QVERIFY(!QFile::exists(path));
QVERIFY(OptimizerClient::writeToken(path, QByteArray()));
QVERIFY(OptimizerClient::defaultTokenPath().endsWith("/energy-optimizer.token"));
::umask(previousUmask);
QCOMPARE(s_log, QStringList()); // l'écriture ne journalise rien
}
void TestOptimizerClient::testTokenNeverInAnyLog()
{
// Décision Q2 : le token n'apparaît dans AUCUNE ligne de journal, à aucun niveau. Le journal
// est capturé debug compris, et chaque chemin qui manipule le token est parcouru : envoi,
// refus par fermeture, refus par erreur qui RECOPIE le token, droits trop larges, succès,
// GetPlan, perte de connexion.
const QByteArray token = "Zq81-secret-de-test-4e7b";
const QString tokenPath = m_dir.filePath("energy-optimizer.token");
QVERIFY(OptimizerClient::writeToken(tokenPath, token));
FakeOptimizer server(FakeOptimizer::Unix, m_dir.path());
QVERIFY(server.listen());
SimClock clock;
OptimizerClient client(OptimizerEndpoint::parse(server.endpoint()), tokenPath, "9.9", &clock);
client.start();
QJsonObject hs;
QVERIFY(server.waitMessage(&hs));
// Le secret a bien transité : sans cela, son absence du journal ne prouverait rien.
QCOMPARE(hs.value("params").toObject().value("token").toString().toUtf8(), token);
server.abortClient();
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
clock.advance(1000);
QVERIFY(server.waitMessage(&hs));
server.sendJson(QJsonObject{{"jsonrpc", "2.0"}, {"id", hs.value("id")},
{"error", QJsonObject{{"code", -32001},
{"message", "token " + QString::fromUtf8(token) + " refusé"}}}});
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
QVERIFY(QFile::setPermissions(tokenPath, QFileDevice::ReadOwner | QFileDevice::WriteOwner | QFileDevice::ReadOther));
clock.advance(2000);
QVERIFY(server.waitMessage(&hs));
server.replyResult(hs.value("id").toInteger(), handshakeResult());
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Ready; }));
const quint64 id = client.requestPlan({{"n", 1}});
QJsonObject req;
QVERIFY(server.waitMessage(&req));
server.replyResult(qint64(id), QJsonObject{{"planId", "p"}});
QVERIFY(waitFor([&] { return client.pendingRequestId() == 0; }));
server.abortClient();
QVERIFY(waitFor([&] { return client.state() == OptimizerClient::Waiting; }));
client.stop();
// Le journal a bien coulé, à tous les niveaux, sur chaque chemin parcouru — une absence ne
// s'affirme que dans un flux dont on a prouvé qu'il coule.
QVERIFY(countLog("handshake envoyé") >= 3); // debug
QCOMPARE(countLog("(closed-during-handshake)"), 1); // avertissement
QCOMPARE(countLog("(handshake-refused)"), 1);
QCOMPARE(countLog("erreur JSON-RPC -32001"), 1);
QCOMPARE(countLog("lisible au-delà de son propriétaire"), 1);
QCOMPARE(countLog("[Optimiseur] connecté à"), 1); // info
QCOMPARE(countLog("(connection-lost)"), 1);
QVERIFY(s_log.size() >= 8);
const QString secret = QString::fromUtf8(token);
for (const QString &line : std::as_const(s_log))
QVERIFY2(!line.contains(secret) && !line.contains("Zq81"), qPrintable("token dans le journal : " + line));
}
QTEST_GUILESS_MAIN(TestOptimizerClient)
#include "testoptimizerclient.moc"