Jungfraujoch/tests/BrokerHttpAuthTest.cpp

// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#include <catch2/catch_all.hpp>
#include <httplib.h>
#include <nlohmann/json.hpp>
#include <chrono>
#include <thread>

#include "../broker/JFJochBrokerHttp.h"
#include "../common/DetectorModules.h"

// The tokens given to /start guard the endpoints that expose the dataset, and nothing else, for as
// long as that dataset is the current one. Runs the real HTTP layer against a broker with no
// receiver: /start is then accepted and finishes at once with nothing to acquire.
TEST_CASE("JFJochBrokerHttp_BearerTokens", "[broker]") {
    DiffractionExperiment experiment;
    SpotFindingSettings spot_finding;
    JFJochBrokerHttp broker(experiment, spot_finding);
    broker.AddDetectorSetup(DetJF4M());

    httplib::Server server;
    broker.attach(server);
    const int port = server.bind_to_any_port("127.0.0.1");
    REQUIRE(port > 0);
    // Stopped and joined however the test ends, so a failed REQUIRE does not abort on a joinable thread.
    struct ServerThread {
        httplib::Server &server;
        std::thread thread;
        explicit ServerThread(httplib::Server &s) : server(s), thread([&s] { s.listen_after_bind(); }) {}
        ~ServerThread() { server.stop(); thread.join(); }
    } server_thread(server);
    server.wait_until_ready();

    httplib::Client client("127.0.0.1", port);
    const httplib::Headers good{{"Authorization", "Bearer secret-1"}};
    const httplib::Headers other{{"Authorization", "Bearer secret-2"}};
    const httplib::Headers wrong{{"Authorization", "Bearer nope"}};

    auto start = [&](const nlohmann::json &extra) {
        nlohmann::json body = {{"beam_x_pxl", 1000}, {"beam_y_pxl", 1000},
                               {"detector_distance_mm", 100}, {"incident_energy_keV", 12.4},
                               {"images_per_trigger", 1}, {"ntrigger", 1},
                               {"file_prefix", "protected_run"}};
        body.update(extra);
        auto res = client.Post("/start", body.dump(), "application/json");
        REQUIRE(res);
        REQUIRE(res->status == 200);
        // With no receiver the measurement thread idles for 30 s before the run ends; wait it out.
        for (int i = 0; i < 120; i++) {
            auto status = client.Get("/status");
            REQUIRE(status);
            if (nlohmann::json::parse(status->body).at("state") == "Idle")
                return;
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
        FAIL("broker did not return to Idle after the start");
    };

    REQUIRE(client.Post("/initialize")->status == 200);

    // Nothing started yet: open, whatever the header says.
    REQUIRE(client.Get("/statistics/data_collection")->status == 200);
    REQUIRE(client.Get("/statistics/data_collection", wrong)->status == 200);
    REQUIRE(client.Get("/statistics")->status == 200);

    start({{"tokens", {"secret-1", "secret-2"}}});

    SECTION("protected endpoints refuse without a token and say nothing about the dataset") {
        for (const char *path: {"/statistics/data_collection", "/result/scan", "/image_buffer/start.cbor",
                                "/image_buffer/image.cbor?id=0", "/image_buffer/image.tiff?id=0",
                                "/image_buffer/image.jpeg?id=0", "/preview/plot?type=bkg_estimate",
                                "/preview/plot.bin?type=bkg_estimate", "/live/events",
                                "/rugnux/status", "/rugnux/model.cif", "/rugnux/map.ccp4?type=2fofc"}) {
            INFO(path);
            auto res = client.Get(path);
            REQUIRE(res);
            REQUIRE(res->status == 401);
            REQUIRE(res->get_header_value("WWW-Authenticate") == "Bearer");
            REQUIRE(res->body.find("protected_run") == std::string::npos);
            REQUIRE(client.Get(path, wrong)->status == 401);
        }
        // Live rugnux settings and its model prepare the next collection: open like the rest of the configuration.
        REQUIRE(client.Get("/config/rugnux")->status == 200);
        REQUIRE(client.Put("/config/rugnux", R"({"enable":false})", "application/json")->status == 200);
        REQUIRE(client.Put("/config/rugnux/model", "", "application/octet-stream")->status == 400);
        REQUIRE(client.Delete("/config/rugnux/model")->status == 200);
    }

    SECTION("either token opens them") {
        auto res = client.Get("/statistics/data_collection", good);
        REQUIRE(res->status == 200);
        REQUIRE(res->body.find("protected_run") != std::string::npos);
        REQUIRE(client.Get("/statistics/data_collection", other)->status == 200);
        REQUIRE(client.Get("/preview/plot?type=bkg_estimate", good)->status == 200);
        REQUIRE(client.Get("/rugnux/status", good)->status == 200);
        // No live run placed a model: nothing to serve, but the token is accepted.
        REQUIRE(client.Get("/rugnux/model.cif", good)->status == 404);
        REQUIRE(client.Get("/rugnux/map.ccp4", good)->status == 404);
        REQUIRE(client.Get("/rugnux/map.ccp4?type=fofc", good)->status == 404);
        REQUIRE(client.Get("/rugnux/map.ccp4?type=anom", good)->status == 404);
        REQUIRE(client.Get("/rugnux/map.ccp4?type=anomalous", good)->status == 400);
    }

    SECTION("the aggregate statistics drop the measurement block instead of refusing") {
        auto res = client.Get("/statistics");
        REQUIRE(res->status == 200);
        REQUIRE_FALSE(nlohmann::json::parse(res->body).contains("measurement"));
        REQUIRE(res->body.find("protected_run") == std::string::npos);

        auto with_token = client.Get("/statistics", good);
        REQUIRE(with_token->status == 200);
        REQUIRE(nlohmann::json::parse(with_token->body).contains("measurement"));
    }

    SECTION("with no receiver there is no collection to follow") {
        REQUIRE(client.Get("/live/events", good)->status == 409);
    }

    SECTION("initialize drops the tokens and everything they protected") {
        REQUIRE(client.Post("/initialize")->status == 200);
        for (int i = 0; i < 120 && nlohmann::json::parse(client.Get("/status")->body).at("state") != "Idle"; i++)
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        auto res = client.Get("/statistics/data_collection");
        REQUIRE(res->status == 200);
        REQUIRE(res->body.find("protected_run") == std::string::npos);
        auto stats = client.Get("/statistics");
        REQUIRE(stats->status == 200);
        REQUIRE(stats->body.find("protected_run") == std::string::npos);
        REQUIRE(client.Get("/rugnux/status")->status == 200);
        REQUIRE(client.Get("/result/scan")->status != 401);
    }

    SECTION("the control plane and the broker status stay open") {
        REQUIRE(client.Get("/status")->status == 200);
        REQUIRE(client.Get("/image_buffer/status")->status == 200);
        REQUIRE(client.Get("/config/detector")->status == 200);
    }

    SECTION("a refused start keeps the current tokens, an accepted one replaces them") {
        // Not idle: /start is refused, the dataset stays protected by its own tokens.
        REQUIRE(client.Post("/deactivate")->status == 200);
        nlohmann::json body = {{"beam_x_pxl", 1000}, {"beam_y_pxl", 1000},
                               {"detector_distance_mm", 100}, {"incident_energy_keV", 12.4}};
        REQUIRE(client.Post("/start", body.dump(), "application/json")->status == 500);
        REQUIRE(client.Get("/statistics/data_collection")->status == 401);
        REQUIRE(client.Get("/statistics/data_collection", good)->status == 200);

        // Back from Inactive; the next start without tokens leaves the dataset open.
        REQUIRE(client.Post("/initialize")->status == 200);
        start(nlohmann::json::object());
        REQUIRE(client.Get("/statistics/data_collection")->status == 200);
        REQUIRE(client.Get("/statistics/data_collection", wrong)->status == 200);

        // And with new tokens, the old ones no longer open it.
        start({{"tokens", {"secret-3"}}});
        const httplib::Headers newer{{"Authorization", "Bearer secret-3"}};
        REQUIRE(client.Get("/statistics/data_collection", good)->status == 401);
        REQUIRE(client.Get("/statistics/data_collection", newer)->status == 200);
    }

}

// A module placement that does not name exactly the detector's modules is the request's fault: /start
// answers 400, whether a module is left out, placed twice or not the detector's at all.
TEST_CASE("JFJochBrokerHttp_StartModulePlacementMismatch", "[broker]") {
    DiffractionExperiment experiment;
    SpotFindingSettings spot_finding;
    JFJochBrokerHttp broker(experiment, spot_finding);
    broker.AddDetectorSetup(DetJF4M());

    httplib::Server server;
    broker.attach(server);
    const int port = server.bind_to_any_port("127.0.0.1");
    REQUIRE(port > 0);
    struct ServerThread {
        httplib::Server &server;
        std::thread thread;
        explicit ServerThread(httplib::Server &s) : server(s), thread([&s] { s.listen_after_bind(); }) {}
        ~ServerThread() { server.stop(); thread.join(); }
    } server_thread(server);
    server.wait_until_ready();

    httplib::Client client("127.0.0.1", port);
    REQUIRE(client.Post("/initialize")->status == 200);

    const auto assembled = ModulesAsImaged(DetJF4M().GetGeometry()).Get();
    REQUIRE(assembled.size() == 8);
    const auto entry = [](const DetectorModule &m) {
        const ModulePlacement p = PlacementOf(m);
        return nlohmann::json{{"x0", p.x0}, {"y0", p.y0},
                              {"fast_axis", DirectionName(p.fast)}, {"slow_axis", DirectionName(p.slow)},
                              {"x0_plane", p.x0_plane}, {"y0_plane", p.y0_plane},
                              {"fast_step", {p.fast_step[0], p.fast_step[1]}},
                              {"slow_step", {p.slow_step[0], p.slow_step[1]}}};
    };
    const auto start = [&](const nlohmann::json &placement) {
        nlohmann::json body = {{"beam_x_pxl", 1000}, {"beam_y_pxl", 1000},
                               {"detector_distance_mm", 100}, {"incident_energy_keV", 12.4},
                               {"images_per_trigger", 1}, {"ntrigger", 1},
                               {"file_prefix", "placement_run"}, {"module_placement", placement}};
        auto res = client.Post("/start", body.dump(), "application/json");
        REQUIRE(res);
        return res->status;
    };

    nlohmann::json missing = nlohmann::json::array();
    for (size_t m = 1; m < assembled.size(); m++)
        missing.push_back(entry(assembled[m]));
    CHECK(start(missing) == 400);

    nlohmann::json twice = missing;
    twice.push_back(entry(assembled[1]));
    CHECK(start(twice) == 400);

    nlohmann::json elsewhere = entry(assembled[0]);
    elsewhere["y0"] = elsewhere["y0"].get<int64_t>() + 1;
    nlohmann::json unknown = missing;
    unknown.push_back(elsewhere);
    CHECK(start(unknown) == 400);

    // Nothing was started.
    CHECK(nlohmann::json::parse(client.Get("/status")->body).at("state") == "Idle");
}

// The online rotation-indexing settings and quick_integration are deprecated: a request may leave them
// out or set them, and they come back at their old defaults either way.
TEST_CASE("JFJochBrokerHttp_DeprecatedAnalysisSettings", "[broker]") {
    DiffractionExperiment experiment;
    SpotFindingSettings spot_finding;
    JFJochBrokerHttp broker(experiment, spot_finding);
    broker.AddDetectorSetup(DetJF4M());

    httplib::Server server;
    broker.attach(server);
    const int port = server.bind_to_any_port("127.0.0.1");
    REQUIRE(port > 0);
    struct ServerThread {
        httplib::Server &server;
        std::thread thread;
        explicit ServerThread(httplib::Server &s) : server(s), thread([&s] { s.listen_after_bind(); }) {}
        ~ServerThread() { server.stop(); thread.join(); }
    } server_thread(server);
    server.wait_until_ready();

    httplib::Client client("127.0.0.1", port);
    REQUIRE(client.Post("/initialize")->status == 200);

    nlohmann::json indexing = nlohmann::json::parse(client.Get("/config/indexing")->body);
    indexing.erase("rotation_indexing");
    indexing.erase("rotation_indexing_min_angular_range_deg");
    indexing.erase("rotation_indexing_angular_stride_deg");
    CHECK(client.Put("/config/indexing", indexing.dump(), "application/json")->status == 200);
    indexing["rotation_indexing"] = true;
    indexing["rotation_indexing_min_angular_range_deg"] = 30.0;
    CHECK(client.Put("/config/indexing", indexing.dump(), "application/json")->status == 200);
    indexing = nlohmann::json::parse(client.Get("/config/indexing")->body);
    CHECK(indexing.at("rotation_indexing") == false);
    CHECK(indexing.at("rotation_indexing_min_angular_range_deg") == 20.0);
    CHECK(indexing.at("rotation_indexing_angular_stride_deg") == 0.5);

    nlohmann::json spot = nlohmann::json::parse(client.Get("/config/spot_finding")->body);
    spot.erase("quick_integration");
    CHECK(client.Put("/config/spot_finding", spot.dump(), "application/json")->status == 200);
    spot["quick_integration"] = true;
    CHECK(client.Put("/config/spot_finding", spot.dump(), "application/json")->status == 200);
    spot = nlohmann::json::parse(client.Get("/config/spot_finding")->body);
    CHECK(spot.at("quick_integration") == false);
}