testAsyncOpAwaitPrim.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Unit test for op::AwaitPrimAsync
 */
#include <daq/config.hpp>
 
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <log4cplus/logger.h>
 
#include <daq/error.hpp>
#include <daq/op/awaitPrim.hpp>
#include <daq/op/initiate.hpp>
 
#include "testAsyncOpBase.hpp"
#include "utils.hpp"
 
using namespace daq;
using namespace ::testing;
 
namespace {
 
class RecStatusFake : public recif::RecStatus {
public:
    RecStatusFake(std::string id, recif::RecStatusNames status, std::vector<std::string> files)
        : m_id(std::move(id)), m_status(status), m_files(std::move(files)) {
    }
    virtual std::vector<std::string> getDpFiles() const override {
        return m_files;
    }
    virtual void setDpFiles(const std::vector<std::string>& dp_files) override {
        m_files = dp_files;
    }
 
    virtual double getEndTime() const override {
        return 0.0;
    }
    virtual void setEndTime(double end_time) override {
    }
 
    virtual int32_t getFilesGenerated() const override {
        return m_files.size();
    }
    virtual void setFilesGenerated(int32_t files_generated) override {
    }
 
    virtual int32_t getFramesProcessed() const override {
        return 1;
    }
    virtual void setFramesProcessed(int32_t frames_processed) override {
    }
 
    virtual int32_t getFramesRemaining() const override {
        return 0;
    }
    virtual void setFramesRemaining(int32_t frames_remaining) override {
    }
 
    virtual std::string getId() const override {
        return m_id;
    }
    virtual void setId(const std::string& id) override {
        m_id = id;
    }
 
    virtual std::string getInfo() const override {
        return {};
    }
    virtual void setInfo(const std::string& info) override {
    }
 
    virtual double getRemainingTime() const override {
        return 1.0;
    }
    virtual void setRemainingTime(double remaining_time) override {
    }
 
    virtual int64_t getSizeRecorded() const override {
        return 1;
    }
    virtual void setSizeRecorded(int64_t size_recorded) override {
    }
 
    virtual double getStartTime() const override {
        return 1.0;
    }
    virtual void setStartTime(double start_time) override {
    }
 
    virtual ::recif::RecStatusNames getStatus() const override {
        return m_status;
    }
    virtual void setStatus(::recif::RecStatusNames status) override {
        m_status = status;
    }
 
    virtual double getTimeElapsed() const override {
        return 1.0;
    }
    virtual void setTimeElapsed(double time_elapsed) override {
    }
    bool hasKey() const override {
        return false;
    }
    std::unique_ptr<recif::RecStatus> cloneKey() const override {
        throw std::runtime_error("not clonable");
    }
    std::unique_ptr<recif::RecStatus> clone() const override {
        throw std::runtime_error("not clonable");
    }
    bool keyEquals(const recif::RecStatus& other) const override {
        return false;
    }
 
private:
    std::string m_id;
    recif::RecStatusNames m_status;
    std::vector<std::string> m_files;
};
 
class RecWaitStatusFake : public recif::RecWaitStatus {
public:
    RecWaitStatusFake(recif::RecWaitStatusNames wait_status,
                      std::shared_ptr<recif::RecStatus> rec_status)
        : m_wait_status{wait_status}, m_rec_status{std::move(rec_status)} {
    }
    std::shared_ptr<recif::RecStatus> getRecStat() const override {
        return m_rec_status;
    }
    // Does not exist in MAL for Devenv 3
    // NOLINTNEXTLINE
    void setRecStat(const std::shared_ptr<::recif::RecStatus>& status) {
        m_rec_status = status;
    }
 
    recif::RecWaitStatusNames getStatus() const override {
        return m_wait_status;
    }
    void setStatus(recif::RecWaitStatusNames wait_status) override {
        m_wait_status = wait_status;
    }
    bool hasKey() const override {
        return false;
    }
    std::unique_ptr<recif::RecWaitStatus> cloneKey() const override {
        throw std::runtime_error("not clonable");
    }
    std::unique_ptr<recif::RecWaitStatus> clone() const override {
        throw std::runtime_error("not clonable");
    }
    bool keyEquals(const recif::RecWaitStatus& other) const override {
        return false;
    }
 
private:
    recif::RecWaitStatusNames m_wait_status;
    std::shared_ptr<recif::RecStatus> m_rec_status;
};
}  // namespace
/**
 * @ingroup daq_ocm_libdaq_test
 */
struct TestAsyncOpAwaitPrim : TestAsyncOpBase {
    std::shared_ptr<recif::RecWaitStatus> MakeWaitNotCompletedStatus() {
        auto status = std::make_shared<RecWaitStatusFake>(
            recif::Timeout,
            std::make_shared<RecStatusFake>(
                m_id, recif::RecStatusNames::Active, std::vector<std::string>()));
        return status;
    }
    std::shared_ptr<recif::RecWaitStatus> MakeWaitCompletedStatus() {
        auto status = std::make_shared<RecWaitStatusFake>(
            recif::Success,
            std::make_shared<RecStatusFake>(
                m_id,
                recif::RecStatusNames::Completed,
                std::vector<std::string>{"/file1.fits", "/file2.fits"}));
        return status;
    }
 
    daq::op::AwaitOpParams MakeAwaitOpParams() {
        return daq::op::AwaitOpParams(MakeParams(), std::chrono::milliseconds(0));
    }
};
 
TEST_F(TestAsyncOpAwaitPrim, AwaitCompletesOnFirstRequest) {
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2;
    auto reply_1 = MakeWaitCompletedStatus();
    auto reply_2 = MakeWaitCompletedStatus();
    EXPECT_CALL(*m_prim_rr_client, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_1.get_future())));
    EXPECT_CALL(*m_prim_rr_client2, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_2.get_future())));
 
    // Run
    auto fut = op::InitiateOperation<op::AwaitPrimAsync>(m_executor, MakeAwaitOpParams());
    // "Send replies"
    prim_promise_1.set_value(reply_1);
    prim_promise_2.set_value(reply_2);
 
    // 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_FALSE(result.error) << "There should have been no errors as each request was successful";
    EXPECT_EQ(result.result.size(), 4u) << "Each primary source returned two files by default";
}
 
TEST_F(TestAsyncOpAwaitPrim, AwaitCompletesOnSecondRequest) {
    // Only the second source is completed on second attempt
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_2;
    auto reply_1 = MakeWaitCompletedStatus();
    auto reply_2_1 = MakeWaitNotCompletedStatus();
    auto reply_2_2 = MakeWaitCompletedStatus();
    EXPECT_CALL(*m_prim_rr_client, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_1.get_future())));
    EXPECT_CALL(*m_prim_rr_client2, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_2_1.get_future())))
        .WillOnce(Return(ByMove(prim_promise_2_2.get_future())));
 
    // Run
    auto fut = op::InitiateOperation<op::AwaitPrimAsync>(m_executor, MakeAwaitOpParams());
    // "Send replies"
    prim_promise_1.set_value(reply_1);
    prim_promise_2_1.set_value(reply_2_1);
 
    // Run ready handlers
    m_io_ctx.poll();
    ASSERT_FALSE(fut.is_ready());
 
    prim_promise_2_2.set_value(reply_2_2);
 
    // 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_FALSE(result.error) << "There should have been no errors as each request was successful";
    EXPECT_EQ(result.result.size(), 4u) << "Each primary source returned two files by default";
}
 
TEST_F(TestAsyncOpAwaitPrim, SourceThrowsExceptionIsRetriedSuccessfully) {
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_1;
    // Source 2 will first throw exception, which should be retried and second attempt will be
    // succesful.
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_2;
    auto reply_1 = MakeWaitCompletedStatus();
    auto reply_2_2 = MakeWaitCompletedStatus();
    EXPECT_CALL(*m_prim_rr_client, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_1.get_future())));
    EXPECT_CALL(*m_prim_rr_client2, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_2_1.get_future())))
        .WillOnce(Return(ByMove(prim_promise_2_2.get_future())));
 
    // Run
    ASSERT_TRUE(m_status->GetAlerts().empty());
 
    auto fut = op::InitiateOperation<op::AwaitPrimAsync>(m_executor, MakeAwaitOpParams());
    // "Send replies"
    prim_promise_1.set_value(reply_1);
    prim_promise_2_1.set_exception(recif::ExceptionErr("random error", 1));
 
    // Run ready handlers
    m_io_ctx.poll();
    ASSERT_FALSE(fut.is_ready());
    // Status should now contain alert
    ASSERT_EQ(m_status->GetAlerts().size(), 1u);
    auto alert = m_status->GetAlerts()[0];
    EXPECT_THAT(alert.description, HasSubstr("prim-source-2"));
 
    prim_promise_2_2.set_value(reply_2_2);
 
    // 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_TRUE(result.error) << "A erors as each request was successful";
    EXPECT_EQ(result.result.size(), 4u) << "Each primary source returned two files by default";
    EXPECT_TRUE(m_status->GetAlerts().empty())
        << "Alert should have cleared after successful retry";
}
 
TEST_F(TestAsyncOpAwaitPrim, SourceThrowsExceptionIsAbortedSuccessfully) {
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_1;
    boost::promise<std::shared_ptr<recif::RecWaitStatus>> prim_promise_2_2;
    auto reply_1 = MakeWaitCompletedStatus();
    auto reply_2_2 = MakeWaitCompletedStatus();
    EXPECT_CALL(*m_prim_rr_client, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_1.get_future())));
    EXPECT_CALL(*m_prim_rr_client2, RecWait(_))
        .WillOnce(Return(ByMove(prim_promise_2_1.get_future())));
 
    // Run
    auto [fut, abort] =
        op::InitiateAbortableOperation<op::AwaitPrimAsync>(m_executor, MakeAwaitOpParams());
    // "Send replies"
    prim_promise_1.set_value(reply_1);
    prim_promise_2_1.set_exception(recif::ExceptionErr("random error", 1));
 
    // Run ready handlers
    m_io_ctx.poll();
    ASSERT_FALSE(fut.is_ready());
 
    // Abort operation. This should set the promise (but might not be ready immedately due to
    // continuations)
    EXPECT_TRUE(abort());
    // Execute handlers
    MakeTestProgress(m_io_ctx, &fut);
    EXPECT_TRUE(fut.is_ready()) << "aborting operation should immediately set operation result";
 
    ASSERT_FALSE(fut.has_exception()) << "Future has unexpected exception!";
    auto result = fut.get();
    EXPECT_TRUE(result.error) << "Operation was completed with error (exception from source)";
    EXPECT_EQ(result.result.size(), 2u) << "Only result from first source was received before "
                                           "being aborted";
 
    // Send reply to trigger cleanup of pending continuations
    prim_promise_2_2.set_value(reply_1);
    MakeTestProgress(m_io_ctx);
}