testDpmDaqController.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::DpmDaqController`
 */
#include <gtest/gtest.h>
 
#include <daq/daqController.hpp>
#include <mal/MalException.hpp>
 
#include "mock/dpmClientMock.hpp"
#include "statusObserver.hpp"
#include "utils.hpp"
 
using namespace ::testing;
using namespace std::chrono;
 
namespace daq {
/**
 * Fixture for daq::DaqController life cycle tests
 *
 * @ingroup daq_ocm_libdaq_test
 */
class TestDpmDaqController : public ::testing::Test {
public:
    TestDpmDaqController()
        : m_io_ctx()
        , m_status(std::make_shared<ObservableStatus>("id", "fileid"))
        , m_event_log(std::make_shared<ObservableEventLog>()) {
        m_props.id = "id";
        auto& s = m_props.meta_sources.emplace_back();
        s.name = "meta";
        s.rr_uri = "zpb.rr://127.0.0.1/daq";
        m_props.results.emplace_back("location", "path");
 
        // Default state
        m_status->SetState(State::NotScheduled);
    }
 
    void SetUp() override {
        m_dpm_status = *m_status;
        m_dpm_client = std::make_shared<DpmClientMock>();
    }
 
    void PostSetUp() {
        m_controller = std::make_shared<DpmDaqController>(
            m_io_ctx, m_props, m_status, m_event_log, m_dpm_client);
    }
 
    boost::asio::io_context m_io_ctx;                 // NOLINT
    std::shared_ptr<ObservableStatus> m_status;       // NOLINT
    std::shared_ptr<ObservableEventLog> m_event_log;  // NOLINT
    DaqContext m_props;                            // NOLINT
    std::shared_ptr<DpmClientMock> m_dpm_client;      // NOLINT
    /**
     * Fake status update sample from DPM
     */
    Status m_dpm_status = Status("id", "fileid");
 
    std::shared_ptr<DpmDaqController> m_controller;
};
 
TEST_F(TestDpmDaqController, StatusUpdateInNotScheduledSucceeds) {
    // Setup 
    // New state will be Scheduled
    m_dpm_status.state = State::Scheduled;
    PostSetUp();
 
    // Test
    EXPECT_EQ(m_controller->GetState(), State::NotScheduled);
    m_dpm_client->status_signal(m_dpm_status);
    EXPECT_EQ(m_controller->GetState(), State::Scheduled)
        << "Expected new state from status update signal";
}
 
/**
 * Test that DPM can override status of OCM as it is the authoratitive source of DAQ status.
 */
TEST_F(TestDpmDaqController, StatusOverrideFromOcmAbortedToDpmMergingSucceeds) {
    // Setup 
    // Put DAQ it in state Aborted with error from OCM point-of-view.
    m_status->SetState(State::Aborted, true);
    // The DPM state is Merging however.
    m_dpm_status.state = State::Merging;
    m_dpm_status.timestamp = Status::Clock::now();
    PostSetUp();
 
    // Test
    // Update timestamp of status so it looks new
    m_dpm_status.timestamp = Status::Clock::now();
    EXPECT_EQ(m_controller->GetState(), State::Aborted);
    m_dpm_client->status_signal(m_dpm_status);
    EXPECT_EQ(m_controller->GetState(), State::Merging)
        << "Expected new state from status update signal";
    EXPECT_FALSE(m_controller->GetErrorFlag());
}
 
TEST_F(TestDpmDaqController, StartAsyncThrows) {
    PostSetUp();
 
    auto fut = m_controller->StartAsync();
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::runtime_error);
}
 
TEST_F(TestDpmDaqController, StopAsyncThrows) {
    PostSetUp();
 
    auto fut = m_controller->StopAsync(ErrorPolicy::Tolerant);
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::runtime_error);
}
 
 
TEST_F(TestDpmDaqController, UpdateKeywordsThrows) {
    PostSetUp();
 
    EXPECT_THROW(m_controller->UpdateKeywords({}), std::runtime_error);
}
 
TEST_F(TestDpmDaqController, ScheduleMergeAsyncSucceedsIfDpmSucceeds) {
    // Setup
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, ScheduleAsync(_)).WillOnce(InvokeWithoutArgs([&] {
        return reply.get_future();
    }));
 
    // Test
    auto fut = m_controller->ScheduleMergeAsync();
 
    // Fake reply
    reply.set_value(State::Scheduled);
    MakeTestProgress(m_io_ctx, &fut);
 
    EXPECT_TRUE(fut.is_ready());
    auto result = fut.get();
    EXPECT_EQ(State::Scheduled, result);
    EXPECT_EQ(State::Scheduled, m_controller->GetState());
}
 
TEST_F(TestDpmDaqController,
       ScheduleMergeAsyncSucceedsIfDpmSucceedsWithStatusSignalReceivedBeforeReply) {
    // Setup
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, ScheduleAsync(_)).WillOnce(InvokeWithoutArgs([&] {
        return reply.get_future();
    }));
 
    // Test
    auto fut = m_controller->ScheduleMergeAsync();
 
    // Simulate status update from DPM before reply being received
    m_dpm_status.state = State::Scheduled;
    m_dpm_client->status_signal(m_dpm_status);
    EXPECT_EQ(State::Scheduled, m_controller->GetState());
 
    // Fake reply
    reply.set_value(State::Scheduled);
    MakeTestProgress(m_io_ctx, &fut);
 
    EXPECT_TRUE(fut.is_ready());
    auto result = fut.get();
    EXPECT_EQ(State::Scheduled, result);
    EXPECT_EQ(State::Scheduled, m_controller->GetState());
}
 
TEST_F(TestDpmDaqController, ScheduleMergeAsyncFailsIfDpmFails) {
    // Setup
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, ScheduleAsync(_)).WillOnce(InvokeWithoutArgs([&] {
        return reply.get_future();
    }));
 
    // Test
    auto fut = m_controller->ScheduleMergeAsync();
 
    // Fake reply
    reply.set_exception(std::runtime_error("some_error"));
    MakeTestProgress(m_io_ctx, &fut);
 
    EXPECT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::exception);
    EXPECT_EQ(State::NotScheduled, m_controller->GetState());
    EXPECT_TRUE(m_controller->GetStatus()->GetError());
}
 
TEST_F(TestDpmDaqController, ScheduleMergeAsyncFailsIfTimeout) {
    // Setup
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, ScheduleAsync(_)).WillOnce(InvokeWithoutArgs([&] {
        return reply.get_future();
    }));
 
    // Test
    auto fut = m_controller->ScheduleMergeAsync();
    reply.set_exception(elt::mal::TimeoutException("TIMEOUT"));
    MakeTestProgress(m_io_ctx, &fut);
    EXPECT_TRUE(fut.is_ready());
 
    // Fake reply
    EXPECT_THROW(fut.get(), elt::mal::TimeoutException);
    EXPECT_EQ(State::NotScheduled, m_controller->GetState()) << "State shoudln't have changed";
    EXPECT_FALSE(m_controller->GetStatus()->GetError())
        << "Error flag should not be set for timeouts";
}
 
TEST_F(TestDpmDaqController, ScheduleMergeAsyncFailsIfAlreadyScheduled) {
    // Setup
    m_status->SetState(State::Scheduled);
    PostSetUp();
 
    // Test
    auto fut = m_controller->ScheduleMergeAsync();
    EXPECT_TRUE(fut.is_ready());
 
    // Fake reply
    EXPECT_THROW(fut.get(), std::runtime_error);
    EXPECT_EQ(State::Scheduled, m_controller->GetState());
    EXPECT_FALSE(m_controller->GetStatus()->GetError());
}
 
TEST_F(TestDpmDaqController, AbortAsyncAbortImmediatelyIfNoPendingRequestsExist) {
    // Setup
    PostSetUp();
 
    // Run
    auto fut = m_controller->AbortAsync(ErrorPolicy::Strict);
 
    EXPECT_TRUE(fut.is_ready());
    auto result = fut.get();
    EXPECT_EQ(State::Aborted, result.state);
    EXPECT_FALSE(result.error);
    EXPECT_EQ(State::Aborted, m_controller->GetState());
}
 
TEST_F(TestDpmDaqController, AbortAsyncWithStrictPolicyAbortsIfDpmAborts) {
    // Setup
    // Put it in state Scheduled or after
    m_status->SetState(State::Scheduled);
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, AbortAsync("id"))
        .WillOnce(InvokeWithoutArgs([&] { return reply.get_future(); }));
 
    // Run
    auto fut = m_controller->AbortAsync(ErrorPolicy::Strict);
 
    // Fake successful reply
    reply.set_value(State::Aborted);
 
    // Progress test
    MakeTestProgress(m_io_ctx, &fut);
 
    EXPECT_TRUE(fut.is_ready());
    auto result = fut.get();
    EXPECT_EQ(State::Aborted, result.state);
    EXPECT_FALSE(result.error);
    EXPECT_EQ(State::Aborted, m_controller->GetState());
}
 
TEST_F(TestDpmDaqController, AbortAsyncWithStrictPolicyDoesNothingIfDpmAbortFails) {
    // Setup
    m_status->SetState(State::Scheduled);
    PostSetUp();
 
    using R = State;
    boost::promise<R> reply;
    EXPECT_CALL(*m_dpm_client, AbortAsync("id"))
        .WillOnce(InvokeWithoutArgs([&] { return reply.get_future(); }));
 
    // Run
    auto fut = m_controller->AbortAsync(ErrorPolicy::Strict);
 
    // Fake reply
    reply.set_exception(std::runtime_error("ERROR"));
 
    // Progress test
    MakeTestProgress(m_io_ctx, &fut);
 
    EXPECT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), std::runtime_error);
    EXPECT_EQ(State::Scheduled, m_controller->GetState());
    EXPECT_FALSE(m_controller->GetStatus()->GetError())
        << "Failed to abort is not a condition for marking DAQ as error";
}
 
} // namespace daq