testManager.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief daq::ManagerImpl tests
 */
#include <daq/manager.hpp>
#include <fmt/ostream.h>
#include <regex>
 
#include "mock/daqController.hpp"
#include "mock/dpmClientMock.hpp"
#include "mock/mockWorkspace.hpp"
#include "statusObserver.hpp"
#include "utils.hpp"
#include <gtest/gtest.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/error/report.hpp>
 
using namespace daq;
using namespace ::testing;
using namespace std::literals::string_view_literals;
using namespace std::chrono_literals;
 
struct FixtureBase : ::testing::Test {
    template <class T>
    T ExpectNoException(boost::future<T>& f) {
        if (!f.is_ready()) {
            ADD_FAILURE() << "Future is not ready";
            throw std::runtime_error("test failure");
        }
        try {
            return f.get();
        } catch (...) {
            LOG4CPLUS_ERROR("test",
                            "Future contained exception\n"
                                << daq::error::NestedExceptionReporter(std::current_exception()));
            throw;
        }
    }
    ManagerParams m_params = {"INS"};
};
 
/**
 * @ingroup daq_ocm_libdaq_test
 */
class TestManagerImplLifecycle : public FixtureBase {
public:
    void SetUp() override {
    }
    void TearDown() override {
    }
};
 
/**
 * @ingroup daq_ocm_libdaq_test
 */
class TestManagerImpl : public FixtureBase {
public:
    TestManagerImpl()
        : m_io_ctx()
        , m_executor(m_io_ctx)
        , m_dpm_client(std::make_shared<DpmClientMock>())
        , m_event_log(std::make_shared<daq::ObservableEventLog>())
        , m_manager(m_executor,
                    m_params,
                    m_workspace,
                    m_event_log,
                    m_daq_factory,
                    m_dpm_client,
                    log4cplus::Logger::getInstance("test")) {
    }
    /**
     * Creates manager and adds two data acquisitions.
     */
    void SetUp() override {
        m_daq_id_1 = "daq1";
        m_daq_id_2 = "daq2";
        m_daq_ctx_1.id = m_daq_id_1;
        m_daq_ctx_1.file_id = "fileid1";
        m_daq_ctx_2.id = m_daq_id_2;
        m_daq_ctx_2.file_id = "fileid2";
 
        m_daq1_status = std::make_shared<ObservableStatus>(m_daq_id_1, "fileid1");
        m_daq2_status = std::make_shared<ObservableStatus>(m_daq_id_2, "fileid2");
        m_daq1 = std::make_shared<DaqControllerMock>();
        m_daq2 = std::make_shared<DaqControllerMock>();
        m_daq_factory.ocm_mocks["daq1"] = m_daq1;
        m_daq_factory.ocm_mocks["daq2"] = m_daq2;
 
        m_dpm_daq1 = std::make_shared<DaqControllerMock>();
        m_dpm_daq2 = std::make_shared<DaqControllerMock>();
        m_daq_factory.dpm_mocks["daq1"] = m_dpm_daq1;
        m_daq_factory.dpm_mocks["daq2"] = m_dpm_daq2;
 
        EXPECT_CALL(*m_daq1, GetId()).Times(AnyNumber()).WillRepeatedly(ReturnRef(m_daq_id_1));
        EXPECT_CALL(*m_daq2, GetId()).Times(AnyNumber()).WillRepeatedly(ReturnRef(m_daq_id_2));
        EXPECT_CALL(*m_daq1, GetStatus()).Times(AnyNumber()).WillRepeatedly(Return(m_daq1_status));
        EXPECT_CALL(*m_daq2, GetStatus()).Times(AnyNumber()).WillRepeatedly(Return(m_daq2_status));
        EXPECT_CALL(Const(*m_daq1), GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq1_status));
        EXPECT_CALL(Const(*m_daq2), GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq2_status));
        EXPECT_CALL(Const(*m_daq1), GetContext())
            .Times(AnyNumber())
            .WillRepeatedly(ReturnRef(m_daq_ctx_1));
 
        m_daq1_dpm_status = std::make_shared<ObservableStatus>(m_daq_id_1, "fileid1");
        m_daq1_dpm_status->SetState(State::Collecting);
        m_daq2_dpm_status = std::make_shared<ObservableStatus>(m_daq_id_2, "fileid2");
        m_daq2_dpm_status->SetState(State::Merging);
 
        EXPECT_CALL(*m_dpm_daq1, GetId()).WillRepeatedly(ReturnRef(m_daq_id_1));
        EXPECT_CALL(*m_dpm_daq1, GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq1_status));
        EXPECT_CALL(Const(*m_dpm_daq1), GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq1_status));
        EXPECT_CALL(Const(*m_dpm_daq1), GetContext())
            .Times(AnyNumber())
            .WillRepeatedly(ReturnRef(m_daq_ctx_1));
 
        EXPECT_CALL(*m_dpm_daq2, GetId()).WillRepeatedly(ReturnRef(m_daq_id_2));
        EXPECT_CALL(*m_dpm_daq2, GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq2_dpm_status));
        EXPECT_CALL(Const(*m_dpm_daq2), GetStatus())
            .Times(AnyNumber())
            .WillRepeatedly(Return(m_daq2_dpm_status));
        EXPECT_CALL(Const(*m_dpm_daq2), GetContext())
            .Times(AnyNumber())
            .WillRepeatedly(ReturnRef(m_daq_ctx_2));
    }
 
    auto StartDaq1() -> boost::future<State> {
        EXPECT_CALL(*m_daq1, StartAsync())
            .WillOnce(Return(ByMove(boost::make_ready_future<State>(m_daq1_status->GetState()))));
        EXPECT_CALL(m_workspace, StoreList(std::vector<std::string>({"daq1"})));
        EXPECT_CALL(m_workspace, StoreStatus(Field(&Status::id, "daq1"))).Times(AtLeast(1));
        EXPECT_CALL(m_workspace, StoreContext(_));
        return m_manager.StartDaqAsync(m_daq_ctx_1);
    }
 
    void TearDown() override {
        // Run any pending handlers to allow mocks to expire without leaking
        m_io_ctx.restart();
        m_io_ctx.poll();
 
        m_daq_factory.ocm_mocks.clear();
        m_daq_factory.dpm_mocks.clear();
    }
 
    boost::asio::io_context m_io_ctx;
    rad::IoExecutor m_executor;
    MockWorkspace m_workspace;
    std::shared_ptr<DpmClientMock> m_dpm_client;
    std::shared_ptr<daq::ObservableEventLog> m_event_log;
    std::string m_daq_id_1;
    std::string m_daq_id_2;
    DaqContext m_daq_ctx_1;
    DaqContext m_daq_ctx_2;
    std::shared_ptr<DaqControllerMock> m_daq1;
    std::shared_ptr<DaqControllerMock> m_daq2;
    std::shared_ptr<DaqControllerMock> m_dpm_daq1;
    std::shared_ptr<DaqControllerMock> m_dpm_daq2;
    std::shared_ptr<ObservableStatus> m_daq1_status;
    std::shared_ptr<ObservableStatus> m_daq2_status;
    std::shared_ptr<ObservableStatus> m_daq1_dpm_status;
    std::shared_ptr<ObservableStatus> m_daq2_dpm_status;
    DaqControllerFactoryFake m_daq_factory;
    ManagerImpl m_manager;
};
 
TEST_F(TestManagerImplLifecycle, AddDaqNotifiesObserver) {
    // Setup
    boost::asio::io_context io_ctx;
    rad::IoExecutor executor(io_ctx);
    DaqControllerFactoryFake factory;
    auto event_log = std::make_shared<ObservableEventLog>();
    DaqContext daq_ctx;
    MockWorkspace workspace;
    std::shared_ptr<DpmClientMock> dpm_client = std::make_shared<DpmClientMock>();
 
    ManagerImpl mgr(executor,
                    m_params,
                    workspace,
                    event_log,
                    factory,
                    dpm_client,
                    log4cplus::Logger::getInstance("test"));
    auto id = std::string("id");
    auto status = std::make_shared<ObservableStatus>(id, "fileid");
    daq_ctx.id = id;
    daq_ctx.file_id = id;
    auto daq = std::make_shared<DaqControllerMock>();
    factory.ocm_mocks[id] = daq;
 
    EXPECT_CALL(*daq, StartAsync())
        .WillOnce(Return(ByMove(boost::make_ready_future<State>(status->GetState()))));
    EXPECT_CALL(*daq, GetId()).WillRepeatedly(ReturnRef(id));
    EXPECT_CALL(*daq, GetStatus()).WillRepeatedly(Return(status));
    EXPECT_CALL(Const(*daq), GetStatus()).WillRepeatedly(Return(status));
    EXPECT_CALL(Const(*daq), GetContext()).Times(AnyNumber()).WillRepeatedly(ReturnRef(daq_ctx));
 
    EXPECT_CALL(workspace, StoreList(std::vector<std::string>({"id"})));
    EXPECT_CALL(workspace, StoreStatus(Field(&Status::id, "id")));
    EXPECT_CALL(workspace, StoreContext(daq_ctx));
 
    StatusObserverMock o;
    EXPECT_CALL(o, CallOperator(_));
    mgr.GetStatusSignal().ConnectObserver(std::reference_wrapper(o));
 
    // Run
    auto f = mgr.StartDaqAsync(daq_ctx);
    io_ctx.poll();
    ExpectNoException(f);
}
 
TEST_F(TestManagerImplLifecycle, AwaitStateCompletesWithAbandonedManager) {
    // Setup
    boost::asio::io_context io_ctx;
    rad::IoExecutor executor(io_ctx);
    DaqControllerFactoryFake factory;
    auto event_log = std::make_shared<ObservableEventLog>();
    DaqContext daq_ctx;
    MockWorkspace workspace;
    std::shared_ptr<DpmClientMock> dpm_client = std::make_shared<DpmClientMock>();
 
    auto id = std::string("id");
    daq_ctx.id = id;
    daq_ctx.file_id = id;
    auto status = std::make_shared<ObservableStatus>(id, "fileid");
    auto daq = std::make_shared<DaqControllerMock>();
    factory.ocm_mocks[id] = daq;
    EXPECT_CALL(*daq, StartAsync())
        .WillOnce(Return(ByMove(boost::make_ready_future<State>(status->GetState()))));
    EXPECT_CALL(*daq, GetId()).Times(AnyNumber()).WillRepeatedly(ReturnRef(id));
    EXPECT_CALL(*daq, GetStatus()).Times(AnyNumber()).WillRepeatedly(Return(status));
    EXPECT_CALL(Const(*daq), GetStatus()).Times(AnyNumber()).WillRepeatedly(Return(status));
    EXPECT_CALL(Const(*daq), GetContext()).Times(AnyNumber()).WillRepeatedly(ReturnRef(daq_ctx));
 
    EXPECT_CALL(workspace, StoreList(std::vector<std::string>({"id"})));
    EXPECT_CALL(workspace, StoreStatus(Field(&Status::id, "id")));
    EXPECT_CALL(workspace, StoreContext(daq_ctx));
 
    boost::future<Result<Status>> res;
    {
        ManagerImpl mgr(executor,
                        m_params,
                        workspace,
                        event_log,
                        factory,
                        dpm_client,
                        log4cplus::Logger::getInstance("test"));
        mgr.StartDaqAsync(daq_ctx);
 
        // Initiate await that should be aborted when manager is destroyed
        mgr.AwaitDaqStateAsync("id"sv, State::Acquiring, 5ms).swap(res);
        ASSERT_FALSE(res.is_ready());
    }
 
    // Run
    MakeTestProgress(io_ctx, &res);
    ASSERT_TRUE(res.is_ready());
    EXPECT_THROW(res.get(), DaqOperationAborted);
}
 
TEST_F(TestManagerImpl, MakeId) {
    {
        std::regex regex("INSTR\\.\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}");
        auto id = MakeIdCandidate("INSTRUMENT", 0);
        EXPECT_TRUE(std::regex_match(id, regex))
            << "Instrument ID should be truncated to 5 characters if too long";
    }
    {
        std::regex regex("INS\\.\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}");
        auto id = m_manager.MakeDaqId();
        EXPECT_TRUE(std::regex_match(id, regex));
    }
    {
        auto id1 = MakeIdCandidate("INSTRUMENT", 0);
        auto id2 = MakeIdCandidate("INSTRUMENT", 1);
        EXPECT_NE(id1, id2) << "Adding jitter should have made the ID different";
    }
}
 
TEST_F(TestManagerImpl, StartDaqWithSameIdThrowsInvalidArgument) {
    // Setup
    EXPECT_CALL(*m_daq1, StartAsync())
        .WillOnce(Return(ByMove(boost::make_ready_future<State>(State::Starting))));
    EXPECT_CALL(m_workspace, StoreList(std::vector<std::string>({"daq1"})));
    EXPECT_CALL(m_workspace, StoreStatus(Field(&Status::id, "daq1")));
    EXPECT_CALL(m_workspace, StoreContext(_));
 
    m_manager.StartDaqAsync(m_daq_ctx_1);
 
    // Tes
    auto f = m_manager.StartDaqAsync(m_daq_ctx_1);
    m_io_ctx.poll();
    ASSERT_TRUE(f.is_ready());
    EXPECT_THROW(f.get(), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, StartDaqAsyncStartsDaq) {
    // Setup
    EXPECT_CALL(*m_daq1, StartAsync())
        .WillOnce(Return(ByMove(boost::make_ready_future<State>(State::Starting))));
    EXPECT_CALL(m_workspace, StoreList(std::vector<std::string>({"daq1"})));
    EXPECT_CALL(m_workspace, StoreStatus(Field(&Status::id, "daq1")));
    EXPECT_CALL(m_workspace, StoreContext(_));
 
    // Run
    auto fut = m_manager.StartDaqAsync(m_daq_ctx_1);
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready());
    EXPECT_EQ(fut.get(), State::Starting);
}
 
/**
 * Tests the behaviour that if DaqController::StartAsync() fails it should automatically
 * force-abort the acquisition. This is mainly done because putting that responsibility on the user
 * is not productive.
 *
 * note: This behaviour can be made configurable.
 */
TEST_F(TestManagerImpl, StartDaqAbortsIfDaqControllerFails) {
    // Setup
    EXPECT_CALL(*m_daq1, StartAsync())
        .WillOnce(Return(
            ByMove(boost::make_exceptional_future<State>(std::runtime_error("START FAILED")))));
    boost::promise<daq::Status> abort_reply;
    EXPECT_CALL(*m_daq1, AbortAsync(ErrorPolicy::Tolerant))
        .WillOnce(Return(ByMove(abort_reply.get_future())));
    EXPECT_CALL(m_workspace, StoreList(std::vector<std::string>({"daq1"})));
    EXPECT_CALL(m_workspace, StoreStatus(Field(&Status::id, "daq1")));
    EXPECT_CALL(m_workspace, StoreContext(_));
 
    // Test
    auto f = m_manager.StartDaqAsync(m_daq_ctx_1);
    m_io_ctx.poll();
    ASSERT_FALSE(f.is_ready());
 
    // Send reply
    abort_reply.set_exception(std::logic_error("ABORT ALSO FAILED"));
 
    MakeTestProgress(m_io_ctx, &f);
    ASSERT_TRUE(f.is_ready());
    EXPECT_THROW(f.get(), std::runtime_error);
}
 
TEST_F(TestManagerImpl, StopNonexistingDaqThrowsInvalidArgument) {
    auto fut = m_manager.StopDaqAsync("nonexistant-id"sv, daq::ErrorPolicy::Strict);
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, StopDaqAsyncStopsDaq) {
    // Setup
    StartDaq1();
 
    Status::TimePoint t;
    EXPECT_CALL(*m_daq1, StopAsync(daq::ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(
            boost::make_ready_future<Status>(Status("daq1", "fileid", State::Stopped, false, t)))));
 
    // Run
    auto fut = m_manager.StopDaqAsync("daq1"sv, daq::ErrorPolicy::Strict);
    auto status = ExpectNoException(fut);
    EXPECT_EQ(status.state, State::Stopped);
    EXPECT_EQ(status.error, false);
}
 
TEST_F(TestManagerImpl, AbortNonexistingDaqThrowsInvalidArgumentEvenIfTolerant) {
    auto fut = m_manager.AbortDaqAsync("nonexistant-id"sv, ErrorPolicy::Tolerant);
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, AbortDaqAsyncAbortsDaq) {
    // Setup
    StartDaq1();
 
    auto reply_status = daq::Status("daq1", "fileid");
    reply_status.state = State::AbortingAcquiring;
    EXPECT_CALL(*m_daq1, AbortAsync(ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(boost::make_ready_future<Status>(reply_status))));
 
    // Run
    auto fut = m_manager.AbortDaqAsync("daq1"sv, ErrorPolicy::Strict);
    auto result = ExpectNoException(fut);
    EXPECT_EQ(result.state, State::AbortingAcquiring);
    EXPECT_FALSE(result.error);
}
 
TEST_F(TestManagerImpl, UpdateKeywordsUpdatesKeywords) {
    // Setup
    StartDaq1();
 
    daq::fits::KeywordVector keywords = {
        daq::fits::ValueKeyword("OBJECT", "OBJECT,SKY"),
        daq::fits::EsoKeyword("OBS TPLNO", static_cast<uint64_t>(2))};
    EXPECT_CALL(*m_daq1, UpdateKeywords(keywords));
 
    // Run
    m_manager.UpdateKeywords("daq1"sv, keywords);
}
 
TEST_F(TestManagerImpl, UpdateKeywordsForNonexistingDaqThrowsInvalidArgument) {
    daq::fits::KeywordVector keywords = {
        daq::fits::ValueKeyword("OBJECT", "OBJECT,SKY"),
        daq::fits::EsoKeyword("OBS TPLNO", static_cast<uint64_t>(2))};
    EXPECT_THROW(m_manager.UpdateKeywords("nonexistant-id"sv, keywords), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, GetStatus) {
    // Setup
    StartDaq1();
 
    // Run
    auto status = m_manager.GetStatus("daq1"sv);
    EXPECT_EQ(status.id, "daq1");
    EXPECT_EQ(status.state, daq::State::NotStarted);
    EXPECT_FALSE(status.error);
}
 
TEST_F(TestManagerImpl, GetStatusThrowsIfDaqDoesNotExist) {
    // Run
    EXPECT_CALL(m_workspace, LoadArchivedStatus("nonexistant")).WillOnce(Return(std::nullopt));
    EXPECT_THROW(m_manager.GetStatus("nonexistant"sv), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, GetStatusReturnsArchivedStatus) {
    // Run
    auto status = Status("id", "id");
    EXPECT_CALL(m_workspace, LoadArchivedStatus("archived")).WillOnce(Return(status));
    EXPECT_EQ(m_manager.GetStatus("archived"sv), status);
}
 
TEST_F(TestManagerImpl, AwaitDaqStateWorksForCompletedArchivedDaqs) {
    // Run
    auto status = Status("id", "id");
    status.state = State::Completed;
    EXPECT_CALL(m_workspace, LoadArchivedStatus("archived")).WillOnce(Return(status));
    auto fut = m_manager.AwaitDaqStateAsync("archived"sv, daq::State::Completed, 10ms);
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready());
    auto [timeout, result] = fut.get();
    EXPECT_FALSE(timeout);
    EXPECT_EQ(result, status);
}
 
TEST_F(TestManagerImpl, AwaitDaqStateReturnsExceptionalFutureForIncompleteArchivedDaqs) {
    // Run
    auto status = Status("id", "id");
    status.state = State::Collecting;
    EXPECT_CALL(m_workspace, LoadArchivedStatus("archived")).WillOnce(Return(status));
    auto fut = m_manager.AwaitDaqStateAsync("archived"sv, daq::State::Completed, 10ms);
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::invalid_argument);
}
 
TEST_F(TestManagerImpl, AwaitDaqStateReturnsReadyFutureIfConditionIsFulfilled) {
    // Run
    StartDaq1();
    auto fut = m_manager.AwaitDaqStateAsync("daq1"sv, daq::State::NotStarted, 10ms);
    m_io_ctx.poll();
 
    ASSERT_TRUE(fut.is_ready());
    auto [timeout, result] = fut.get();
    EXPECT_FALSE(timeout);
    EXPECT_EQ(result, m_daq1->GetStatus()->GetStatus());
}
 
TEST_F(TestManagerImpl, AwaitDaqStateIsReadyWhenConditionIsFulfilled) {
    // Run
    StartDaq1();
    auto fut = m_manager.AwaitDaqStateAsync("daq1"sv, daq::State::Acquiring, 10ms);
    ASSERT_FALSE(fut.is_ready());
    m_daq1_status->SetState(daq::State::Acquiring);
    MakeTestProgress(m_io_ctx, &fut);
 
    ASSERT_TRUE(fut.is_ready());
    auto [timeout, result] = fut.get();
    EXPECT_FALSE(timeout);
    EXPECT_EQ(result, m_daq1->GetStatus()->GetStatus());
}
 
TEST_F(TestManagerImpl, AwaitDaqStateIsReadyWhenItTimesout) {
    // Run
    StartDaq1();
    auto fut = m_manager.AwaitDaqStateAsync("daq1"sv, daq::State::Acquiring, 0ms);
    ASSERT_FALSE(fut.is_ready());
    MakeTestProgress(m_io_ctx, &fut);
 
    ASSERT_TRUE(fut.is_ready()) << "Timer should have triggered to make the future ready";
    auto [timeout, result] = fut.get();
    EXPECT_TRUE(timeout);
    EXPECT_EQ(result, m_daq1->GetStatus()->GetStatus());
}
 
TEST_F(TestManagerImpl, StoppedDaqTransitionsToNotScheduledAndIsScheduledThenCompleted) {
    // Setup
    StartDaq1();
 
    // Test
    // Simulate completed Acquisition Phase
    m_daq1_status->SetState(State::Stopped);
 
    EXPECT_CALL(*m_daq1, GetEventLog()).WillOnce(Return(m_event_log));
    EXPECT_CALL(*m_daq1, GetContext()).WillOnce(ReturnRef(m_daq_ctx_1));
 
    EXPECT_CALL(*m_dpm_daq1, ScheduleMergeAsync())
        .WillOnce(Return(ByMove(boost::make_ready_future<State>(State::Scheduled))));
    // When DAQ transitions to using DpmDaqController workspace is updated
    EXPECT_CALL(m_workspace, StoreList(_)).Times(AnyNumber());
    EXPECT_CALL(m_workspace, StoreContext(_)).Times(AnyNumber());
 
    // Run pending completion handlers
    m_io_ctx.poll();
 
    // After poll we expect Manager to have created DPM version of daqcontroller and transitioned it
    // to NotScheduled.
    EXPECT_EQ(m_daq1_status->GetState(), State::NotScheduled);
 
    auto daqs = m_manager.GetDaqControllers();
    ASSERT_EQ(daqs.size(), 1u);
    EXPECT_EQ(daqs[0].get(), m_dpm_daq1.get());
 
    // Simulate start and then completion emitting state changes in ObservableStatus.
    // Although signal observers are called synchronously we still pump event loop in case
    // any callbacks are scheduled.
    {
        EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq1_status->GetId())).Times(1);
        m_daq1_status->SetState(State::Merging);
        m_io_ctx.restart();
        m_io_ctx.poll();
    }
 
    {
        EXPECT_CALL(*m_dpm_client, StopMonitorStatus(m_daq_ctx_1.id)).Times(1);
        EXPECT_CALL(m_workspace, ArchiveDaq(m_daq_ctx_1.id)).Times(1);
 
        m_daq1_status->SetState(State::Completed);
        m_io_ctx.restart();
        m_io_ctx.poll();
        EXPECT_FALSE(m_manager.HaveDaq(m_daq_ctx_1.id));
    }
}
 
TEST_F(TestManagerImpl, RestoreFromWorkspaceAddsDaq) {
    // Setup
    // daq1 should create OCM daq controller
    // daq2 should create DPM daq controller
    //
    // Set time to be not stale
    m_daq_ctx_1.creation_time = std::chrono::system_clock::now();
    m_daq_ctx_2.creation_time = std::chrono::system_clock::now();
    m_daq1_status->SetState(State::Acquiring);
    m_daq2_status->SetState(State::Merging);
 
    std::vector<std::string> to_load = {m_daq_ctx_1.id, m_daq_ctx_2.id};
 
    EXPECT_CALL(m_workspace, LoadList()).WillOnce(Return(to_load));
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_1.id)).WillOnce(Return(m_daq_ctx_1));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_1.id))
        .WillOnce(Return(m_daq1_status->GetStatus()));
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_2.id)).Times(0);
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_2.id)).WillOnce(Return(m_daq_ctx_2));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_2.id))
        .WillOnce(Return(m_daq2_status->GetStatus()));
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_2.id));
 
    EXPECT_CALL(m_workspace, StoreList(to_load)).Times(1);
 
    // Test
    m_manager.RestoreFromWorkspace();
}
 
TEST_F(TestManagerImpl, RestoreFromWorkspaceArchivesStaleDaqs) {
    // Setup
    // daq1 should be archived
    // daq2 should create DPM daq controller
    //
    // Set time to be not stale
    m_daq_ctx_1.creation_time = std::chrono::system_clock::now() - m_params.acquiring_stale_age;
    m_daq_ctx_2.creation_time = std::chrono::system_clock::now() - m_params.merging_stale_age;
 
    std::vector<std::string> to_load = {m_daq_ctx_1.id, m_daq_ctx_2.id};
 
    EXPECT_CALL(m_workspace, LoadList()).WillOnce(Return(to_load));
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_1.id)).WillOnce(Return(m_daq_ctx_1));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_1.id))
        .WillOnce(Return(m_daq1_status->GetStatus()));
    EXPECT_CALL(m_workspace, ArchiveDaq(m_daq_ctx_1.id));
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_2.id)).WillOnce(Return(m_daq_ctx_2));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_2.id))
        .WillOnce(Return(m_daq2_dpm_status->GetStatus()));
    EXPECT_CALL(m_workspace, ArchiveDaq(m_daq_ctx_2.id));
 
    EXPECT_CALL(m_workspace, StoreList(std::vector<std::string>())).Times(1);
 
    // Test
    m_manager.RestoreFromWorkspace();
}
 
TEST_F(TestManagerImpl, RestoreFromWorkspaceSkipsMissingDaq) {
    // Setup
    // daq1 should be archived
    // daq2 should create DPM daq controller
    //
    // Set time to be not stale
    m_daq_ctx_1.creation_time = std::chrono::system_clock::now();
    m_daq_ctx_2.creation_time = std::chrono::system_clock::now();
 
    std::vector<std::string> to_load = {m_daq_ctx_1.id, m_daq_ctx_2.id};
    std::vector<std::string> to_store = {m_daq_ctx_2.id};
 
    EXPECT_CALL(m_workspace, LoadList()).WillOnce(Return(to_load));
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_1.id))
        .WillOnce(Throw(std::runtime_error("ouch")));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_1.id))
        .Times(Between(0, 1))
        .WillRepeatedly(Return(m_daq1_status->GetStatus()));
    EXPECT_CALL(m_workspace, ArchiveDaq(m_daq_ctx_1.id));
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_1.id)).Times(0);
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_2.id)).WillOnce(Return(m_daq_ctx_2));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_2.id))
        .WillOnce(Return(m_daq2_dpm_status->GetStatus()));
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_2.id)).Times(1);
 
    EXPECT_CALL(m_workspace, StoreList(to_store)).Times(1);
 
    // Test
    m_manager.RestoreFromWorkspace();
}
 
TEST_F(TestManagerImpl, RestoreFromWorkspaceIgnoresArchiveFailure) {
    // Setup
    // daq1 should be archived
    // daq2 should create DPM daq controller
    //
    // Set time to be not stale
    m_daq_ctx_1.creation_time = std::chrono::system_clock::now();
    m_daq_ctx_2.creation_time = std::chrono::system_clock::now();
    std::vector<std::string> to_load = {m_daq_ctx_1.id, m_daq_ctx_2.id};
    std::vector<std::string> to_store = {m_daq_ctx_2.id};
 
    EXPECT_CALL(m_workspace, LoadList()).WillOnce(Return(to_load));
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_1.id))
        .WillOnce(Throw(std::runtime_error("ouch")));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_1.id))
        .Times(Between(0, 1))
        .WillRepeatedly(Return(m_daq1_status->GetStatus()));
    EXPECT_CALL(m_workspace, ArchiveDaq(m_daq_ctx_1.id))
        .WillOnce(Throw(std::runtime_error("ouch")));
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_1.id)).Times(0);
 
    EXPECT_CALL(m_workspace, LoadContext(m_daq_ctx_2.id)).WillOnce(Return(m_daq_ctx_2));
    EXPECT_CALL(m_workspace, LoadStatus(m_daq_ctx_2.id))
        .WillOnce(Return(m_daq2_dpm_status->GetStatus()));
 
    EXPECT_CALL(*m_dpm_client, StartMonitorStatus(m_daq_ctx_2.id)).Times(1);
 
    EXPECT_CALL(m_workspace, StoreList(to_store)).Times(1);
 
    // Test
    m_manager.RestoreFromWorkspace();
}
 
TEST_F(TestManagerImpl, ContextSignalUpdatesWorkspace) {
    // Setup
    StartDaq1();
 
    EXPECT_CALL(m_workspace, StoreContext(Field(&DaqContext::id, "daq1")));
 
    // Run
    m_daq1->signal(m_daq_ctx_1);
}