testDaqController.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::DaqControllerImpl
 */
#include <daq/dpm/scheduler.hpp>
 
#include <log4cplus/loggingmacros.h>
#include <utility>
 
#include "mock/mockAsyncProcess.hpp"
#include "mock/mockScheduler.hpp"
#include "mock/mockWorkspace.hpp"
 
#include <gmock/gmock.h>
#include <gtest/gtest.h>
 
namespace daq::dpm {
 
using namespace ::testing;
 
struct FakeRsyncFactory {
    using Hook = std::function<void(MockRsyncAsyncProcess&)>;
    std::unique_ptr<RsyncAsyncProcessIf> operator()(boost::asio::io_context&,
                                                    std::string,
                                                    std::string,
                                                    RsyncOptions const&,
                                                    RsyncAsyncProcess::DryRun) {
        auto rsync = std::make_unique<MockRsyncAsyncProcess>();
        if (hook) {
            hook(*rsync);
        }
        procs.push_back(rsync.get());
        return rsync;
    }
    void SetHook(Hook f) {
        hook = std::move(f);
    }
    std::vector<MockRsyncAsyncProcess*> procs;
    Hook hook;
};
 
struct FakeProcFactory {
    using Hook = std::function<void(MockAsyncProcess&)>;
    std::unique_ptr<AsyncProcessIf> operator()(boost::asio::io_context&, std::vector<std::string>) {
        auto rsync = std::make_unique<MockRsyncAsyncProcess>();
        if (hook) {
            hook(*rsync);
        }
        procs.push_back(rsync.get());
        return rsync;
    }
    void SetHook(Hook f) {
        hook = std::move(f);
    }
    std::vector<MockAsyncProcess*> procs;
    Hook hook;
};
 
class TestDaqControllerBase : public ::testing::Test {
public:
    TestDaqControllerBase() : 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());
    }
 
protected:
    Resources m_resources;
    boost::asio::io_context m_io_ctx;
    rad::IoExecutor m_executor;
    DaqControllerOptions m_options;
};
 
class TestDaqControllerInit : public TestDaqControllerBase {};
 
class TestDaqController : public TestDaqControllerBase {
public:
    TestDaqController() {
    }
    void SetUp() override {
        TestDaqControllerBase::SetUp();
        // Set default status. This can be modified by each test-case before calling PostSetup()
        m_initial_status.id = "TEST.ID";
        m_initial_status.state = State::Scheduled;
        m_initial_status.error = false;
        m_initial_status.timestamp = Status::TimePoint::clock::now();
        m_dpspec.target.file_id = "TEST.ID";
        m_dpspec.target.source = DpSpec::SourceFitsFile{"source1", "host:target.fits", {}};
        m_dpspec.sources.push_back(DpSpec::SourceFitsFile{"source2", "host:source.fits", {}});
    }
 
    /**
     * Specifically sets up expectations and other things based on current state.
     * It considers:
     * - m_dpspec for sources and populates m_resolver based on that.
     * - m_initial_status controls which state to start from.
     */
    void PostSetUp() {
        // Set up resolver so that it matches m_dpspec sources.
        std::filesystem::path sources_root = "sources";
        if (m_dpspec.target.source.has_value()) {
            m_resolver.Add({m_dpspec.target.source->source_name, m_dpspec.target.source->origin},
                           sources_root / ParseSourceOrigin(m_dpspec.target.source->origin).path);
        }
        for (auto const& s : m_dpspec.sources) {
            if (std::holds_alternative<DpSpec::SourceFitsFile>(s)) {
                auto const& source = std::get<DpSpec::SourceFitsFile>(s);
                m_resolver.Add({source.source_name, source.origin},
                               sources_root / ParseSourceOrigin(source.origin).path);
            }
        }
        auto ws = std::make_unique<daq::dpm::MockDaqWorkspace>();
        m_ws_mock_ptr = ws.get();
        EXPECT_CALL(*ws, LoadStatus()).WillRepeatedly(Return(m_initial_status));
        EXPECT_CALL(*ws, LoadSourceLookup()).WillRepeatedly(Return(m_resolver.GetMapping()));
        EXPECT_CALL(*ws, LoadSpecification()).WillRepeatedly(Return(m_dpspec));
        EXPECT_CALL(*ws, GetSourcesPath()).WillRepeatedly(Return(std::filesystem::path("sources")));
        EXPECT_CALL(*ws, GetSourceLookupPath())
            .WillRepeatedly(Return(std::filesystem::path("sources.json")));
        EXPECT_CALL(*ws, GetPath())
            .Times(AnyNumber())
            .WillRepeatedly(Return(std::filesystem::path("")));
        m_daq = std::make_unique<DaqControllerImpl>(m_executor,
                                                    std::move(ws),
                                                    m_resources,
                                                    std::reference_wrapper(m_rsync_factory),
                                                    std::reference_wrapper(m_proc_factory),
                                                    m_options);
        m_daq->Start();
    }
 
    void TearDown() override {
        m_daq.reset();
        TestDaqControllerBase::TearDown();
    }
 
    /**
     * Poll one handler at a time until predicate has been satisifed or io_context runs out of work.
     */
    template <class Pred>
    void PollUntil(Pred&& p) {
        while (true) {
            m_io_ctx.restart();
            if (m_io_ctx.poll_one() == 0 || p()) {
                break;
            }
        }
    }
 
protected:
    Status m_initial_status = {};
    DpSpec m_dpspec;
    SourceResolver m_resolver;
    // Not-owned pointer to a DaqWorkspace
    daq::dpm::MockDaqWorkspace* m_ws_mock_ptr = nullptr;
    std::unique_ptr<DaqControllerImpl> m_daq;
    FakeRsyncFactory m_rsync_factory;
    FakeProcFactory m_proc_factory;
};
 
using namespace ::testing;
 
TEST_F(TestDaqController, ScheduledTransitionsToTransferring) {
    // Additional setup
    PostSetUp();
    EXPECT_CALL(*m_ws_mock_ptr, StoreSourceLookup(_));
    EXPECT_CALL(*m_ws_mock_ptr, StoreStatus(Field(&Status::state, State::Transferring)));
 
    // Run
    // Only run handlers until we reach State::Transferring
    PollUntil([&] { return m_daq->GetState() == State::Transferring; });
    EXPECT_EQ(m_daq->GetState(), State::Transferring);
}
 
TEST_F(TestDaqController, TransferringWithoutFilesTransitionsToMerging) {
    // Additional setup
    m_dpspec.target.source = std::nullopt;
    m_dpspec.sources.clear();
    m_initial_status.state = State::Transferring;
    PostSetUp();
 
    EXPECT_CALL(*m_ws_mock_ptr, StoreStatus(Field(&Status::state, State::Merging)));
 
    // Run
    // Only run handlers until we reach State::Merging
    PollUntil([&] { return m_daq->GetState() == State::Merging; });
    EXPECT_EQ(m_daq->GetState(), State::Merging);
}
 
TEST_F(TestDaqController, TransferringStartsTransfersAndWhenCompletedTransitionsToMerging) {
    // Additional setup
    ASSERT_TRUE(m_dpspec.target.source);
    ASSERT_TRUE(m_dpspec.sources.size() == 1);
    m_initial_status.state = State::Transferring;
    m_resources.net_receive.SetLimit(0);
    PostSetUp();
 
    EXPECT_EQ(m_resolver.GetMapping().size(), 2u)
        << "Setup should have added target and one extra source";
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/target.fits")))
        .Times(AnyNumber())
        .WillRepeatedly(Return(false));
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/source.fits")))
        .Times(AnyNumber())
        .WillRepeatedly(Return(false));
 
    // Run
    m_io_ctx.poll();
    EXPECT_EQ(m_daq->GetState(), State::Transferring);
    // After promise is completed the files will exist
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/source.fits")))
        .WillOnce(Return(true));
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/target.fits")))
        .WillOnce(Return(true));
    EXPECT_CALL(*m_ws_mock_ptr, StoreStatus(Field(&Status::state, State::Merging)));
 
    // Complete async file operations by setting promises
    for (auto& proc : m_rsync_factory.procs) {
        proc->promise.set_value(0);
    }
    PollUntil([&] { return m_daq->GetState() == State::Merging; });
 
    // Execute pending handlers that should have been registered after promise was set.
    EXPECT_EQ(m_daq->GetState(), State::Merging);
}
 
TEST_F(TestDaqController, TransferringFailsDoesNotTransition) {
    // Additional setup
    ASSERT_TRUE(m_dpspec.target.source);
    ASSERT_TRUE(m_dpspec.sources.size() == 1);
    m_initial_status.state = State::Transferring;
    m_resources.net_receive.SetLimit(0);
    PostSetUp();
 
    EXPECT_EQ(m_resolver.GetMapping().size(), 2u)
        << "Setup should have added target and one extra source";
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/target.fits")))
        .WillOnce(Return(false));
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("sources/source.fits")))
        .WillOnce(Return(false));
 
    // Test
    // Initiates transfers
    m_io_ctx.poll();
 
    EXPECT_EQ(m_daq->GetState(), State::Transferring);
    EXPECT_CALL(*m_ws_mock_ptr,
                StoreStatus(AllOf(Field(&Status::state, State::Transferring),
                                  Field(&Status::error, true))));
 
    // Run test that emulate a successful first file transfer and then subsequent failures.
    // Complete async file operations by setting promises
    int error = 0;
    for (auto& proc : m_rsync_factory.procs) {
        proc->promise.set_value(error++);
    }
    // Execute pending handlers that should have been registered after promise was set.
    m_io_ctx.restart();
    m_io_ctx.poll();
 
    EXPECT_TRUE(m_daq->IsStopped());
    EXPECT_EQ(m_daq->GetState(), State::Transferring);
}
 
/**
 * [recovery]
 * If transfer previously failed and then stopped this should be automatically recoverable if
 * DAQ is started again and requested files are available on FS.
 */
TEST_F(TestDaqController, RecoverAutomaticallyFromTransferringIfFilesExist) {
    // Additional setup
    m_initial_status.state = State::Transferring;
    m_initial_status.error = true;
    PostSetUp();
 
    // Files exist!
    EXPECT_CALL(*m_ws_mock_ptr, Exists(_)).WillRepeatedly(Return(true));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Merging), Field(&Status::error, false))));
 
    // Run
    // Only run handlers until we reach State::Merging
    PollUntil([&] { return m_daq->GetState() == State::Merging; });
    EXPECT_EQ(m_daq->GetState(), State::Merging);
}
 
TEST_F(TestDaqController, MergingSuccessful) {
    // Setup
    m_initial_status.state = State::Merging;
    PostSetUp();
 
    EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
        .WillRepeatedly(Return("specification.json"));
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("TEST.ID.fits")))
        .WillOnce(Return(false));
 
    // Run
    // First the process will be created
    m_io_ctx.poll();
 
    ASSERT_EQ(m_proc_factory.procs.size(), 1u);
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("TEST.ID.fits")))
        .WillRepeatedly(Return(true));
 
    // expect alert to be set
    EXPECT_CALL(*m_ws_mock_ptr,
                StoreStatus(AllOf(
                    Field(&Status::alerts,
                          ElementsAre(Field(&Alert::description,
                                            HasSubstr("Writing keywords required resizing")))),
                    Field(&Status::state, State::Merging),
                    Field(&Status::error, false))));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), Field(&Status::error, false))));
    EXPECT_CALL(*m_ws_mock_ptr, MakeResultSymlink(std::filesystem::path("TEST.ID.fits")));
 
    // Emit alert from merger
    std::string message =
        R"({"content":{"id":"primary_hdu_resize","message":"Writing keywords required resizing of primary HDU: Add space for at least 73 keywords to avoid resize"},"timestamp":1650964356522093759,"type":"alert"})";
    m_proc_factory.procs[0]->stdout(8, message);
 
    // Invalid messages should be ignored
    m_proc_factory.procs[0]->stdout(8, "This is invalid JSON");
 
    // Then we complete
    m_proc_factory.procs[0]->promise.set_value(0);
 
    // Only run handlers until we reach State::Merging
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
/**
 * [recovery]
 * - User manually invokes daqDpmMerge to create output.
 */
TEST_F(TestDaqController, RecoverAutomaticallyFromMergeFailureIfResultExists) {
    // Setup
    // Start in Merging with failure.
    m_initial_status.state = State::Merging;
    m_initial_status.error = true;
    PostSetUp();
 
    EXPECT_CALL(*m_ws_mock_ptr, Exists(std::filesystem::path("TEST.ID.fits")))
        .WillRepeatedly(Return(true));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), Field(&Status::error, false))));
 
    // Run
    // First the process will be created
    m_io_ctx.poll();
    ASSERT_EQ(m_proc_factory.procs.size(), 0u);
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
}  // namespace daq::dpm