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 <fmt/format.h>
#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,  // NOLINT
                                                    std::string,  // NOLINT
                                                    RsyncOptions const&,
                                                    RsyncAsyncProcess::DryRun) {
        auto rsync = std::make_unique<MockRsyncAsyncProcess>();
        if (hook) {
            hook(*rsync);
        } else {
            AddDefaultExpectations(*rsync);
        }
        procs.push_back(rsync.get());
        return rsync;
    }
    void AddDefaultExpectations(MockRsyncAsyncProcess& mock) const {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    }
    void SetHook(Hook f) {
        hook = std::move(f);
    }
    std::vector<MockRsyncAsyncProcess*> procs;
    Hook hook;
    std::vector<std::string> rsync_dummy_args;
};
 
struct FakeProcFactory {
    using Hook = std::function<void(MockAsyncProcess&)>;
    // NOLINTNEXTLINE
    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.file_id = "TEST.2023-08-22T14:40:40.245";
        m_initial_status.state = State::Scheduled;
        m_initial_status.timestamp = Status::TimePoint::clock::now();
        ASSERT_TRUE(m_initial_status.receivers.empty());
        m_dpspec.target.file_id = "TEST.ID";
        m_dpspec.target.source = json::FitsFileSource{"source1", "host:target.fits", {}};
        m_dpspec.sources.push_back(json::FitsFileSource{"source2", "host:source.fits", {}});
        m_dpspec.receivers.push_back(json::OlasReceiver{/*.host*/ "", /* .path */ "/local/olas/"});
        m_dpspec.receivers.push_back(
            json::OlasReceiver{/*.host*/ "host", /* .path */ "/remote/olas/"});
    }
 
    /**
     * 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() {
        m_arcfile_path = fmt::format("/local/olas/{}.fits", m_initial_status.file_id);
 
        // 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->location},
                sources_root / json::ParseSourceLocation(m_dpspec.target.source->location).path);
        }
        for (auto const& s : m_dpspec.sources) {
            if (std::holds_alternative<json::FitsFileSource>(s)) {
                auto const& source = std::get<json::FitsFileSource>(s);
                m_resolver.Add({source.source_name, source.location},
                               sources_root / json::ParseSourceLocation(source.location).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, GetResultPath()).WillRepeatedly(Return(std::filesystem::path("")));
        EXPECT_CALL(*ws, GetSourceLookupPath())
            .WillRepeatedly(Return(std::filesystem::path("sources.json")));
        EXPECT_CALL(*ws, GetPath())
            .Times(AnyNumber())
            .WillRepeatedly(Return(std::filesystem::path("")));
        EXPECT_CALL(*ws, GetLogsPath())
            .Times(AnyNumber())
            .WillRepeatedly(Return(std::filesystem::path("logs")));
        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();
            try {
                if (m_io_ctx.poll_one() == 0 || p()) {
                    break;
                }
            } catch (std::exception const& ex) {
                LOG4CPLUS_ERROR("test", "Exception bubbled up through ASIO: " << ex.what());
                throw;
            }
        }
    }
 
protected:
    Status m_initial_status = {};
    json::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;
    // Expected path with name of the archived file
    std::filesystem::path m_arcfile_path;
};
 
using namespace ::testing;
 
TEST_F(TestDaqController, ScheduledTransitionsToCollecting) {
    // Additional setup
    PostSetUp();
    EXPECT_CALL(*m_ws_mock_ptr, StoreSourceLookup(_));
    EXPECT_CALL(*m_ws_mock_ptr, StoreStatus(Field(&Status::state, State::Collecting)));
 
    // Run
    // Only run handlers until we reach State::Collecting
    PollUntil([&] { return m_daq->GetState() == State::Collecting; });
    EXPECT_EQ(m_daq->GetState(), State::Collecting);
}
 
TEST_F(TestDaqController, CollectingWithoutFilesTransitionsToMerging) {
    // Additional setup
    m_dpspec.target.source = std::nullopt;
    m_dpspec.sources.clear();
    m_initial_status.state = State::Collecting;
    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, CollectingStartsTransfersAndWhenCompletedTransitionsToMerging) {
    // Additional setup
    ASSERT_TRUE(m_dpspec.target.source);
    ASSERT_TRUE(m_dpspec.sources.size() == 1);
    m_initial_status.state = State::Collecting;
    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"), std::filesystem::file_type::regular))
        .Times(AnyNumber())
        .WillRepeatedly(Return(false));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        Exists(std::filesystem::path("sources/source.fits"), std::filesystem::file_type::regular))
        .Times(AnyNumber())
        .WillRepeatedly(Return(false));
 
    // Run
    m_io_ctx.poll();
    EXPECT_EQ(m_daq->GetState(), State::Collecting);
    // After promise is completed the files will exist
    EXPECT_CALL(
        *m_ws_mock_ptr,
        Exists(std::filesystem::path("sources/source.fits"), std::filesystem::file_type::regular))
        .WillOnce(Return(true));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        Exists(std::filesystem::path("sources/target.fits"), std::filesystem::file_type::regular))
        .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, CollectingFailsDoesNotTransition) {
    // Additional setup
    ASSERT_TRUE(m_dpspec.target.source);
    ASSERT_TRUE(m_dpspec.sources.size() == 1);
    m_initial_status.state = State::Collecting;
    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"), std::filesystem::file_type::regular))
        .WillOnce(Return(false));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        Exists(std::filesystem::path("sources/source.fits"), std::filesystem::file_type::regular))
        .WillOnce(Return(false));
 
    // Test
    // Initiates transfers
    m_io_ctx.poll();
 
    EXPECT_EQ(m_daq->GetState(), State::Collecting);
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Collecting), ResultOf(HasError, 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::Collecting);
}
 
/**
 * [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, RecoverAutomaticallyFromCollectingIfFilesExist) {
    // Additional setup
    m_initial_status.state = State::Collecting;
    // Emulate alert from previous failure (note that source mapping must match)
    m_initial_status.alerts.push_back(
        MakeAlert(alert::COLLECTING_RSYNC, "sources/source.fits", ""));
    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), ResultOf(HasError, 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, GetLogsPath()).WillRepeatedly(Return("logs"));
    EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
        .WillRepeatedly(Return("specification.json"));
    EXPECT_CALL(*m_ws_mock_ptr,
                Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
        .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"), std::filesystem::file_type::regular))
        .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),
                    ResultOf(HasError, true))));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, true))));
    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::Releasing; });
    EXPECT_EQ(m_daq->GetState(), State::Releasing);
}
 
/**
 * [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.alerts.push_back(MakeAlert(alert::MERGING_MERGE, "", ""));
    PostSetUp();
 
    EXPECT_CALL(*m_ws_mock_ptr,
                Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
        .WillRepeatedly(Return(true));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, 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);
}
 
TEST_F(TestDaqController, NoopReleasingCompletes) {
    // Setup
    // Start in Releasing without errors.
    m_initial_status.state = State::Releasing;
    // Remove default receivers
    m_dpspec.receivers.clear();
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, false))));
    // Run
    m_io_ctx.poll();
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
TEST_F(TestDaqController, ReleasingToOlasIsSuccessfulWithHardLink) {
    // Setup
    m_initial_status.state = State::Releasing;
    m_initial_status.result = "TEST.ID.fits";
    std::vector<std::string> rsync_dummy_args = {"rsync", "from", "to"};
    m_rsync_factory.SetHook([&](MockRsyncAsyncProcess& mock) {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    });
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, false))))
        .Times(AtLeast(1));
 
    {
        EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
            .WillRepeatedly(Return("specification.json"));
        EXPECT_CALL(
            *m_ws_mock_ptr,
            Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
            .WillRepeatedly(Return(false));
 
        // Only the OLAS receiver without host will attempt to be hard-linked, which we mock as
        // successful.
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeHardLink(std::filesystem::path("TEST.ID.fits"),
                                 std::filesystem::path(m_arcfile_path)))
            .WillOnce(Return(std::error_code()));
 
        // Run
        // First the first receiver should have have used hard-link and second rsync transfer
        // process will be created
        m_io_ctx.poll();
    }
 
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(0).state, ReceiverStatus::State::Success)
        << "Expected first receiver, which is local, to have used hard links which completes "
           "immediately";
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(1).state, ReceiverStatus::State::Started);
 
    // When rsync completes we expect to transition to completed.
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, false))));
 
    // Then we complete rsync process
    ASSERT_EQ(m_rsync_factory.procs.size(), 1u);
    m_rsync_factory.procs[0]->promise.set_value(0);
 
    // Run handlers until we reach State::Completed
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
 
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
TEST_F(TestDaqController, ReleasingToOlasIsSuccessfulWithSymlink) {
    // Setup
    m_initial_status.state = State::Releasing;
    m_initial_status.result = "TEST.ID.fits";
    std::vector<std::string> rsync_dummy_args = {"rsync", "from", "to"};
    m_rsync_factory.SetHook([&](MockRsyncAsyncProcess& mock) {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    });
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, false))))
        .Times(AtLeast(1));
 
    {
        EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
            .WillRepeatedly(Return("specification.json"));
        EXPECT_CALL(
            *m_ws_mock_ptr,
            Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
            .WillRepeatedly(Return(false));
 
        // Only the OLAS receiver without host will attempt to be hard-linked, which we mock as
        // successful.
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeHardLink(std::filesystem::path("TEST.ID.fits"), m_arcfile_path))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeSymlink(std::filesystem::path("TEST.ID.fits"), m_arcfile_path))
            .WillOnce(Return(std::error_code()));
 
        // Run
        // First the first receiver should have have used hard-link and second rsync transfer
        // process will be created
        m_io_ctx.poll();
    }
 
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(0).state, ReceiverStatus::State::Success)
        << "Expected first receiver, which is local, to have used hard links which completes "
           "immediately";
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(1).state, ReceiverStatus::State::Started);
 
    // When rsync completes we expect to transition to completed.
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, false))));
 
    // Then we complete rsync process
    ASSERT_EQ(m_rsync_factory.procs.size(), 1u);
    m_rsync_factory.procs[0]->promise.set_value(0);
 
    // Run handlers until we reach State::Completed
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
 
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
TEST_F(TestDaqController, ReleasingToOlasHonorsOptionAllowSymlink) {
    // Setup
    json::OlasReceiver receiver{};
    receiver.path = "/local/olas/";
    receiver.options.allow_symlink = false;  // note: default is true
    m_dpspec.receivers.clear();
    m_dpspec.receivers.push_back(receiver);
 
    m_initial_status.state = State::Releasing;
    m_initial_status.result = "TEST.ID.fits";
    std::vector<std::string> rsync_dummy_args = {"rsync", "from", "to"};
    m_rsync_factory.SetHook([&](MockRsyncAsyncProcess& mock) {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    });
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, false))))
        .Times(AtLeast(1));
 
    {
        EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
            .WillRepeatedly(Return("specification.json"));
        EXPECT_CALL(
            *m_ws_mock_ptr,
            Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
            .WillRepeatedly(Return(false));
 
        // Only the OLAS receiver without host will attempt to be hard-linked, which we mock as
        // successful.
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeHardLink(std::filesystem::path("TEST.ID.fits"), m_arcfile_path))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
        EXPECT_CALL(*m_ws_mock_ptr, MakeSymlink(_, _)).Times(0);
 
        // Run
        // First the first receiver should have have used hard-link and second rsync transfer
        // process will be created
        m_io_ctx.poll();
    }
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(0).state, ReceiverStatus::State::Started);
 
    // When rsync completes we expect to transition to completed.
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, false))));
 
    // Then we complete rsync process
    ASSERT_EQ(m_rsync_factory.procs.size(), 1u);
    m_rsync_factory.procs[0]->promise.set_value(0);
 
    // Run handlers until we reach State::Completed
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
 
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
TEST_F(TestDaqController, ReleasingToOlasFailingHardLinkRevertsToRsync) {
    // Setup
    m_initial_status.state = State::Releasing;
    m_initial_status.result = "TEST.ID.fits";
    std::vector<std::string> rsync_dummy_args = {"rsync", "from", "to"};
    m_rsync_factory.SetHook([&](MockRsyncAsyncProcess& mock) {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    });
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, false))))
        .Times(AtLeast(1));
 
    {
        EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
            .WillRepeatedly(Return("specification.json"));
        EXPECT_CALL(
            *m_ws_mock_ptr,
            Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
            .WillRepeatedly(Return(false));
 
        // Only the OLAS receiver without host will attempt to be hard-linked, which we mock as
        // successful.
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeHardLink(std::filesystem::path("TEST.ID.fits"),
                                 std::filesystem::path(m_arcfile_path)))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeSymlink(std::filesystem::path("TEST.ID.fits"),
                                std::filesystem::path(m_arcfile_path)))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
 
        // Run
        // First the first receiver should have have used hard-link and second rsync transfer
        // process will be created
        m_io_ctx.poll();
    }
 
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(0).state, ReceiverStatus::State::Started);
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(1).state, ReceiverStatus::State::Started);
 
    // When rsync completes we expect to transition to completed.
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, false))));
 
    // Then we complete rsync process
    ASSERT_EQ(m_rsync_factory.procs.size(), 2u);
    m_rsync_factory.procs[0]->promise.set_value(0);
    m_rsync_factory.procs[1]->promise.set_value(0);
 
    // Run handlers until we reach State::Completed
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
 
    EXPECT_EQ(m_daq->GetState(), State::Completed);
}
 
TEST_F(TestDaqController, ReleasingToOlasCompletesEvenIfTransferFails) {
    // Tests that failure to release file is not fatal (but DAQ should have error flag set).
    // Setup
    m_initial_status.state = State::Releasing;
    m_initial_status.result = "TEST.ID.fits";
    std::vector<std::string> rsync_dummy_args = {"rsync", "from", "to"};
    m_rsync_factory.SetHook([&](MockRsyncAsyncProcess& mock) {
        EXPECT_CALL(mock, GetArguments()).WillRepeatedly(ReturnRef(rsync_dummy_args));
        EXPECT_CALL(mock, GetPid()).WillRepeatedly(Return(1234));
    });
    PostSetUp();
 
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, false))))
        .Times(AtLeast(1));
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Releasing), ResultOf(HasError, true))))
        .Times(AtLeast(1));
 
    {
        EXPECT_CALL(*m_ws_mock_ptr, GetSpecificationPath())
            .WillRepeatedly(Return("specification.json"));
        EXPECT_CALL(
            *m_ws_mock_ptr,
            Exists(std::filesystem::path("TEST.ID.fits"), std::filesystem::file_type::regular))
            .WillRepeatedly(Return(false));
 
        // Only the OLAS receiver without host will attempt to be hard-linked, which we mock as
        // successful.
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeHardLink(std::filesystem::path("TEST.ID.fits"), m_arcfile_path))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
 
        EXPECT_CALL(*m_ws_mock_ptr,
                    MakeSymlink(std::filesystem::path("TEST.ID.fits"), m_arcfile_path))
            .WillOnce(Return(std::make_error_code(std::errc::not_supported)));
        // Run
        // First the first receiver should have have used hard-link and second rsync transfer
        // process will be created
        m_io_ctx.poll();
    }
 
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(0).state, ReceiverStatus::State::Started);
    EXPECT_EQ(m_daq->GetStatus().GetReceiverStatus(1).state, ReceiverStatus::State::Started);
 
    // When rsync completes we expect to transition to completed.
    EXPECT_CALL(
        *m_ws_mock_ptr,
        StoreStatus(AllOf(Field(&Status::state, State::Completed), ResultOf(HasError, true))));
 
    // Then we complete rsync process
    ASSERT_EQ(m_rsync_factory.procs.size(), 2u);
    m_rsync_factory.procs[0]->promise.set_value(1);
    m_rsync_factory.procs[1]->promise.set_value(0);
 
    // Run handlers until we reach State::Completed
    PollUntil([&] { return m_daq->GetState() == State::Completed; });
 
    EXPECT_EQ(m_daq->GetState(), State::Completed);
    EXPECT_EQ(1, m_daq->GetStatus().GetAlerts().size()) << "Expected one alert from failed rsync";
}
}  // namespace daq::dpm