testOcmDaqController.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Unit test for `daq::OcmDaqController`
 */
// NOLINT
#include <memory>
#include <stdexcept>
#include <thread>
 
#include <gtest/gtest.h>
 
#include <daq/daqController.hpp>
#include <daq/error.hpp>
#include <daq/op/abort.hpp>
#include <daq/op/awaitPrim.hpp>
#include <daq/op/initiate.hpp>
#include <daq/op/start.hpp>
#include <daq/op/stop.hpp>
 
#include "mock/metadaqifMock.hpp"
#include "mock/mockAsyncOperations.hpp"
#include "mock/recifMock.hpp"
#include "statusObserver.hpp"
#include "utils.hpp"
 
using namespace daq;
using namespace ::testing;
using namespace std::chrono;
 
/**
 * Fixture for daq::DaqController life cycle tests
 *
 * @ingroup daq_ocm_libdaq_test
 */
class TestOcmDaqControllerLifeCycle : public ::testing::Test {
public:
    TestOcmDaqControllerLifeCycle()
        : m_io_ctx()
        , m_status(std::make_shared<ObservableStatus>("id", "fileid"))
        , m_event_log(std::make_shared<ObservableEventLog>()) {
        m_prim_rr_client = std::make_shared<RecCmdsAsyncMock>();
        m_meta_rr_client = std::make_shared<MetaDaqAsyncMock>();
        m_ops.start = op::InitiateOperation<op::StartAsync, boost::future<void>, op::AsyncOpParams>;
        m_ops.stop = op::InitiateOperation<op::StopAsync,
                                           boost::future<Result<DpParts>>,
                                           ErrorPolicy,
                                           op::AsyncOpParams>;
        m_ops.abort = op::InitiateOperation<op::AbortAsync,
                                            boost::future<Result<void>>,
                                            ErrorPolicy,
                                            op::AsyncOpParams>;
        m_ops.await_prim = op::InitiateAbortableOperation<op::AwaitPrimAsync,
                                                          boost::future<Result<DpParts>>,
                                                          op::AwaitOpParams>;
        m_context.id = "id";
    }
    boost::asio::io_context m_io_ctx;                        // NOLINT
    std::shared_ptr<ObservableStatus> m_status;              // NOLINT
    std::shared_ptr<ObservableEventLog> m_event_log;         // NOLINT
    std::shared_ptr<PrimSource::RrClient> m_prim_rr_client;  // NOLINT
    std::shared_ptr<MetaSource::RrClient> m_meta_rr_client;  // NOLINT
    OcmAsyncOperations m_ops;                                // NOLINT
    DaqContext m_context;                                    // NOLINT
    DaqSources m_sources;
};
 
template <class Iterator>
void SetSourceState(Iterator begin, Iterator end, State state) {
    auto it = begin;
    for (; it != end; ++it) {
        it->SetState(state);
    }
}
 
void SetAllSourceState(op::AsyncOpParams& params, State state) {
    SetSourceState(params.meta_sources.begin(), params.meta_sources.end(), state);
    SetSourceState(params.prim_sources.begin(), params.prim_sources.end(), state);
}
 
/**
 *
 * Developer notes:
 * OcmDaqController use boost::when_all to compose futures. This does not support executors and will
 * spawn a thread to perform the work. This means that the tests will either have to block
 * indefinitely with future::get() or use a timeout.
 *
 * @ingroup daq_ocm_libdaq_test
 */
struct TestState : ::testing::Test {
    std::shared_ptr<PrimSource::RrClient> m_prim_rr_client;  // NOLINT
    std::shared_ptr<MetaDaqAsyncMock> m_meta_rr_client;      // NOLINT
    std::shared_ptr<MetaDaqAsyncMock> m_meta_rr_client2;     // NOLINT
    std::unique_ptr<MockAsyncOperations> m_mock_ops;         // NOLINT
 
    boost::asio::io_context m_io_ctx;                 // NOLINT
    StatusObserverMock m_observer;                    // NOLINT
    std::shared_ptr<ObservableStatus> m_status;       // NOLINT
    std::shared_ptr<ObservableEventLog> m_event_log;  // NOLINT
    std::shared_ptr<OcmDaqController> m_daq;          // NOLINT
    DaqContext m_context;                             // NOLINT
    DaqSources m_sources;
    std::vector<DpPart> m_files;     // NOLINT
    fits::KeywordVector m_keywords;  // NOLINT
 
    TestState()
        : m_io_ctx()
        , m_status(std::make_shared<ObservableStatus>("id", "fileid"))
        , m_event_log(std::make_shared<ObservableEventLog>()) {
    }
 
    void SetUp() override {
        m_files.emplace_back("foo", "bar");
        m_keywords.emplace_back(fits::EsoKeyword("FOO", "BAR"));
        m_keywords.emplace_back(fits::ValueKeyword("FOO", "BAR"));
 
        m_prim_rr_client = std::make_shared<RecCmdsAsyncMock>();
        m_meta_rr_client = std::make_shared<MetaDaqAsyncMock>();
        m_meta_rr_client2 = std::make_shared<NiceMock<MetaDaqAsyncMock>>();
        auto tup = CreateMockAsyncOperations();
        m_mock_ops.swap(std::get<std::unique_ptr<MockAsyncOperations>>(tup));
 
        // Connect listener
        // @todo: Add expectations for observer
        // m_status->ConnectObserver(std::reference_wrapper(m_observer));
 
        MetaSource s1("meta-source-1", m_meta_rr_client);
        MetaSource s2("meta-source-2", m_meta_rr_client2);
        m_context.id = "id";
        m_sources.GetPrimarySources() = std::vector<daq::PrimSource>{},
        m_sources.GetMetadataSources() = std::vector<daq::MetaSource>{s1, s2},
 
        PreDaqControllerHook();
        m_daq = std::make_shared<daq::OcmDaqController>(m_io_ctx,
                                                        m_context,
                                                        m_sources,
                                                        m_status,
                                                        m_event_log,
                                                        std::get<OcmAsyncOperations>(tup));
 
        ASSERT_TRUE(m_daq);
        ASSERT_EQ(m_status->GetState(), m_daq->GetState());
    }
 
    void TearDown() override {
        m_daq.reset();
        m_meta_rr_client.reset();
        m_meta_rr_client2.reset();
        m_prim_rr_client.reset();
    }
 
    virtual void PreDaqControllerHook() {
    }
    virtual void PreStartAsyncHook() {
    }
 
    /**
     * Executes a successful StartAsync() call
     */
    void StartDaq() {
        // Setup
        // Set up mock so that op::StartAsync invocation returns the future from our promise.
        boost::promise<void> reply_promise;
        std::optional<op::AsyncOpParams> params;
        EXPECT_CALL(*m_mock_ops, Start(_))
            .WillOnce(DoAll(Invoke([&](auto p) { params.emplace(p); }),
                            Return(ByMove(reply_promise.get_future()))));
        PreStartAsyncHook();
 
        // Run
        auto fut = m_daq->StartAsync();
        EXPECT_EQ(State::Starting, m_daq->GetState());
        EXPECT_FALSE(fut.is_ready());
 
        // "Send reply"
        reply_promise.set_value();
        ASSERT_TRUE(params);
        SetAllSourceState(*params, State::Acquiring);
 
        // Execute scheduled handlers
        MakeTestProgress(m_io_ctx, &fut);
 
        ASSERT_TRUE(fut.is_ready());
        EXPECT_EQ(State::Acquiring, fut.get());
    }
 
    void AbortDaq() {
        // Setup
        // Set up mock so that op::StartAsync invocation returns the future from our promise.
        std::optional<op::AsyncOpParams> params;
        boost::promise<Result<void>> reply_promise;
        EXPECT_CALL(*m_mock_ops, Abort(ErrorPolicy::Strict, _))
            .WillOnce(DoAll(Invoke([&](auto policy, auto p) { params.emplace(p); }),
                            Return(ByMove(reply_promise.get_future()))));
 
        // Run
        auto fut = m_daq->AbortAsync(ErrorPolicy::Strict);
 
        EXPECT_EQ(State::AbortingAcquiring, m_daq->GetState())
            << "Expected state to be in Stopping after requesting to abort";
 
        // "Send reply"
        reply_promise.set_value({false});
        ASSERT_TRUE(params);
        SetAllSourceState(*params, State::Aborted);
 
        // Execute handlers
        MakeTestProgress(m_io_ctx, &fut);
 
        ASSERT_TRUE(fut.is_ready());
        ASSERT_FALSE(fut.has_exception()) << "Future has unexpected exception!";
        auto result = fut.get();
        EXPECT_EQ(State::Aborted, result.state);
        EXPECT_FALSE(result.error);
        EXPECT_EQ(State::Aborted, m_daq->GetState());
    }
 
    void StopDaq() {
        std::optional<op::AsyncOpParams> params;
        boost::promise<Result<DpParts>> reply_promise;
 
        EXPECT_CALL(*m_mock_ops, Stop(ErrorPolicy::Strict, _))
            .WillOnce(DoAll(Invoke([&](auto policy, auto p) { params.emplace(p); }),
                            Return(ByMove(reply_promise.get_future()))));
 
        // Run
        auto fut = m_daq->StopAsync(ErrorPolicy::Strict);
 
        EXPECT_EQ(State::Stopping, m_daq->GetState())
            << "Expected state to be in Stopping after requesting to stop";
 
        ASSERT_TRUE(params);
        SetAllSourceState(*params, State::Stopped);
 
        // "Send reply"
        Result<DpParts> reply{false, m_files};
        reply_promise.set_value(reply);
 
        // Execute handlers
        MakeTestProgress(m_io_ctx, &fut);
 
        ASSERT_TRUE(fut.is_ready());
        ASSERT_FALSE(fut.has_exception()) << "Future has unexpected exception!";
        auto status = fut.get();
        EXPECT_EQ(State::Stopped, status.state)
            << "Expected state to be Stopped since there were no errors";
        EXPECT_FALSE(status.error);
        EXPECT_EQ(State::Stopped, m_daq->GetState());
    }
};
 
struct TestAwait : TestState {};
 
struct TestNotStarted : TestState {};
 
struct TestAcquiring : TestState {
    void PreDaqControllerHook() {
        m_status->SetState(State::Acquiring);
    }
};
 
struct TestStopped : TestState {
    void PreDaqControllerHook() {
        m_status->SetState(State::Stopped);
    }
};
 
/**
 * Fixture for
 */
struct TestAwaitWithPrimSource : TestState {
    void PreDaqControllerHook() {
        // Add a primary source, which was not needed for other tests
        PrimSource s1("prim-source-1", m_prim_rr_client);
        m_sources.GetPrimarySources() = std::vector<daq::PrimSource>{s1};
    }
    void PreStartAsyncHook() {
        EXPECT_CALL(*m_mock_ops, AwaitPrim(_))
            .WillOnce(Return(ByMove(m_await_promise.get_future())));
    }
 
    boost::promise<Result<DpParts>> m_await_promise;  // NOLINT
};
 
// Simple test to understand boost::future::unwrap()
TEST(TestBoost, Unwrap) {
    EXPECT_THROW(boost::make_ready_future()
                     .then([](auto f) -> boost::future<void> {
                         try {
                             throw std::runtime_error("Meow");
                         } catch (...) {
                             return boost::make_exceptional_future<void>();
                         }
                     })
                     .unwrap()
                     .get(),
                 std::runtime_error);
}
 
TEST_F(TestOcmDaqControllerLifeCycle, ConstructorFailsIfNoSourcesAreProvided) {
    ASSERT_THROW(
        OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, m_ops),
        std::invalid_argument);
}
 
TEST_F(TestOcmDaqControllerLifeCycle, ConstructorFailsObservableStatusIdDoesNotMatchDaqContextId) {
    MetaSource s("source-id", m_meta_rr_client);
    m_sources.GetMetadataSources() = {s};
    m_context.id = "not-id";
    ASSERT_THROW(
        OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, m_ops),
        std::invalid_argument);
}
 
TEST_F(TestOcmDaqControllerLifeCycle, ConstructorFailsIfAsyncOperationIsInvalid) {
    MetaSource s("source-id", m_meta_rr_client);
    m_sources.GetMetadataSources() = {s};
    {
        auto ops = m_ops;
        ops.start = {};
        ASSERT_THROW(
            OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, ops),
            std::invalid_argument);
    }
    {
        auto ops = m_ops;
        ops.stop = {};
        ASSERT_THROW(
            OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, ops),
            std::invalid_argument);
    }
    {
        auto ops = m_ops;
        ops.abort = {};
        ASSERT_THROW(
            OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, ops),
            std::invalid_argument);
    }
    {
        auto ops = m_ops;
        ops.await_prim = {};
        ASSERT_THROW(
            OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, ops),
            std::invalid_argument);
    }
}
 
TEST_F(TestOcmDaqControllerLifeCycle, ConstructorSucceedsIfSingleMetadataSourceIsUsed) {
    MetaSource s("source-id", m_meta_rr_client);
    m_sources.GetMetadataSources() = {s};
    OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, m_ops);
}
 
TEST_F(TestOcmDaqControllerLifeCycle, DestructionAbortsAsyncWait) {
    MetaSource s("source-id", m_meta_rr_client);
    m_sources.GetMetadataSources() = {s};
    boost::future<State> fut;
    {
        auto daq =
            OcmDaqController::Create(m_io_ctx, m_context, m_sources, m_status, m_event_log, m_ops);
        fut = daq->AwaitAsync({"source-id"}, 100ms);
        ASSERT_FALSE(fut.is_ready());
    }
 
    MakeTestProgress(m_io_ctx, &fut);
    ASSERT_TRUE(fut.is_ready())
        << "Future should have been cancelled since daq should have been deleted.";
    EXPECT_TRUE(fut.has_exception());
    EXPECT_THROW(fut.get(), DaqOperationAborted);
}
 
TEST_F(TestState, NotStarted) {
    ASSERT_EQ(State::NotStarted, m_daq->GetState()) << "The initial state should be NotStarted";
}
 
TEST_F(TestState, GetStatusReturnsSameStatusObject) {
    auto status_ptr = m_daq->GetStatus();
    EXPECT_EQ(status_ptr.get(), m_status.get());
}
 
TEST_F(TestState, CannotStopStoppedOcmDaqController) {
    SCOPED_TRACE("CannotStopStoppedOcmDaqController");
    StartDaq();
    StopDaq();
 
    ASSERT_EQ(State::Stopped, m_daq->GetState()) << "Setup failed";
 
    // Try to stop again
    auto fut = m_daq->StopAsync(ErrorPolicy::Strict);
    EXPECT_TRUE(fut.has_exception());
    EXPECT_THROW(fut.get(), std::exception);
}
 
TEST_F(TestState, CannotAbortStoppedOcmDaqController) {
    SCOPED_TRACE("CannotAbortStoppedOcmDaqController");
    StartDaq();
    StopDaq();
 
    ASSERT_EQ(State::Stopped, m_daq->GetState()) << "Setup failed";
 
    // Try to stop again
    auto fut = m_daq->AbortAsync(ErrorPolicy::Strict);
    EXPECT_TRUE(fut.has_exception());
    EXPECT_THROW(fut.get(), std::exception);
}
 
TEST_F(TestState, StartingFailsToSendStartDaqWillAbortAndSetErrorFlagAndStayInStarting) {
    // Setup
    boost::promise<void> reply_promise;
    EXPECT_CALL(*m_mock_ops, Start(_)).WillOnce(Return(ByMove(reply_promise.get_future())));
 
    // Run
    auto fut = m_daq->StartAsync();
    EXPECT_EQ(State::Starting, m_daq->GetState());
 
    // Set up mock future so that it results in an exception exception.
    reply_promise.set_exception(std::runtime_error("Fake test failure"));
 
    // Run async handlers
    MakeTestProgress(m_io_ctx, &fut);
 
    ASSERT_TRUE(fut.is_ready());
    EXPECT_TRUE(fut.has_exception()) << "Expected future to contain exception";
    EXPECT_THROW(fut.get(), std::exception) << "Expected exception to derive from std::exception";
    EXPECT_EQ(true, m_daq->GetErrorFlag());
}
 
TEST_F(TestState, StartAsyncReturnsExceptionalFutureInStateStarting) {
    // Setup
    boost::promise<void> reply_promise;
    EXPECT_CALL(*m_mock_ops, Start(_)).WillOnce(Return(ByMove(reply_promise.get_future())));
 
    auto fut = m_daq->StartAsync();
    ASSERT_EQ(State::Starting, m_daq->GetState());
    EXPECT_FALSE(fut.is_ready());
 
    // Run
    // @todo: Shouldn't this be communicated through the future?
    auto fut2 = m_daq->StartAsync();
    ASSERT_TRUE(fut2.has_exception());
    EXPECT_THROW(fut2.get(), std::exception)
        << "Multiple simultaneous start operations are not supported and an exception "
           "was exected";
 
    // Complete pending operations to avoid "leaking" mock objects
    // "Send reply"
    reply_promise.set_value();
 
    // Make progress
    MakeTestProgress(m_io_ctx, &fut);
}
 
/**
 * It's possible to abort but not stop (and keep)
 */
TEST_F(TestState, StopAsyncThrowsIfNotStarted) {
    auto fut = m_daq->StopAsync(ErrorPolicy::Strict);
    EXPECT_THROW(fut.get(), std::exception)
        << "It should not be possible to stop a data acquisition that has not started";
}
 
TEST_F(TestState, StopAsyncDoesNotThrowWithTolerantPolicy) {
    // Setup
    StartDaq();
 
    std::optional<op::AsyncOpParams> params;
    boost::promise<Result<DpParts>> reply_promise;
    EXPECT_CALL(*m_mock_ops, Stop(ErrorPolicy::Tolerant, _))
        .WillOnce(DoAll(Invoke([&](auto policy, auto p) { params.emplace(p); }),
                        Return(ByMove(reply_promise.get_future()))));
 
    // Since there are no primary sources in TestState fixture we expect no abort
    EXPECT_CALL(m_mock_ops->mock_abort, Abort()).Times(0);
 
    // Run
    auto fut = m_daq->StopAsync(ErrorPolicy::Tolerant);
 
    EXPECT_EQ(State::Stopping, m_daq->GetState())
        << "Expected state to be in Stopping after requesting to stop";
 
    // "Send reply"
    Result<DpParts> reply{true, {}};
    reply_promise.set_value(reply);
    ASSERT_TRUE(params);
    // Since we are forcing stop it should be acceptable that sources are not stopped.
    SetAllSourceState(*params, State::Stopping);
 
    // Execute handlers
    MakeTestProgress(m_io_ctx, &fut);
 
    ASSERT_TRUE(fut.is_ready());
    ASSERT_FALSE(fut.has_exception()) << "Future has unexpected exception!";
    auto status = fut.get();
    EXPECT_EQ(State::Stopped, status.state)
        << "Expected state to be Stopped since there were no errors";
    EXPECT_TRUE(status.error) << "Error flag should be set since the reply_promise had an error";
    EXPECT_EQ(State::Stopped, m_daq->GetState());
}
 
/**
 * It should be possible to abort a data acquisition even if it's not started.
 */
TEST_F(TestState, AbortAsyncIsOkIfNotStarted) {
    auto fut = m_daq->AbortAsync(ErrorPolicy::Strict);
    ASSERT_TRUE(fut.is_ready())
        << "Aborting a NotStarted data acquisition should be ready immediately";
    EXPECT_FALSE(fut.has_exception()) << "Future should not have failed";
 
    auto result = fut.get();
    EXPECT_EQ(State::Aborted, result.state) << "Unexpected state";
    EXPECT_FALSE(result.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
/**
 * It's possible to abort but not stop (and keep) if data acquisition is starting
 */
TEST_F(TestState, StopAsyncThrowsIfStarting) {
    // Setup
    boost::promise<void> reply_promise;
    EXPECT_CALL(*m_mock_ops, Start(_)).WillOnce(Return(ByMove(reply_promise.get_future())));
 
    auto fut = m_daq->StartAsync();
    ASSERT_EQ(State::Starting, m_daq->GetState())
        << "Setup failed, unexpected state, aborting test";
    ASSERT_FALSE(fut.is_ready());
 
    // Run
    {
        auto fut = m_daq->StopAsync(ErrorPolicy::Strict);
        EXPECT_TRUE(fut.has_exception())
            << "Cannot stop unless DAQ is in State::Acquiring, current state: "
            << m_daq->GetState();
 
        EXPECT_THROW(fut.get(), std::exception)
            << "Cannot stop if data acquisition is `Starting`. An exeption was expected";
    }
    // Complete pending operations to avoid "leaking" mock objects
    // "Send reply"
    reply_promise.set_value();
 
    // Make progress
    MakeTestProgress(m_io_ctx, &fut);
}
 
/**
 * Test sequence:
 *
 * 1. Send StartDaq
 * 2. Send AbortDaq
 * 3. StartDaq still succeeds in this case (simulates serial handling of client requests at source).
 * 4. AbortDaq suceeds.
 */
TEST_F(TestState, AbortingIsOkWhenStarting) {
    SCOPED_TRACE("AbortingIsOkWhenStarting");
    // Setup
    // Set up mock so that StartDaq invocation returns the future from our promise.
    boost::promise<void> start_promise;
    EXPECT_CALL(*m_mock_ops, Start(_)).WillOnce(Return(ByMove(start_promise.get_future())));
 
    // Run
    //
    // Start data acquisition
    auto start_fut = m_daq->StartAsync();
    ASSERT_EQ(State::Starting, m_daq->GetState());
    EXPECT_FALSE(start_fut.is_ready());
 
    // Setup
    // And ditto for Abort
    boost::promise<Result<void>> abort_promise;
    EXPECT_CALL(*m_mock_ops, Abort(ErrorPolicy::Strict, _))
        .WillOnce(Return(ByMove(abort_promise.get_future())));
 
    // Run
    //
    // Abort data acquisition
    auto abort_fut = m_daq->AbortAsync(ErrorPolicy::Strict);
 
    // Complete pending operations to avoid "leaking" mock objects
    // "Send reply"
    start_promise.set_value();
 
    // Make progress
    MakeTestProgress(m_io_ctx, &start_fut);
 
    ASSERT_TRUE(start_fut.is_ready()) << "Cannot proceed with test since future is not ready";
    EXPECT_FALSE(start_fut.has_exception())
        << "Mock did not simulate failure so future should be ok";
    // @todo: What state do we expect to be in?
 
    // Complete pending operations to avoid "leaking" mock objects
    // "Send reply"
    abort_promise.set_value({false});
 
    // Make progress
    MakeTestProgress(m_io_ctx, &abort_fut);
 
    ASSERT_TRUE(abort_fut.is_ready()) << "Cannot proceed with test since future is not ready";
    EXPECT_FALSE(abort_fut.has_exception())
        << "Mock didn't simulate failure so future should be OK";
    auto result = abort_fut.get();
    EXPECT_EQ(State::Aborted, result.state);
    EXPECT_FALSE(result.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
TEST_F(TestState, StartAsyncCompletesSuccessfully) {
    SCOPED_TRACE("Acquiring");
 
    StartDaq();
}
 
TEST_F(TestState, AbortOcmDaqControllerInStateAborting) {
    // Setup
    SCOPED_TRACE("AbortOcmDaqControllerInStateAborting");
    StartDaq();
 
    ASSERT_EQ(State::Acquiring, m_daq->GetState()) << "Test Setup failed";
 
    AbortDaq();
    ASSERT_EQ(State::Aborted, m_daq->GetState()) << "Test setup failed";
 
    // Test that abort fails if daq is already aborted
    auto fut = m_daq->AbortAsync(ErrorPolicy::Strict);
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::runtime_error);
}
 
TEST_F(TestState, AbortOcmDaqControllerInStateStarting) {
    // Setup
    SCOPED_TRACE("AbortOcmDaqControllerInStateStarting");
    StartDaq();
    ASSERT_EQ(State::Acquiring, m_daq->GetState()) << "Test Setup failed";
 
    // Test
    AbortDaq();
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
TEST_F(TestState, AbortAsyncReturnsWithErrorInsteadOfExceptionForTolerantPolicy) {
    // Setup
    SCOPED_TRACE("NewAbortSupersedesFailedAbort");
    StartDaq();
 
    boost::promise<Result<void>> abort_promise_1;
 
    // Expect two calls to abort since the first one will fail
    EXPECT_CALL(*m_mock_ops, Abort(ErrorPolicy::Tolerant, _))
        .WillOnce(Return(ByMove(abort_promise_1.get_future())));
 
    // Run
    auto fut1 = m_daq->AbortAsync(ErrorPolicy::Tolerant);
 
    EXPECT_EQ(State::AbortingAcquiring, m_daq->GetState())
        << "Expected state to be in Stopping after requesting to abort";
 
    // "Send reply1" where first source fails and second is ok.
    abort_promise_1.set_value({true});
    MakeTestProgress(m_io_ctx, &fut1);
 
    ASSERT_TRUE(fut1.has_value());
 
    auto result1 = fut1.get();
    EXPECT_EQ(State::Aborted, result1.state);
    EXPECT_TRUE(result1.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
/**
 * It is possible to abort even though an abort operation has already been started.
 * Nothing special happens in this case though.
 * @todo The command bein superseeded should probably fail.
 */
TEST_F(TestState, NewAbortSupersedesSuccessfulAbort) {
    // Setup
    SCOPED_TRACE("NewAbortSupersedesSuccessfulAbort");
    StartDaq();
 
    // First abort
    boost::promise<Result<void>> abort_promise_1;
    // Second abort
    boost::promise<Result<void>> abort_promise_2;
 
    // Expect two calls to abort
    EXPECT_CALL(*m_mock_ops, Abort(ErrorPolicy::Strict, _))
        .Times(2)
        .WillOnce(Return(ByMove(abort_promise_1.get_future())))
        .WillOnce(Return(ByMove(abort_promise_2.get_future())));
 
    // Run
    // Launch async operations concurrently
    auto fut1 = m_daq->AbortAsync(ErrorPolicy::Strict);
    auto fut2 = m_daq->AbortAsync(ErrorPolicy::Strict);
 
    EXPECT_EQ(State::AbortingAcquiring, m_daq->GetState())
        << "Expected state to be in Stopping after requesting to abort";
 
    // "Send reply1" to cause interleaving
    abort_promise_1.set_value({false});
    MakeTestProgress(m_io_ctx, &fut1);
 
    ASSERT_TRUE(fut1.is_ready());
    ASSERT_FALSE(fut1.has_exception()) << "Future has unexpected exception!";
    auto result1 = fut1.get();
    EXPECT_EQ(State::Aborted, result1.state);
    EXPECT_FALSE(result1.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
 
    // "Send reply2"
    abort_promise_2.set_value({false});
    MakeTestProgress(m_io_ctx, &fut2);
    auto result2 = fut2.get();
    EXPECT_EQ(State::Aborted, result2.state);
    EXPECT_FALSE(result2.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
TEST_F(TestState, NewAbortSupersedesFailedAbortWithStrictPolicy) {
    // Setup
    SCOPED_TRACE("NewAbortSupersedesFailedAbort");
    StartDaq();
 
    boost::promise<Result<void>> abort_promise_1;
    // Second abort
    boost::promise<Result<void>> abort_promise_2;
 
    // Expect two calls to abort since the first one will fail
    EXPECT_CALL(*m_mock_ops, Abort(ErrorPolicy::Strict, _))
        .Times(2)
        .WillOnce(Return(ByMove(abort_promise_1.get_future())))
        .WillOnce(Return(ByMove(abort_promise_2.get_future())));
 
    // Run
    auto fut1 = m_daq->AbortAsync(ErrorPolicy::Strict);
 
    EXPECT_EQ(State::AbortingAcquiring, m_daq->GetState())
        << "Expected state to be in Stopping after requesting to abort";
 
    // "Send reply1" where first source fails and second is ok.
    abort_promise_1.set_exception(DaqSourceErrors(std::vector<std::exception_ptr>()));
    MakeTestProgress(m_io_ctx, &fut1);
 
    ASSERT_TRUE(fut1.is_ready());
    ASSERT_TRUE(fut1.has_exception()) << "Future has unexpected exception!";
    EXPECT_THROW(fut1.get(), DaqSourceErrors);
    EXPECT_EQ(State::AbortingAcquiring, m_daq->GetState());
 
    // Abort again, this time it works.
    auto fut2 = m_daq->AbortAsync(ErrorPolicy::Strict);
 
    // "Send reply2"
    abort_promise_2.set_value({false});
 
    MakeTestProgress(m_io_ctx, &fut2);
    ASSERT_TRUE(fut2.has_value());
 
    auto result2 = fut2.get();
    EXPECT_EQ(State::Aborted, result2.state);
    EXPECT_FALSE(result2.error);
    EXPECT_EQ(State::Aborted, m_daq->GetState());
}
 
TEST_F(TestState, StopOcmDaqControllerSuccessfully) {
    // Setup
    SCOPED_TRACE("StopOcmDaqControllerSuccessfully");
    StartDaq();
 
    ASSERT_EQ(State::Acquiring, m_daq->GetState()) << "Test Setup failed";
 
    StopDaq();
}
 
TEST_F(TestAwait, AwaitNonExistantSourceFails) {
    // Run
    auto fut = m_daq->AwaitAsync({"non-existant"}, 0ms);
    ASSERT_TRUE(fut.has_exception());
    EXPECT_THROW(fut.get(), std::invalid_argument);
}
 
TEST_F(TestAwait, AwaitTimeout) {
    // Run
    auto fut = m_daq->AwaitAsync({"meta-source-1"}, 1ms);
    MakeTestProgress(m_io_ctx, &fut);
 
    ASSERT_TRUE(fut.has_exception());
    EXPECT_THROW(fut.get(), DaqOperationTimeout);
}
 
TEST_F(TestAwaitWithPrimSource, AwaitStopSingleSourceIsOk) {
    SCOPED_TRACE("AwaitSingleSourceIsOk");
    // Setup
    auto fut = m_daq->AwaitAsync({"meta-source-1"}, 300ms);  // large timeout needed under valgrind
    EXPECT_FALSE(fut.is_ready())
        << "The future shouldn't be ready yet as we haven't started the data acquisition!";
 
    // Run
    StartDaq();
    EXPECT_FALSE(fut.is_ready())
        << "Wait condition not fulfilled, so future should not be ready yet";
    // Stop should cancel await op
    EXPECT_CALL(m_mock_ops->mock_abort, Abort()).WillOnce(Return(true));
    StopDaq();
 
    ASSERT_TRUE(fut.is_ready());
    ASSERT_FALSE(fut.has_exception());
    auto state = fut.get();
    EXPECT_TRUE(state == State::Stopping || state == State::Stopped)
        << "Await condition should have been ready once source is stopped (meaning that DAQ is "
           "Stopping or Stopped, depending on the order of the source)";
}
 
TEST_F(TestAwaitWithPrimSource, AwaitAbortAllSources) {
    SCOPED_TRACE("AwaitAbortAllMetadataSources");
    // Setup
    auto fut = m_daq->AwaitAsync({"meta-source-1", "meta-source-2"}, 150ms);
    EXPECT_FALSE(fut.is_ready())
        << "The future shouldn't be ready yet as we haven't started the data acquisition!";
 
    // Run
    StartDaq();
    EXPECT_FALSE(fut.is_ready());
 
    // Abort should cancel await op
    EXPECT_CALL(m_mock_ops->mock_abort, Abort()).WillOnce(Return(true));
    AbortDaq();
 
    EXPECT_TRUE(fut.is_ready());
    ASSERT_FALSE(fut.has_exception());
    auto state = fut.get();
    EXPECT_TRUE(state == State::AbortingAcquiring || state == State::Aborted)
        << "Await condition should have been ready once source is stopped (meaning that DAQ is "
           "Aborting or Aborted, depending on the order of the source)";
}
 
TEST_F(TestAwait, AwaitStopSingleSourceWhenConditionIsFulfilled) {
    SCOPED_TRACE("AwaitStopSingleSourceWhenConditionIsFulfilled");
    // Setup
    // Run
    StartDaq();
    StopDaq();
 
    auto fut = m_daq->AwaitAsync({"meta-source-1"}, 150ms);
 
    EXPECT_TRUE(fut.is_ready()) << "Condition already fulfilled so future should be ready";
    ASSERT_FALSE(fut.has_exception());
    EXPECT_EQ(State::Stopped, fut.get());
}
 
TEST_F(TestNotStarted, CanUpdateKeywords) {
    // Setup
 
    // Run
    m_daq->UpdateKeywords(m_keywords);
 
    EXPECT_EQ(m_keywords, m_daq->GetContext().keywords);
}
 
TEST_F(TestStopped, CannotUpdateKeywordsInStopped) {
    // Setup
 
    // Run
    EXPECT_THROW(m_daq->UpdateKeywords(m_keywords), std::runtime_error);
}
 
TEST_F(TestAwaitWithPrimSource, StartWillAwait) {
    SCOPED_TRACE("StartWillAwait");
    // Setup
    // Run
    StartDaq();
}
 
/**
 * Tests that DaqController automatically stops DAQ when the await-op completes.
 */
TEST_F(TestAwaitWithPrimSource, AutomaticStop) {
    SCOPED_TRACE("AutomaticStop");
    // Setup
    StartDaq();
 
    // Run
    // DaqController is monitoring the completion of all primary data sources.
    // By setting the value we simulate that completion. This should then trigger StopDaq.
    DpPart prim_part{"s1", "/tmp/file.fits"};
    m_await_promise.set_value({false, {prim_part}});
 
    // Setup expectations for stopping
    Result<DpParts> stop_op_reply{false, m_files};
    EXPECT_CALL(*m_mock_ops, Stop(ErrorPolicy::Strict, _))
        .WillOnce(Return(ByMove(boost::make_ready_future<Result<DpParts>>(stop_op_reply))));
    ASSERT_EQ(State::Acquiring, this->m_status->GetState());
    EXPECT_EQ(0u, m_daq->GetContext().results.size());
 
    // There's no future to await-on, so we run until the observed state changes instead.
    MakeTestProgressUntil(
        m_io_ctx, [this]() -> bool { return this->m_status->GetState() == State::Stopped; });
    EXPECT_FALSE(m_status->GetError());
    EXPECT_EQ(2u, m_daq->GetContext().results.size()) << "One from m_files (metadata), "
                                                         "one from primary";
    EXPECT_THAT(m_daq->GetContext().results, Contains(prim_part));
}