#include "common/font_contract.h"
#include "common/ipc.h"
#include "common/sandbox.h"
#include "common/worker_process.h"

#include <QCoreApplication>
#include <QFile>
#include <QLocalSocket>
#include <QSignalSpy>
#include <QTemporaryDir>
#include <QTimer>
#include <QtEndian>
#include <QtTest>
#include <cstdlib>
#include <thread>

using namespace docview;

#ifndef Q_MOC_RUN
namespace {

QCborMap replyTo(const QCborMap &request, const QCborMap &result) {
    QCborMap reply = request;
    reply.remove(QStringLiteral("payload"));
    reply.insert(QStringLiteral("result"), result);
    return reply;
}

bool writeRaw(QLocalSocket &socket, const QCborMap &reply) {
    // Deliberately bypass the sender's envelope validation: this process is
    // the hostile peer, and the test exercises the receiving broker.
    const auto bytes = QCborValue(reply).toCbor();
    char header[4];
    qToBigEndian<quint32>(quint32(bytes.size()), header);
    const bool ok = socket.write(header, 4) == 4 && socket.write(bytes) == bytes.size();
    socket.flush();
    return ok;
}

int fakeWorker(QCoreApplication &app) {
    const auto args = app.arguments();
    const int socketArgument = args.indexOf("--socket");
    const int behaviorArgument = args.indexOf("--fake-worker");
    if (socketArgument < 0 || behaviorArgument < 0 || socketArgument + 1 >= args.size() ||
        behaviorArgument + 1 >= args.size())
        return 120;
    const auto behavior = args[behaviorArgument + 1];
    QLocalSocket socket;
    socket.connectToServer(args[socketArgument + 1]);
    if (!socket.waitForConnected(5000))
        return 121;
    // The socket is the only resource this fixture needs. Use the same
    // fail-closed OS confinement as the real worker, with no file grants.
    QString sandboxError;
    if (!enterSandbox({}, {}, &sandboxError))
        return 122;
    IpcChannel incoming(&socket);
    QObject::connect(&socket, &QLocalSocket::disconnected, &app, &QCoreApplication::quit);
    QObject::connect(&incoming, &IpcChannel::protocolError, &app, [] { std::_Exit(123); });
    int received = 0;
    bool streaming = false;
    QCborMap fontParent;
    QString selectedFontId;
    QByteArray fontBytes;
    const auto fontRequest = [&](const QString &operation, const QCborMap &payload) {
        auto request = fontParent;
        request.insert(QStringLiteral("operation"), operation);
        request.insert(QStringLiteral("payload"), payload);
        return request;
    };
    const auto mapRequest = [&] {
        return fontRequest("font.map", {{"fontRequestId", 1},
                                        {"faceLocal", QByteArray("Helvetica")},
                                        {"weight", 400},
                                        {"italic", false},
                                        {"charset", 0},
                                        {"pitchFamily", 0}});
    };
    QObject::connect(&incoming, &IpcChannel::messageReceived, &app, [&](const QCborMap &request) {
        if (behavior.startsWith("font-") && isFontOperation(request.value("operation").toString())) {
            const auto operation = request.value("operation").toString();
            const auto result = request.value("result").toMap();
            if (behavior == "font-replay") {
                if (!writeRaw(socket, mapRequest()))
                    std::_Exit(125);
                return;
            }
            if (behavior == "font-flow" || behavior == "font-forged-id" || behavior == "font-overread" ||
                behavior == "font-closed-id") {
                if (operation == "font.map") {
                    selectedFontId = result.value("fontId").toString();
                    const auto id = behavior == "font-forged-id" ? QString(32, 'b') : selectedFontId;
                    const auto offset = behavior == "font-overread" ? 3 : 0;
                    if (!writeRaw(socket, fontRequest("font.read", {{"fontRequestId", 2},
                                                                    {"fontId", id},
                                                                    {"offset", offset},
                                                                    {"length", 4}})))
                        std::_Exit(125);
                    return;
                }
                if (operation == "font.read") {
                    fontBytes = result.value("data").toByteArray();
                    if (!writeRaw(socket, fontRequest("font.close",
                                                      {{"fontRequestId", 3}, {"fontId", selectedFontId}})))
                        std::_Exit(125);
                    return;
                }
                if (behavior == "font-closed-id") {
                    if (!writeRaw(socket, fontRequest("font.close",
                                                      {{"fontRequestId", 4}, {"fontId", selectedFontId}})))
                        std::_Exit(125);
                    return;
                }
            }
            if (!writeRaw(socket, replyTo(fontParent, {{"fontResult", result},
                                                       {"fontError", request.value("error").toMap()},
                                                       {"data", fontBytes},
                                                       {"received", received}})))
                std::_Exit(125);
            return;
        }
        ++received;
        const QCborMap result{{"sandboxed", true},
                              {"received", received},
                              {"receivedDuringStream", streaming},
                              {"echo", request.value("payload")}};
        auto reply = replyTo(request, result);
        if (behavior.startsWith("font-") && request.value("operation").toString() == "render") {
            fontParent = request;
            auto font = mapRequest();
            if (behavior == "font-wrong-parent")
                font.insert(QStringLiteral("requestId"), request.value("requestId").toInteger() + 1);
            if (behavior == "font-wrong-session")
                font.insert(QStringLiteral("sessionId"), "other-session");
            if (behavior == "font-wrong-generation")
                font.insert(QStringLiteral("generation"), request.value("generation").toInteger() + 1);
            if (behavior == "font-bad-schema") {
                auto invalid = font.value("payload").toMap();
                invalid.insert(QStringLiteral("path"), "/secret/font.ttf");
                font.insert(QStringLiteral("payload"), invalid);
            }
            if (!writeRaw(socket, font))
                std::_Exit(125);
            if (behavior == "font-double") {
                auto second = mapRequest();
                auto payload = second.value("payload").toMap();
                payload.insert(QStringLiteral("fontRequestId"), 2);
                second.insert(QStringLiteral("payload"), payload);
                if (!writeRaw(socket, second))
                    std::_Exit(125);
            }
            if (behavior == "font-premature-parent" && !writeRaw(socket, reply))
                std::_Exit(125);
            if (behavior == "font-timeout") {
                QTimer::singleShot(20, &app, [&, request] {
                    auto timeout = request;
                    timeout.remove(QStringLiteral("payload"));
                    timeout.insert(QStringLiteral("error"),
                                   QCborMap{{"code", "E_WORKER_TIMEOUT"}, {"message", "fixture timeout"}});
                    if (!writeRaw(socket, timeout))
                        std::_Exit(125);
                });
            }
            return;
        }
        if (behavior == "crash")
            std::_Exit(73); // Unexpected exit with a live request; no core file.
        if (behavior == "sandbox-unavailable")
            std::_Exit(5); // The PDF worker reports sandbox failure with this code.
        if (behavior == "close-without-reply")
            std::_Exit(0);
        if (behavior == "close-reply-exit" && request.value("operation").toString() == "close") {
            if (!writeRaw(socket, replyTo(request, {{"closed", true}})))
                std::_Exit(125);
            while (socket.bytesToWrite() > 0)
                if (!socket.waitForBytesWritten(1000))
                    std::_Exit(126);
            std::_Exit(0);
        }
        if (behavior == "oversize") {
            char header[4];
            qToBigEndian<quint32>(MaxControlFrame + 1, header);
            if (socket.write(header, 4) != 4)
                std::_Exit(124);
            socket.flush();
            return; // No body: rejection must occur at the length boundary.
        }
        if (behavior == "wrong-session")
            reply.insert(QStringLiteral("sessionId"), QStringLiteral("different-session"));
        else if (behavior == "wrong-generation")
            reply.insert(QStringLiteral("generation"), request.value("generation").toInteger() + 1);
        else if (behavior == "wrong-request")
            reply.insert(QStringLiteral("requestId"), request.value("requestId").toInteger() + 1);
        else if (behavior == "wrong-operation")
            reply.insert(QStringLiteral("operation"), QStringLiteral("metadata"));
        else if (behavior == "wrong-version")
            reply.insert(QStringLiteral("protocolVersion"), 2);
        else if (behavior == "error") {
            reply.remove(QStringLiteral("result"));
            reply.insert(QStringLiteral("error"),
                         QCborMap{{"code", "E_FIXTURE"}, {"message", "fixture error"}});
        } else if (behavior == "illegal-more") {
            reply.insert(QStringLiteral("result"), QCborMap{{"more", true}, {"data", QByteArray("chunk")}});
        } else if (behavior == "stream" && request.value("operation").toString() == "extract") {
            streaming = true;
            if (!writeRaw(
                    socket,
                    replyTo(request, {{"more", true}, {"data", QByteArray("alpha")}, {"sandboxed", true}})))
                std::_Exit(125);
            QTimer::singleShot(30, &app, [&, request] {
                if (!writeRaw(socket, replyTo(request, {{"more", true}, {"data", QByteArray("beta")}})))
                    std::_Exit(125);
            });
            QTimer::singleShot(60, &app, [&, request] {
                streaming = false;
                if (!writeRaw(socket,
                              replyTo(request, {{"more", false}, {"size", 9}, {"received", received}})))
                    std::_Exit(125);
            });
            return;
        }
        if (!writeRaw(socket, reply))
            std::_Exit(125);
    });
    return app.exec();
}

bool launch(WorkerProcess &worker, const QString &behavior) {
    return worker.start(QCoreApplication::applicationFilePath(), {"--fake-worker", behavior});
}

} // namespace
#endif

class WorkerProcessTest : public QObject {
    Q_OBJECT
  private slots:
    void initTestCase() {
#ifndef Q_OS_LINUX
        QSKIP("The production worker sandbox is verified only on Linux; no confinement bypass is used");
#endif
    }

    void successfulEnvelopesPreserveIdentityAndOrder() {
        WorkerProcess worker("broker-session", 37);
        QSignalSpy ready(&worker, &WorkerProcess::ready);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        QList<QCborMap> replies;
        QVERIFY(launch(worker, "echo"));
        // Queue before the asynchronous socket connection is ready.
        worker.request("ping", {{"ordinal", 1}}, [&](const QCborMap &reply) { replies << reply; });
        worker.request("metadata", {{"ordinal", 2}}, [&](const QCborMap &reply) { replies << reply; });
        QTRY_COMPARE_WITH_TIMEOUT(replies.size(), 2, 5000);
        QCOMPARE(ready.size(), 1);
        QVERIFY(worker.isConnected());
        QVERIFY(worker.idle());
        for (int i = 0; i < replies.size(); ++i) {
            const auto reply = replies[i];
            QCOMPARE(reply.value("requestId").toInteger(), i + 1);
            QCOMPARE(reply.value("sessionId").toString(), "broker-session");
            QCOMPARE(reply.value("generation").toInteger(), 37);
            QCOMPARE(reply.value("operation").toString(), i == 0 ? "ping" : "metadata");
            const auto result = reply.value("result").toMap();
            QCOMPARE(result.value("echo").toMap().value("ordinal").toInteger(), i + 1);
            QCOMPARE(result.value("received").toInteger(), i + 1);
            QVERIFY(result.value("sandboxed").toBool());
        }
        worker.stop();
        QCOMPARE(failures.size(), 0);
    }

    void fontReverseRpcPreservesParentAndQueue() {
        WorkerProcess worker("font-broker-session", 5);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        QStringList operations;
        QList<QCborMap> replies;
        worker.setFontRequestHandler([&](const QString &operation, const QCborMap &payload, auto completion) {
            QVERIFY(!payload.contains(QStringLiteral("fontRequestId")));
            QVERIFY(!worker.idle());
            QCOMPARE(replies.size(), 0);
            operations << operation;
            QCborMap result;
            if (operation == "font.map") {
                result = {{"found", true},
                          {"fontId", QString(32, 'a')},
                          {"actualFaceLocal", QByteArray("Selected")},
                          {"charset", 0},
                          {"size", 4},
                          {"sha256", QByteArray(32, 's')},
                          {"faceIndex", 0},
                          {"substituted", true}};
            } else if (operation == "font.read") {
                result = {{"fontId", payload.value("fontId")},
                          {"offset", payload.value("offset")},
                          {"data", QByteArray("font")}};
            } else {
                result = {{"released", true}};
            }
            // A service running on a broker thread may complete off the GUI
            // thread. WorkerProcess must post back before touching its channel.
            std::thread([completion = std::move(completion), result] { completion(result); }).join();
        });
        QVERIFY(launch(worker, "font-flow"));
        worker.request("render", {}, [&](const QCborMap &reply) { replies << reply; });
        worker.request("ping", {}, [&](const QCborMap &reply) { replies << reply; });
        QTRY_COMPARE_WITH_TIMEOUT(replies.size(), 2, 5000);
        QCOMPARE(operations, QStringList({"font.map", "font.read", "font.close"}));
        QCOMPARE(replies[0].value("requestId").toInteger(), 1);
        QCOMPARE(replies[0].value("result").toMap().value("data").toByteArray(), QByteArray("font"));
        QCOMPARE(replies[0].value("result").toMap().value("received").toInteger(), 1);
        QCOMPARE(replies[1].value("result").toMap().value("received").toInteger(), 2);
        QVERIFY(worker.idle());
        QCOMPARE(failures.size(), 0);
    }

    void fontRequestWithoutExplicitServiceFails() {
        WorkerProcess worker("no-font-service", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int replies = 0;
        QVERIFY(launch(worker, "font-map"));
        worker.request("render", {}, [&](const QCborMap &) { ++replies; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(failures[0][0].toString(), "E_WORKER_CRASH");
        QCOMPARE(replies, 0);
    }
    void fontRoleUsesExecutableNameRatherThanParentDirectory() {
        QTemporaryDir directory(QDir::tempPath() + "/docview-archive-font-test-XXXXXX");
        QVERIFY(directory.isValid());
        const auto executable = directory.filePath("docview-pdf-worker");
        QVERIFY(QFile::link(QCoreApplication::applicationFilePath(), executable));
        WorkerProcess worker("font-path-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int replies = 0;
        worker.setFontRequestHandler(
            [](const QString &, const QCborMap &, auto complete) { complete({{"found", false}}); });
        QVERIFY(worker.start(executable, {"--fake-worker", "font-map"}));
        worker.request("render", {}, [&](const QCborMap &) { ++replies; });
        QTRY_COMPARE_WITH_TIMEOUT(replies, 1, 5000);
        QCOMPARE(failures.size(), 0);
    }

    void hostileFontRequest_data() {
        QTest::addColumn<QString>("behavior");
        for (const auto *behavior : {"font-wrong-parent", "font-wrong-session", "font-wrong-generation",
                                     "font-bad-schema", "font-double", "font-premature-parent", "font-replay",
                                     "font-forged-id", "font-overread", "font-closed-id"})
            QTest::newRow(behavior) << QString::fromLatin1(behavior);
    }
    void hostileFontRequest() {
        QFETCH(QString, behavior);
        WorkerProcess worker("hostile-font-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int replies = 0;
        worker.setFontRequestHandler([&](const QString &operation, const QCborMap &payload, auto completion) {
            QCborMap result;
            if (operation == "font.map")
                result = {{"found", true},
                          {"fontId", QString(32, 'a')},
                          {"actualFaceLocal", QByteArray("Selected")},
                          {"charset", 0},
                          {"size", 4},
                          {"sha256", QByteArray(32, 's')},
                          {"faceIndex", 0},
                          {"substituted", true}};
            else if (operation == "font.read")
                result = {{"fontId", payload.value("fontId")},
                          {"offset", payload.value("offset")},
                          {"data", QByteArray("font")}};
            else
                result = {{"released", true}};
            QTimer::singleShot(10, &worker,
                               [completion = std::move(completion), result] { completion(result); });
        });
        QVERIFY(launch(worker, behavior));
        worker.request("render", {}, [&](const QCborMap &) { ++replies; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(failures[0][0].toString(), "E_WORKER_CRASH");
        QCOMPARE(replies, 0);
    }

    void fontCompletionAfterStopAndDestructionIsDiscarded() {
        auto worker = std::make_unique<WorkerProcess>("stopped-font-session", 1);
        QSignalSpy failures(worker.get(), &WorkerProcess::failed);
        std::function<void(QCborMap)> complete;
        int replies = 0;
        worker->setFontRequestHandler(
            [&](const QString &, const QCborMap &, auto completion) { complete = std::move(completion); });
        QVERIFY(launch(*worker, "font-map"));
        worker->request("render", {}, [&](const QCborMap &) { ++replies; });
        QTRY_VERIFY_WITH_TIMEOUT(bool(complete), 5000);
        worker->stop();
        complete({{"found", false}});
        QCoreApplication::processEvents();
        QCOMPARE(replies, 0);
        QCOMPARE(failures.size(), 0);
        worker.reset();
        std::thread([complete] { complete({{"found", false}}); }).join();
        QCoreApplication::processEvents();
        QCOMPARE(replies, 0);
    }

    void parentFontTimeoutDiscardsLateServiceReply() {
        WorkerProcess worker("font-timeout-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        std::function<void(QCborMap)> complete;
        QList<QCborMap> replies;
        worker.setFontRequestHandler(
            [&](const QString &, const QCborMap &, auto completion) { complete = std::move(completion); });
        QVERIFY(launch(worker, "font-timeout"));
        worker.request("render", {}, [&](const QCborMap &reply) { replies << reply; });
        worker.request("ping", {}, [&](const QCborMap &reply) { replies << reply; });
        QTRY_COMPARE_WITH_TIMEOUT(replies.size(), 2, 5000);
        QCOMPARE(replies[0].value("error").toMap().value("code").toString(), "E_WORKER_TIMEOUT");
        QVERIFY(bool(complete));
        complete({{"found", false}});
        QTest::qWait(20);
        QCOMPARE(failures.size(), 0);
        QVERIFY(worker.idle());
    }

    void fontServiceErrorIsReturnedWithSubrequestIdentity() {
        WorkerProcess worker("font-error-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        QList<QCborMap> replies;
        worker.setFontRequestHandler([](const QString &, const QCborMap &, auto complete) {
            complete({{"error", QCborMap{{"code", "E_FONT_LIMIT"}, {"message", "fixture quota"}}}});
        });
        QVERIFY(launch(worker, "font-map"));
        worker.request("render", {}, [&](const QCborMap &reply) { replies << reply; });
        QTRY_COMPARE_WITH_TIMEOUT(replies.size(), 1, 5000);
        const auto error = replies[0].value("result").toMap().value("fontError").toMap();
        QCOMPARE(error.value("code").toString(), "E_FONT_LIMIT");
        QCOMPARE(error.value("fontRequestId").toInteger(), 1);
        QCOMPARE(failures.size(), 0);
    }

    void errorEnvelopeIsDeliveredWithoutBrokerFailure() {
        WorkerProcess worker("error-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        QList<QCborMap> replies;
        QVERIFY(launch(worker, "error"));
        worker.request("open", {}, [&](const QCborMap &reply) { replies << reply; });
        QTRY_COMPARE_WITH_TIMEOUT(replies.size(), 1, 5000);
        QCOMPARE(replies.front().value("error").toMap().value("code").toString(), "E_FIXTURE");
        QVERIFY(worker.idle());
        QCOMPARE(failures.size(), 0);
    }

    void closeAcknowledgementIsTerminal() {
        WorkerProcess worker("close-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int closed = 0, afterClose = 0;
        QVERIFY(launch(worker, "close-reply-exit"));
        worker.request("close", {}, [&](const QCborMap &reply) {
            QVERIFY(reply.value("result").toMap().value("closed").toBool());
            ++closed;
            worker.request("ping", {}, [&](const QCborMap &) { ++afterClose; });
        });
        worker.request("metadata", {}, [&](const QCborMap &) { ++afterClose; });
        QTRY_COMPARE_WITH_TIMEOUT(closed, 1, 5000);
        QTest::qWait(100);
        QCOMPARE(failures.size(), 0);
        QCOMPARE(afterClose, 0);
        QVERIFY(worker.idle());
        QVERIFY(!worker.isConnected());
    }

    void closeExitWithoutAcknowledgementFails() {
        WorkerProcess worker("missing-close-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int closed = 0;
        QVERIFY(launch(worker, "close-without-reply"));
        worker.request("close", {}, [&](const QCborMap &) { ++closed; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(failures.first()[0].toString(), "E_WORKER_CRASH");
        QCOMPARE(closed, 0);
        QVERIFY(worker.idle());
    }

    void hostileEnvelope_data() {
        QTest::addColumn<QString>("behavior");
        for (const auto *behavior : {"wrong-session", "wrong-generation", "wrong-request", "wrong-operation",
                                     "wrong-version", "oversize", "illegal-more"})
            QTest::newRow(behavior) << QString::fromLatin1(behavior);
    }

    void hostileEnvelope() {
        QFETCH(QString, behavior);
        WorkerProcess worker("correct-session", 7);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int callbacks = 0;
        QVERIFY(launch(worker, behavior));
        worker.request("ping", {}, [&](const QCborMap &) { ++callbacks; });
        worker.request("metadata", {}, [&](const QCborMap &) { ++callbacks; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(failures.front()[0].toString(), "E_WORKER_CRASH");
        QVERIFY(!failures.front()[1].toString().isEmpty());
        QCOMPARE(callbacks, 0);
        QVERIFY(worker.idle());
        QTest::qWait(100); // Process termination must not emit a second failure.
        QCOMPARE(failures.size(), 1);
        QCOMPARE(callbacks, 0);
    }

    void unexpectedExitWhileActiveFailsOnce() {
        WorkerProcess worker("crash-session", 1);
        QSignalSpy ready(&worker, &WorkerProcess::ready);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int callbacks = 0;
        QVERIFY(launch(worker, "crash"));
        worker.request("render", {}, [&](const QCborMap &) { ++callbacks; });
        worker.request("pages", {}, [&](const QCborMap &) { ++callbacks; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(ready.size(), 1);
        QCOMPARE(failures.front()[0].toString(), "E_WORKER_CRASH");
        QTest::qWait(100);
        QCOMPARE(failures.size(), 1);
        QCOMPARE(callbacks, 0);
        QVERIFY(worker.idle());
    }

    void unavailableSandboxHasSpecificError() {
        WorkerProcess worker("sandbox-session", 1);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int callbacks = 0;
        QVERIFY(launch(worker, "sandbox-unavailable"));
        worker.request("open", {}, [&](const QCborMap &) { ++callbacks; });
        QTRY_COMPARE_WITH_TIMEOUT(failures.size(), 1, 5000);
        QCOMPARE(failures.front()[0].toString(), "E_SANDBOX_UNAVAILABLE");
        QVERIFY(failures.front()[1].toString().contains("保護"));
        QTest::qWait(100);
        QCOMPARE(failures.size(), 1);
        QCOMPARE(callbacks, 0);
        QVERIFY(worker.idle());
    }

    void extractChunksKeepRequestActiveUntilFinal() {
        WorkerProcess worker("stream-session", 3);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        QStringList delivered;
        QByteArray content;
        bool final = false;
        QVERIFY(launch(worker, "stream"));
        worker.request("extract", {{"path", "chapter.xhtml"}}, [&](const QCborMap &reply) {
            const auto result = reply.value("result").toMap();
            QCOMPARE(reply.value("requestId").toInteger(), 1);
            if (result.value("more").toBool()) {
                QVERIFY(!worker.idle());
                QVERIFY(!final);
                content += result.value("data").toByteArray();
                delivered << "chunk";
            } else {
                final = true;
                QCOMPARE(result.value("size").toInteger(), 9);
                QCOMPARE(result.value("received").toInteger(), 1);
                delivered << "final";
            }
        });
        worker.request("ping", {}, [&](const QCborMap &reply) {
            QVERIFY(final);
            const auto result = reply.value("result").toMap();
            QVERIFY(!result.value("receivedDuringStream").toBool());
            QCOMPARE(result.value("received").toInteger(), 2);
            delivered << "ping";
        });
        QTRY_COMPARE_WITH_TIMEOUT(delivered.size(), 4, 5000);
        QCOMPARE(delivered, QStringList({"chunk", "chunk", "final", "ping"}));
        QCOMPARE(content, "alphabeta");
        QVERIFY(worker.idle());
        QCOMPARE(failures.size(), 0);
    }

    void stopDropsActiveAndQueuedReplies() {
        WorkerProcess worker("stopped-session", 3);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int activeCallbacks = 0, queuedCallbacks = 0;
        QVERIFY(launch(worker, "stream"));
        worker.request("extract", {}, [&](const QCborMap &reply) {
            ++activeCallbacks;
            QVERIFY(reply.value("result").toMap().value("more").toBool());
            QCOMPARE(worker.activeRequestCount(), 1);
            QCOMPARE(worker.activeRequestCount("extract"), 1);
            QCOMPARE(worker.activeRequestCount("ping"), 0);
            QCOMPARE(worker.queuedRequestCount(), 1);
            QCOMPARE(worker.queuedRequestCount("ping"), 1);
            worker.stop();
            QCOMPARE(worker.activeRequestCount(), 0);
            QCOMPARE(worker.queuedRequestCount(), 0);
            QCOMPARE(worker.discardedQueuedRequestCount(), quint64(1));
            worker.stop(); // Repeated stop cannot count the same cancellation twice.
            QCOMPARE(worker.discardedQueuedRequestCount(), quint64(1));
        });
        worker.request("ping", {}, [&](const QCborMap &) { ++queuedCallbacks; });
        QTRY_COMPARE_WITH_TIMEOUT(activeCallbacks, 1, 5000);
        worker.request("metadata", {}, [&](const QCborMap &) { ++queuedCallbacks; });
        QTest::qWait(150);
        QCOMPARE(activeCallbacks, 1);
        QCOMPARE(queuedCallbacks, 0);
        QCOMPARE(failures.size(), 0);
        QVERIFY(worker.idle());
        QCOMPARE(worker.discardedQueuedRequestCount(), quint64(1));
    }

    void discardedQueuedRequestsNeverReachWorker_data() {
        QTest::addColumn<bool>("selective");
        QTest::newRow("operation-only") << true;
        QTest::newRow("all-queued") << false;
    }

    void discardedQueuedRequestsNeverReachWorker() {
        QFETCH(bool, selective);
        WorkerProcess worker("discard-session", 3);
        QSignalSpy failures(&worker, &WorkerProcess::failed);
        int activeCallbacks = 0, droppedCallbacks = 0, keptCallbacks = 0;
        bool final = false;
        const auto kept = [&](const QCborMap &reply) {
            ++keptCallbacks;
            QVERIFY(final);
            QCOMPARE(reply.value("result").toMap().value("received").toInteger(), 2);
        };
        QVERIFY(launch(worker, "stream"));
        worker.request("extract", {}, [&](const QCborMap &reply) {
            ++activeCallbacks;
            if (activeCallbacks == 1) {
                QCOMPARE(worker.activeRequestCount("extract"), 1);
                QCOMPARE(worker.queuedRequestCount(), 3);
                QCOMPARE(worker.queuedRequestCount("render"), 2);
                QCOMPARE(worker.queuedRequestCount("metadata"), 1);
                QCOMPARE(worker.discardedQueuedRequestCount(), quint64(0));
                worker.discardQueued(selective ? "render" : QString{});
                QCOMPARE(worker.queuedRequestCount(), selective ? 1 : 0);
                QCOMPARE(worker.queuedRequestCount("render"), 0);
                QCOMPARE(worker.activeRequestCount("extract"), 1);
                QCOMPARE(worker.discardedQueuedRequestCount(), quint64(selective ? 2 : 3));
                if (!selective)
                    worker.request("ping", {}, kept);
            }
            if (!reply.value("result").toMap().value("more").toBool())
                final = true;
        });
        worker.request("render", {{"page", 1}}, [&](const QCborMap &) { ++droppedCallbacks; });
        worker.request("render", {{"page", 2}}, [&](const QCborMap &) { ++droppedCallbacks; });
        worker.request("metadata", {}, [&](const QCborMap &reply) {
            if (selective)
                kept(reply);
            else
                ++droppedCallbacks;
        });
        QTRY_COMPARE_WITH_TIMEOUT(keptCallbacks, 1, 5000);
        QCOMPARE(activeCallbacks, 3);
        QCOMPARE(droppedCallbacks, 0);
        QVERIFY(worker.idle());
        QCOMPARE(worker.activeRequestCount(), 0);
        QCOMPARE(worker.queuedRequestCount(), 0);
        QCOMPARE(worker.discardedQueuedRequestCount(), quint64(selective ? 2 : 3));
        QTest::qWait(100);
        QCOMPARE(droppedCallbacks, 0);
        QCOMPARE(failures.size(), 0);
    }
};

int main(int argc, char **argv) {
    QCoreApplication app(argc, argv);
    if (app.arguments().contains("--fake-worker"))
        return fakeWorker(app);
    WorkerProcessTest test;
    return QTest::qExec(&test, argc, argv);
}

#include "test_worker_process.moc"