// SPDX-License-Identifier: GPL-3.0-or-later // Copyright (C) 2025 - 2026, Patrick Schurig / ETM PowerSync // // Faux optimiseur dans le processus de test (QLocalServer / QTcpServer) — extrait de la suite // `optimizerclient` (brief 4 T2) pour être partagé avec `socketscheduler` (brief 4 T3). #pragma once #include #include #include #include #include #include #include #include #include #include #include #include #include // Attente d'une condition d'E/S. L'horloge murale ne borne QUE l'attente du test, jamais un // délai du client. template 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. inline void settle(int ms = 50) { QElapsedTimer t; t.start(); while (t.elapsed() < ms) QCoreApplication::processEvents(QEventLoop::AllEvents, 5); } /*! 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(m_client.data())) tcp->flush(); else if (auto *local = qobject_cast(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(m_client.data())) tcp->abort(); else if (auto *local = qobject_cast(m_client.data())) local->abort(); } //! L'optimiseur disparaît : la connexion est coupée et le serveur cesse d'écouter — chaque //! tentative du client échoue jusqu'au prochain listen() (même endpoint, même port). void fermer() { abortClient(); delete m_local; m_local = nullptr; delete m_tcp; m_tcp = nullptr; } int connections = 0; int disconnections = 0; //! Tout ce que le MOTEUR a envoyé à un faux optimiseur dans ce processus, dans l'ordre — jugé //! contre le contrat OpenRPC par la suite (lot « contrat OpenRPC »), puis vidé. static QList &capture() { static QList messages; return messages; } //! Répondeur automatique (brief 4 T3) : appelé sur chaque message reçu ; s'il rend vrai, le //! message est consommé et n'entre pas dans la file de waitMessage(). Vide = tout est filé. std::function responder; 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) { const QJsonObject message = QJsonDocument::fromJson(m_buffer.left(nl)).object(); m_buffer.remove(0, nl + 1); capture().append(message); if (!(responder && responder(message))) m_messages.append(message); } }); if (auto *tcp = qobject_cast(s)) connect(tcp, &QTcpSocket::disconnected, this, [this] { ++disconnections; }); else if (auto *local = qobject_cast(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 m_client; QByteArray m_buffer; QList m_messages; }; inline 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; }