testScheduler.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdpm
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief Unit tests for daq::dpm::SchedulerImpl
 */
 
#include <daq/dpm/scheduler.hpp>
 
#include <utility>
 
#include <daq/json.hpp>
 
#include "mock/mockScheduler.hpp"
#include "mock/mockWorkspace.hpp"
 
#include <gmock/gmock.h>
#include <gtest/gtest.h>
 
namespace daq::dpm {
 
struct FakeDaqControllerFactory {
    using Hook = std::function<void(MockDaqController&)>;
    std::unique_ptr<DaqController> operator()(std::unique_ptr<DaqWorkspace>, Resources&) {
        auto daq = std::make_unique<MockDaqController>();
        if (hook) {
            hook(*daq);
        }
        daq_controllers.push_back(daq.get());
        return daq;
    }
    void SetHook(Hook f) {
        hook = std::move(f);
    }
    std::vector<DaqController*> daq_controllers;
    Hook hook;
};
 
class TestSchedulerBase : public ::testing::Test {
public:
    TestSchedulerBase() : m_io_ctx(), m_executor(m_io_ctx) {
    }
 
    void SetUp() override {
    }
    void TearDown() override {
        // Execute possibly pending completions handlers.
        EXPECT_NO_THROW(m_io_ctx.poll());
    }
 
    auto SerializeStatus(Status const& status) const -> std::string {
        nlohmann::json j;
        j = status;
        return j.dump(2);
    }
 
protected:
    boost::asio::io_context m_io_ctx;
    rad::IoExecutor m_executor;
    std::vector<std::string> m_queue;
 
    daq::dpm::MockWorkspace m_ws_mock;
};
 
class TestSchedulerInit : public TestSchedulerBase {};
 
class TestScheduler : public TestSchedulerBase {
public:
    TestScheduler() : m_status("TEST.ID", "TEST.FILEID") {
    }
    void SetUp() override {
        TestSchedulerBase::SetUp();
        m_spec_str = R"(
            {
              "id": "TEST.ID",
              "target": {
                "fileId": "TEST.FILEID",
                "source": {
                  "sourceName": "dcs",
                  "location": "dcs-host:/path/to/somefile.fits",
                  "path": "dcs/somefile.fits"
                }
              },
              "sources": [
                {
                  "type": "fitsFile",
                  "sourceName": "fcs",
                  "location": "fcs-host:/path/to/somefile.fits",
                  "path": "fcs/somefile.fits"
                }
              ],
              "receivers": [
                  {
                    "type": "olasReceiver",
                    "host": "1.2.3.4",
                    "path": "/remote/olas/"
                  },
                  {
                    "type": "olasReceiver",
                    "path": "/local/olas/"
                  }
              ]
            }
        )";
    }
    void PostSetup() {
        EXPECT_CALL(m_ws_mock, LoadQueue()).WillOnce(::testing::Return(m_queue));
 
        m_scheduler = std::make_unique<SchedulerImpl>(
            m_executor, m_ws_mock, std::reference_wrapper(m_daq_controller_factory), m_options);
    }
    void TearDown() override {
        m_scheduler.reset();
        TestSchedulerBase::TearDown();
    }
 
protected:
    SchedulerOptions m_options;
    std::unique_ptr<SchedulerImpl> m_scheduler;
    std::string m_spec_str;
    FakeDaqControllerFactory m_daq_controller_factory;
    ObservableStatus m_status;
};
 
using namespace ::testing;
 
TEST_F(TestSchedulerInit, Construction) {
    SchedulerOptions m_options;
    EXPECT_CALL(m_ws_mock, LoadQueue()).WillOnce(Return(m_queue));
    EXPECT_NO_THROW(SchedulerImpl(m_executor, m_ws_mock, FakeDaqControllerFactory(), m_options));
}
 
TEST_F(TestScheduler, QueueDaqIsSuccessful) {
    // Setup
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
 
    EXPECT_CALL(*daq_ws, StoreStatus(status));
    EXPECT_CALL(*daq_ws, StoreSpecification(m_spec_str));
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>{"TEST.ID"}));
 
    PostSetup();
 
    // Test
    m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status));
    EXPECT_TRUE(m_scheduler->IsQueued("TEST.ID"));
 
    // Queue same DAQ should fail
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status)), std::invalid_argument);
    EXPECT_TRUE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, QueueDaqRollbackIfStoreQueueFails) {
    // Setup
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
 
    Sequence init, rollback;
 
    EXPECT_CALL(m_ws_mock, GetPath()).Times(AnyNumber()).WillRepeatedly(Return("/tmp/workspace"));
 
    EXPECT_CALL(*daq_ws, StoreStatus(status));
    EXPECT_CALL(*daq_ws, StoreSpecification(m_spec_str));
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID"))
        .InSequence(init, rollback)
        .WillOnce(Return(ByMove(std::move(daq_ws))));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>{"TEST.ID"}))
        .InSequence(init, rollback)
        .WillRepeatedly(Throw(std::runtime_error("FAILED")));
 
    // Rollback
    EXPECT_CALL(m_ws_mock, RemoveDaq("TEST.ID")).InSequence(rollback);
 
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status)), std::exception);
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, QueueDaqRollbackIfStoreSpecificationFails) {
    // Setup
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
 
    Sequence init, rollback;
 
    EXPECT_CALL(m_ws_mock, GetPath()).Times(AnyNumber()).WillRepeatedly(Return("/tmp/workspace"));
 
    EXPECT_CALL(*daq_ws, StoreStatus(status));
    EXPECT_CALL(*daq_ws, StoreSpecification(m_spec_str))
        .WillOnce(Throw(std::runtime_error("ERROR")));
 
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID"))
        .InSequence(init, rollback)
        .WillOnce(Return(ByMove(std::move(daq_ws))));
 
    // Rollback
    EXPECT_CALL(m_ws_mock, RemoveDaq("TEST.ID")).InSequence(rollback);
 
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status)), std::exception);
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, QueueDaqRollbackIfInitializeDaqWorkspaceFails) {
    // Setup
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
    Sequence init, rollback;
    EXPECT_CALL(m_ws_mock, GetPath()).Times(AnyNumber()).WillRepeatedly(Return("/tmp/workspace"));
 
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID"))
        .InSequence(init, rollback)
        .WillRepeatedly(Throw(std::runtime_error("ERROR")));
 
    // Rollback
    EXPECT_CALL(m_ws_mock, RemoveDaq("TEST.ID")).InSequence(rollback);
 
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status)), std::exception);
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, QueueDaqFailsIfParseDpFails) {
    // Setup
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
    Sequence init, rollback;
    EXPECT_CALL(m_ws_mock, GetPath()).Times(AnyNumber()).WillRepeatedly(Return("/tmp/workspace"));
 
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID")).Times(0);
 
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->QueueDaq("not a specification", SerializeStatus(status)),
                 std::invalid_argument);
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, QueueDaqFailsIfStatusParsingFails) {
    // Setup
    Sequence init, rollback;
    EXPECT_CALL(m_ws_mock, GetPath()).Times(AnyNumber()).WillRepeatedly(Return("/tmp/workspace"));
 
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID")).Times(0);
 
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, "not a status"), std::invalid_argument);
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, PollActivatesDaqWhichStartsTransfer) {
    // Setup
    PostSetup();
    m_scheduler->Start();
 
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
 
    EXPECT_CALL(*daq_ws, StoreStatus(status));
    EXPECT_CALL(*daq_ws, StoreSpecification(m_spec_str));
    EXPECT_CALL(m_ws_mock, InitializeDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>{"TEST.ID"}));
 
    m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status));
    EXPECT_TRUE(m_scheduler->IsQueued("TEST.ID"));
 
    // Queue same DAQ should fail
    EXPECT_THROW(m_scheduler->QueueDaq(m_spec_str, SerializeStatus(status)), std::invalid_argument);
    EXPECT_TRUE(m_scheduler->IsQueued("TEST.ID"));
 
    // We expect Poll() should load from workspace and create DaqController from factory
    // and then start that.
    daq_ws = std::make_unique<MockDaqWorkspace>();
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
    m_daq_controller_factory.SetHook([&](MockDaqController& mock) {
        EXPECT_CALL(mock, GetStatus()).WillOnce(ReturnRef(m_status));
        EXPECT_CALL(mock, GetId()).WillRepeatedly(ReturnRef(m_status.GetId()));
        EXPECT_CALL(mock, GetState()).WillRepeatedly(Return(State::Collecting));
        EXPECT_CALL(mock, Start());
    });
 
    // Test[
    m_io_ctx.poll();
}
 
TEST_F(TestScheduler, PollCompletesDaq) {
    // Setup
    m_queue = {"TEST.ID"};
    PostSetup();
 
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
 
    m_status.SetState(State::Collecting);
    EXPECT_TRUE(m_scheduler->IsQueued("TEST.ID"));
 
    // We expect Poll() should load from workspace
    daq_ws = std::make_unique<MockDaqWorkspace>();
 
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
    EXPECT_CALL(m_ws_mock, ArchiveDaq("TEST.ID"));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>()));
 
    m_daq_controller_factory.SetHook([&](MockDaqController& mock) {
        EXPECT_CALL(mock, GetStatus()).WillOnce(ReturnRef(m_status));
        EXPECT_CALL(mock, GetId()).WillRepeatedly(ReturnRef(m_status.GetId()));
        EXPECT_CALL(mock, GetState())
            .WillRepeatedly(Invoke(&m_status, &ObservableStatus::GetState));
 
        // When Start() is invoked on DaqController we set the state to Completed
        // so that it will be recognized as completed and archived (Scheduler should monitor
        // DaqControllers via signals).
        EXPECT_CALL(mock, Start()).WillOnce(Invoke([&]() { m_status.SetState(State::Completed); }));
    });
 
    // Test
    // First poll initiates all operations.
    m_scheduler->Start();
    m_io_ctx.poll();
}
 
TEST_F(TestScheduler, AbortDaqSucceedsIfNotActive) {
    // Setup
    m_queue = {"TEST.ID"};
 
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
    Status status = {};
    status.id = "TEST.ID";
    status.file_id = "TEST.FILEID";
    status.state = State::Scheduled;
 
    EXPECT_CALL(*daq_ws, LoadStatus()).WillOnce(Return(status));
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>{}));
    EXPECT_CALL(m_ws_mock, RemoveDaq("TEST.ID"));
 
    PostSetup();
 
    // Test
    ASSERT_TRUE(m_scheduler->IsQueued(status.id));
    m_scheduler->AbortDaq(status.id);
    EXPECT_FALSE(m_scheduler->IsQueued(status.id));
}
 
TEST_F(TestScheduler, AbortDaqSucceedsIfActive) {
    // Setup
    m_queue = {"TEST.ID"};
    m_status.SetState(State::Scheduled);
    auto daq_ws_init = std::make_unique<MockDaqWorkspace>();
    auto daq_ws_abort = std::make_unique<MockDaqWorkspace>();
 
    EXPECT_CALL(*daq_ws_init, LoadStatus()).WillRepeatedly(Return(m_status.GetStatus()));
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws_init))));
 
    m_daq_controller_factory.SetHook([&](MockDaqController& mock) {
        EXPECT_CALL(mock, GetStatus()).WillOnce(ReturnRef(m_status));
        EXPECT_CALL(mock, GetId()).WillRepeatedly(ReturnRef(m_status.GetId()));
        EXPECT_CALL(mock, GetState()).WillRepeatedly(Return(State::Collecting));
        EXPECT_CALL(mock, Start());
    });
 
    PostSetup();
 
    // Poll to activate DAQ
    m_scheduler->Start();
    m_io_ctx.poll();
 
    EXPECT_CALL(*daq_ws_abort, LoadStatus()).WillRepeatedly(Return(m_status.GetStatus()));
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws_abort))));
    EXPECT_CALL(m_ws_mock, RemoveDaq("TEST.ID"));
    EXPECT_CALL(m_ws_mock, StoreQueue(std::vector<std::string>{}));
 
    // Test
    ASSERT_TRUE(m_scheduler->IsQueued("TEST.ID"));
    m_scheduler->AbortDaq("TEST.ID");
    EXPECT_FALSE(m_scheduler->IsQueued("TEST.ID"));
}
 
TEST_F(TestScheduler, AbortDaqFailsIfDaqDoesNotExit) {
    // Setup
    PostSetup();
 
    // Test
    EXPECT_THROW(m_scheduler->AbortDaq("UNKOWN"), std::runtime_error);
}
 
TEST_F(TestScheduler, GetStatusSucceeds) {
    // Setup
    m_queue = {"TEST.ID"};
    auto daq_ws = std::make_unique<MockDaqWorkspace>();
 
    EXPECT_CALL(*daq_ws, LoadStatus()).WillRepeatedly(Return(m_status.GetStatus()));
    EXPECT_CALL(m_ws_mock, LoadDaq("TEST.ID")).WillOnce(Return(ByMove(std::move(daq_ws))));
 
    PostSetup();
 
    // Test
    ASSERT_TRUE(m_scheduler->IsQueued("TEST.ID"));
    auto status = m_scheduler->GetDaqStatus("TEST.ID");
    EXPECT_EQ(status.id, "TEST.ID");
    EXPECT_EQ(status.state, m_status.GetState());
}
 
TEST_F(TestScheduler, GetStatusSucceedsForArchivedDaq) {
    // Setup
    m_queue = {"TEST.ID"};
    Status archived_status("TEST.ID.ARCHIVED", "TEST.ID.ARCHIVED");
 
    EXPECT_CALL(m_ws_mock, LoadArchivedStatus("TEST.ID.ARCHIVED"))
        .WillOnce(Return(archived_status));
 
    PostSetup();
 
    // Test
    ASSERT_FALSE(m_scheduler->IsQueued("TEST.ID.ARCHIVED"));
    auto status = m_scheduler->GetDaqStatus("TEST.ID.ARCHIVED");
    EXPECT_EQ(status, archived_status);
}
 
}  // namespace daq::dpm