testOcmDaqService.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_server_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Unit tests for OcmDaqService
 */
#include <ocmDaqService.hpp>
 
#include <fmt/format.h>
#include <gtest/gtest.h>
 
#include <daq/error.hpp>
#include <malMock.hpp>
#include <metadaqifMock.hpp>
#include <recifMock.hpp>
 
#include "daqifFake.hpp"
#include "mock/daqifMock.hpp"
#include "mock/managerMock.hpp"
 
using namespace ::testing;
using namespace std::literals;
 
/**
 * Fixture that sets up a OcmDaqService instance with following mockups:
 * - mal::Mal
 * - daq::Manager
 *
 * And prepared request string containing metadata sources
 *
 * @ingroup daq_ocm_server_test
 */
struct TestOcmDaqService : Test {
    TestOcmDaqService()
        : m_io_ctx()
        , m_mal_mock()
        , m_mgr_mock()
        , m_event_log(std::make_shared<daq::ObservableEventLog>())
        , m_daq_impl(std::make_shared<OcmDaqService>(
              m_io_ctx, m_mal_mock, m_mgr_mock, "ocm", "/tmp", m_event_log)) {
    }
    void SetUp() override {
        m_metadata_sources = "meta@zpb.rr://uri";
        m_prim_sources = "prim@zpb.rr://uri";
        m_daq_properties = R"(
        {
            "keywords": [
                    {
                        "type": "esoKeyword",
                        "name": "FOO BAR",
                        "value": true
                    },
                    {
                        "type": "valueKeyword",
                        "name": "FOOBAR",
                        "value": true
                    }
                ]
        }
        )";
        m_daqv2_spec = R"(
        {
          "id": "id",
          "sources": [
            {
              "type": "metadataSource",
              "sourceName": "name",
              "rrUri": "uri"
            }
          ]
        }
        )";
    }
    void TearDown() override {
    }
 
    boost::asio::io_context m_io_ctx;  // NOLINT
    MalMock m_mal_mock;                // NOLINT
    ManagerMock m_mgr_mock;            // NOLINT
    std::shared_ptr<daq::ObservableEventLog> m_event_log;
    std::shared_ptr<OcmDaqService> m_daq_impl;  // NOLINT
    std::string m_prim_sources;                 // NOLINT
    std::string m_metadata_sources;             // NOLINT
    std::string m_daq_properties;               // NOLINT
    std::string m_daqv2_spec;                   // NOLINT
};
 
/**
 * Fixture for testing when OcmDaqService is abandoned
 * (e.g. when service is de-registered).
 *
 * @ingroup daq_ocm_server_test
 */
struct TestOcmDaqServiceAbandoned : TestOcmDaqService {};
 
TEST(TestParseSingleSource, Successful) {
    ParsedSource expected("name", "zpb://rr.uri");
 
    ParsedSource parsed = ParseSourceUri(fmt::format("{}@{}", expected.name, expected.rr_uri));
    EXPECT_EQ(expected, parsed);
}
 
TEST(TestParseSingleSource, LeadingOrTrailingSpacesAreAllowed) {
    ParsedSource expected("name", "zpb://rr.uri");
    EXPECT_EQ(expected, ParseSourceUri(fmt::format("{}@{} ", expected.name, expected.rr_uri)));
    EXPECT_EQ(expected, ParseSourceUri(fmt::format(" {}@{}", expected.name, expected.rr_uri)));
    EXPECT_EQ(expected, ParseSourceUri(fmt::format(" {}@{} ", expected.name, expected.rr_uri)));
}
 
TEST(TestParseSingleSource, Incomplete) {
    ParsedSource expected("name", "zpb://rr.uri");
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{} ", expected.name, expected.rr_uri)),
                 std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{} ", expected.name, expected.rr_uri)),
                 std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("@,")), std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("{},{}", expected.name, expected.rr_uri)),
                 std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{}", expected.name, expected.rr_uri)),
                 std::invalid_argument);
}
 
TEST(TestParseMultipleSources, Successful) {
    ParsedSource expected1("name", "zpb://rr.uri");
    ParsedSource expected2("name", "zpb://rr.uri");
    std::vector<ParsedSource> result{expected1, expected2};
 
    EXPECT_EQ(
        result,
        ParseSourceUris(fmt::format(
            "{}@{} {}@{}", expected1.name, expected1.rr_uri, expected2.name, expected2.rr_uri)));
    EXPECT_EQ(
        result,
        ParseSourceUris(fmt::format(
            "{}@{}  {}@{}", expected1.name, expected1.rr_uri, expected2.name, expected2.rr_uri)));
 
    EXPECT_EQ(
        result,
        ParseSourceUris(fmt::format(
            " {}@{}  {}@{} ", expected1.name, expected1.rr_uri, expected2.name, expected2.rr_uri)));
}
 
TEST(TestParseSingleSource, Empty) {
    ParsedSource expected("name", "zpb://rr.uri");
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{} ", "", expected.rr_uri)),
                 std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{} ", expected.name, "")), std::invalid_argument);
    EXPECT_THROW(ParseSourceUri(fmt::format("{}@,{} ", expected.name, expected.rr_uri)),
                 std::invalid_argument);
}
 
TEST(TestParseDaqContext, Successful) {
    auto max_err_ms = 0.001;
    {
        auto properties_str = R"(
            {
                "keywords": [
                        {
                            "type": "esoKeyword",
                            "name": "FOO BAR",
                            "value": true
                        },
                        {
                            "type": "valueKeyword",
                            "name": "FOOBAR",
                            "value": true
                        }
                    ],
                "awaitInterval": 0.1
            }
            )";
        auto properties = ParseStartDaqContext(properties_str);
        EXPECT_THAT(static_cast<double>(properties.await_interval.count()),
                    DoubleNear(100, max_err_ms));
    }
    {
        auto properties_str = R"(
            {
                "awaitInterval": 1
            }
            )";
        auto properties = ParseStartDaqContext(properties_str);
        EXPECT_THAT(static_cast<double>(properties.await_interval.count()),
                    DoubleNear(1000, max_err_ms));
    }
}
 
TEST(TestParseDaqContext, Failures) {
    EXPECT_THROW(ParseStartDaqContext(R"({"awaitInterval": "0.1"})"), std::invalid_argument);
    EXPECT_THROW(ParseStartDaqContext(R"({"awaitInterval": -0.1})"), std::invalid_argument);
}
 
TEST_F(TestOcmDaqService, StartDaqShouldTestIfIdAlreadyExistsAndReturnFailureIfItDoes) {
    // Setup
    // Setup so that ID already exist
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(true));
    // MakeDaqId is possibly called since a file_id is required.
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).Times(Between(0, 1)).WillOnce(Return("file-id"));
 
    // Run
    auto fut = m_daq_impl->StartDaq("id", "prefix", m_prim_sources, m_metadata_sources, "");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready by now since daq with id already exist";
    EXPECT_THROW(fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StartDaqShouldCreateAndAssignIdIfNotProvided) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);
 
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).WillOnce(Return("id"));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(Field(&daq::DaqContext::id, std::string("id"))))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::State>(daq::State::Acquiring))));
    // Have to construct and return void* and still keep a reference to it..
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto fut = m_daq_impl->StartDaq("", "prefix", m_prim_sources, m_metadata_sources, "");
    // Requred to trigger any .then continuations, even though future is ready
    m_io_ctx.poll();
 
    ASSERT_TRUE(fut.is_ready()) << "future should be ready immediately since we faked a "
                                   "synchronous start with make_ready_future";
    EXPECT_TRUE(fut.has_value());
    auto reply = fut.get();
    ASSERT_TRUE(reply);
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, StartDaqV2ShouldTestIfIdAlreadyExistsAndReturnFailureIfItDoes) {
    // Setup
    // Setup so that ID already exist
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(true));
    // MakeDaqId is possibly called since a file_id is required.
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).Times(Between(0, 1)).WillOnce(Return("file-id"));
 
    // Run
    auto fut = m_daq_impl->StartDaqV2(m_daqv2_spec);
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready by now since daq with id already exist";
    EXPECT_THROW(fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StartDaqFailsIfJsonIsInvalid) {
    // Setup
 
    // Run
    auto fut = m_daq_impl->StartDaq("id", "prefix", "", m_metadata_sources, "INVALID JSON");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready by now since daq with id already exist";
    EXPECT_THROW(fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StartDaqFailsIfJsonSchemaIsInvalid) {
    // Setup
 
    // Run
    // note: Expects JSON object
    auto fut = m_daq_impl->StartDaq("id", "prefix", "", m_metadata_sources, "[]");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StartDaqFailsIfPrefixContainsParentPaths) {
    // Setup
 
    // Run
    // note: Expects JSON object
    auto fut = m_daq_impl->StartDaq("id", "/path/prefix", "", m_metadata_sources, m_daq_properties);
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready());
    EXPECT_THROW(fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StartDaqShouldAddAndStartDaqControllerIfArgumentsAreOk) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);
 
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).Times(Between(0, 1)).WillOnce(Return("file-id"));
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(Field(&daq::DaqContext::id, "id")))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::State>(daq::State::Acquiring))));
    // Have to construct and return void* and still keep a reference to it..
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto fut = m_daq_impl->StartDaq("id", "prefix", "", m_metadata_sources, m_daq_properties);
 
    // Requred to trigger any .then continuations, even though future is ready
    m_io_ctx.poll();
 
    ASSERT_TRUE(fut.is_ready()) << "future should be ready immediately since we faked a "
                                   "synchronous start with make_ready_future";
    EXPECT_TRUE(fut.has_value());
    auto reply = fut.get();
    ASSERT_TRUE(reply);
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, StartDaqShouldFailIfAddDaqFails) {
    // Setup
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).Times(Between(0, 1)).WillOnce(Return("file-id"));
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(_)).WillOnce(Throw(std::runtime_error("error")));
 
    // Run
    auto fut = m_daq_impl->StartDaq("id", "prefix", "", m_metadata_sources, "");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready immediately since we faked a "
                                   "synchronous start with make_ready_future";
    EXPECT_THROW(fut.get(), daqif::DaqException)
        << "future should have contained the ICD exception type";
}
 
TEST_F(TestOcmDaqService, StartDaqShouldFailIfManagerStartFails) {
    // Setup
 
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).WillOnce(Return("fileid"));
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(Field(&daq::DaqContext::id, "id")))
        .WillOnce(Return(ByMove(boost::make_exceptional_future<daq::State>(
            daq::DaqSourceErrors({daq::DaqSourceError("start", "source", "message")})))));
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv)).WillOnce(Return(daq::Status()));
 
    // Run
    auto fut = m_daq_impl->StartDaq("id", "prefix", "", m_metadata_sources, "");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready())
        << "future should be ready now since we faked a synchronous start with make_ready_future";
    EXPECT_THROW(fut.get(), daqif::DaqException)
        << "future should have contained the ICD exception type";
}
 
TEST_F(TestOcmDaqService, StartDaqV2ShouldCreateAndAssignIdIfNotProvided) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);
 
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).WillOnce(Return("id"));
    // Impl should check if id is already used.
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(Field(&daq::DaqContext::id, std::string("id"))))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::State>(daq::State::Acquiring))));
    // Have to construct and return void* and still keep a reference to it..
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
 
    auto fut = m_daq_impl->StartDaqV2(R"(
    {
      "sources": [
        {
          "type": "metadataSource",
          "sourceName": "name",
          "rrUri": "uri"
        }
      ]
    }
    )");
    // Requred to trigger any .then continuations, even though future is ready
    m_io_ctx.poll();
 
    ASSERT_TRUE(fut.is_ready()) << "future should be ready immediately since we faked a "
                                   "synchronous start with make_ready_future";
    try {
        EXPECT_TRUE(fut.has_value());
        auto reply = fut.get();
        ASSERT_TRUE(reply);
        EXPECT_EQ(reply->getId(), "id");
    } catch (daqif::DaqException const& e) {
        FAIL() << e.getMessage();
    } catch (std::exception const& e) {
        FAIL() << e.what();
    }
}
 
TEST_F(TestOcmDaqService, StartDaqV2FailsIfJsonParsingFails) {
    // Setup
 
    // Run
    auto fut = m_daq_impl->StartDaqV2(",]{");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready now as parsing failed";
    EXPECT_THROW(fut.get(), daqif::DaqException)
        << "future should have contained the ICD exception type";
}
 
TEST_F(TestOcmDaqService, StartDaqV2FailsIfSchemaParsingFails) {
    // Setup
 
    // Run
    auto fut = m_daq_impl->StartDaqV2("{}");
    m_io_ctx.poll();
    ASSERT_TRUE(fut.is_ready()) << "future should be ready now as schema validation failed";
    EXPECT_THROW(fut.get(), daqif::DaqException)
        << "future should have contained the ICD exception type";
}
 
TEST_F(TestOcmDaqService, StartDaqV2ShouldAddAndStartDaqControllerIfArgumentsAreOk) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);
 
    // MakeDaqId is used for file-id anyway
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).WillOnce(Return("file-id"));
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock,
                StartDaqAsync(AllOf(Field(&daq::DaqContext::id, "id"),
                                    Field(&daq::DaqContext::file_id, "file-id"))))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::State>(daq::State::Acquiring))));
    // Have to construct and return void* and still keep a reference to it..
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto fut = m_daq_impl->StartDaqV2(m_daqv2_spec);
 
    // Requred to trigger any .then continuations, even though future is ready
    m_io_ctx.poll();
 
    ASSERT_TRUE(fut.is_ready()) << "future should be ready immediately since we faked a "
                                   "synchronous start with make_ready_future";
    EXPECT_TRUE(fut.has_value());
    auto reply = fut.get();
    ASSERT_TRUE(reply);
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, StopDaqShouldFailIfDaqDoesNotExist) {
    // Setup
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(false));
    // Implementation might, or might not stop directly, leaving the id check to mgr
    EXPECT_CALL(m_mgr_mock, StopDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .Times(Between(0, 1))
        .WillOnce(Return(ByMove(
            boost::make_exceptional_future<daq::Status>(std::invalid_argument("no such id")))));
 
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv)).WillOnce(Throw(std::invalid_argument("no such id")));
 
    // Run
    auto reply_fut = m_daq_impl->StopDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, StopDaqShouldSucceedIfMgrOpSuceeds) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);  // Note: Raw pointer since MAL use unsafe APIs
    daq::Status status("id", "fileid", daq::State::Stopped, false, daq::Status::TimePoint());
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, StopDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::Status>(status))));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->StopDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqReply> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, ForceStopDaqShouldSucceedIfMgrOpSuceeds) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);  // Note: Raw pointer since MAL use unsafe APIs
    daq::Status status("id", "fileid", daq::State::Stopped, false, daq::Status::TimePoint());
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, StopDaqAsync("id"sv, daq::ErrorPolicy::Tolerant))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::Status>(status))));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->ForceStopDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqReply> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, AbortDaqShouldFailIfDaqDoesNotExist) {
    // Setup
    // Implementation might, or might not check if ID exist before issuing request to abort
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(false));
    EXPECT_CALL(m_mgr_mock, AbortDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .Times(Between(0, 1))
        .WillOnce(Return(ByMove(
            boost::make_exceptional_future<daq::Status>(std::invalid_argument("no such id")))));
 
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv)).WillOnce(Throw(std::invalid_argument("no such id")));
 
    // Run
    auto reply_fut = m_daq_impl->AbortDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, AbortDaqShouldSucceedIfMgrOpSuceeds) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);  // Note: Raw pointer since MAL use unsafe APIs
 
    auto reply_status = daq::Status("id", "fileid");
    reply_status.state = daq::State::Aborted;
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, AbortDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::Status>(reply_status))));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->AbortDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqReply> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, ForceAbortDaqShouldUseTolerantPolicy) {
    // Setup
    auto* fake_reply(new daqif::DaqReplyFake);  // Note: Raw pointer since MAL use unsafe APIs
    auto reply_status = daq::Status("id", "fileid");
    reply_status.state = daq::State::Aborted;
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, AbortDaqAsync("id"sv, daq::ErrorPolicy::Tolerant))
        .WillOnce(Return(ByMove(boost::make_ready_future<daq::Status>(reply_status))));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->ForceAbortDaq("id");
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqReply> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
}
 
TEST_F(TestOcmDaqService, UpdateKeywordsShouldSucceedIfMgrOpSucceeds) {
    // Setup
    std::string keywords = R"(
        [
           {
              "type":"valueKeyword",
              "name":"OBJECT",
              "value":"OBJECT,SKY"
           },
           {
              "type":"esoKeyword",
              "name":"OBS TPLNO",
              "value":2
           }
        ]
    )";
    // This is what `keywords` should contain when parsed.
    daq::fits::KeywordVector parsed_keywords = {
        daq::fits::ValueKeyword("OBJECT", "OBJECT,SKY"),
        daq::fits::EsoKeyword("OBS TPLNO", static_cast<uint64_t>(2))};
    auto* fake_reply(new daqif::DaqReplyFake);  // Note: Raw pointer since MAL use unsafe APIs
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, UpdateKeywords("id"sv, parsed_keywords));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqReply"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->UpdateKeywords("id", keywords);
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqReply> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
    EXPECT_EQ(reply->getError(), false);
}
 
TEST_F(TestOcmDaqService, UpdateKeywordsFailsIfJsonParseFails) {
    // Setup
    std::string keywords_with_trailing_comma = R"(
        [
           {
              "type":"valueKeyword",
              "name":"OBJECT",
              "value":"OBJECT,SKY"
           },
           {
              "type":"esoKeyword",
              "name":"OBS TPLNO",
              "value":2
           },
        ]
    )";
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, UpdateKeywords("id"sv, _)).Times(0);
 
    // Run
    auto reply_fut = m_daq_impl->UpdateKeywords("id", keywords_with_trailing_comma);
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, UpdateKeywordsFailsIfSchemaParsingFails) {
    // Setup
    std::string keywords_with_unknown_type = R"(
        [
           {
              "type":"unknownKeywordHere",
              "name":"OBJECT",
              "value":"OBJECT,SKY"
           }
        ]
    )";
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, UpdateKeywords("id"sv, _)).Times(0);
 
    // Run
    auto reply_fut = m_daq_impl->UpdateKeywords("id", keywords_with_unknown_type);
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, GetStatusFailsIfDaqDoesNotExist) {
    // Setup
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(false));
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv))
        .Times(Between(0, 1))
        .WillOnce(Throw(std::invalid_argument("no such id")));
 
    // Run
    auto reply_fut = m_daq_impl->GetStatus("id");
    EXPECT_FALSE(reply_fut.is_ready())
        << "future cannot be ready since implementation should use provided executor to provide "
           "thread safety";
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqService, GetStatusSuccedsIfMgrOpSucceeds) {
    // Setup
    daq::Status status("id", "fileid", daq::State::Acquiring, false, daq::Status::TimePoint());
    auto* fake_reply(new daqif::DaqStatusFake);  // Note: Raw pointer since MAL use unsafe APIs
 
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv)).Times(Between(0, 1)).WillOnce(Return(status));
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqStatus"))
        .WillOnce(Return(fake_reply))
        .RetiresOnSaturation();
 
    // Run
    auto reply_fut = m_daq_impl->GetStatus("id");
    EXPECT_FALSE(reply_fut.is_ready())
        << "future cannot be ready since implementation should use provided executor to provide "
           "thread safety";
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    std::shared_ptr<daqif::DaqStatus> reply = reply_fut.get();
    EXPECT_EQ(reply->getId(), "id");
    EXPECT_EQ(reply->getFileId(), "fileid");
    EXPECT_EQ(reply->getError(), false);
    EXPECT_EQ(reply->getState(), daqif::StateAcquiring);
    EXPECT_EQ(reply->getSubState(), daqif::Acquiring);
}
 
TEST_F(TestOcmDaqService, GetActiveReturnsActive) {
    // Setup
    auto status1 = std::make_shared<daq::ObservableStatus>("completed", "fileid1");
    auto status2 = std::make_shared<daq::ObservableStatus>("active1", "fileid2");
    auto status3 = std::make_shared<daq::ObservableStatus>("active2", "fileid3");
    auto status4 = std::make_shared<daq::ObservableStatus>("completed", "fileid4");
    // Make DAQ completed
    status1->SetState(daq::State::Aborted);
    status4->SetState(daq::State::Aborted);
 
    auto daq1 = std::make_shared<DaqControllerFake>(status1);
    auto daq2 = std::make_shared<DaqControllerFake>(status2);
    auto daq3 = std::make_shared<DaqControllerFake>(status3);
    auto daq4 = std::make_shared<DaqControllerFake>(status4);
    std::vector<std::shared_ptr<daq::DaqController const>> daqs;
    daqs.push_back(daq1);
    daqs.push_back(daq2);
    daqs.push_back(daq3);
    daqs.push_back(daq4);
 
    auto* st1 = new daqif::DaqStatusFake;
    auto* st2 = new daqif::DaqStatusFake;
 
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_DaqStatus"))
        .WillOnce(Return(st1))
        .WillOnce(Return(st2))
        .RetiresOnSaturation();
 
    EXPECT_CALL(m_mgr_mock, GetDaqControllers()).WillOnce(Return(daqs));
 
    // Run
    auto reply_fut = m_daq_impl->GetActiveList();
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    auto reply = reply_fut.get();
    EXPECT_EQ(reply.size(), 2u);
    EXPECT_EQ(reply[0]->getId(), "active1");
    EXPECT_EQ(reply[1]->getId(), "active2");
}
 
TEST_F(TestOcmDaqService, AwaitDaqstateFailsWithInvalidArguments) {
    // Setup
    EXPECT_CALL(m_mgr_mock, GetStatus("id"sv)).WillRepeatedly(Return(daq::Status()));
 
    // Run
    {
        // Invalid timeout
        auto fut = m_daq_impl->AwaitDaqState("id", daqif::StateAcquiring, daqif::Acquiring, 0.0);
        m_io_ctx.poll();
        ASSERT_TRUE(fut.is_ready());
        EXPECT_THROW(fut.get(), daqif::DaqException);
    }
    m_io_ctx.reset();
    {
        // Invalid state combination
        auto fut = m_daq_impl->AwaitDaqState("id", daqif::StateMerging, daqif::Acquiring, 1.0);
        m_io_ctx.poll();
        ASSERT_TRUE(fut.is_ready());
        EXPECT_THROW(fut.get(), daqif::DaqException);
    }
}
 
TEST_F(TestOcmDaqService, AwaitDaqstateSucceds) {
    // Setup
    daq::Status status("id", "fileid");
    status.state = daq::State::Acquiring;
 
    boost::promise<daq::Result<daq::Status>> promise;
    EXPECT_CALL(m_mgr_mock, AwaitDaqStateAsync("id"sv, daq::State::Acquiring, 1000ms))
        .WillOnce(Return(ByMove(promise.get_future())));
 
    auto* reply = new daqif::AwaitDaqReplyFake;
 
    EXPECT_CALL(m_mal_mock, getDataEntityUnsafe(_, "daqif_AwaitDaqReply"))
        .WillOnce(Return(reply))
        .RetiresOnSaturation();
 
    // Run
    auto fut = m_daq_impl->AwaitDaqState("id", daqif::StateAcquiring, daqif::Acquiring, 1.0);
    ASSERT_FALSE(fut.is_ready());
    promise.set_value({false, status});
    m_io_ctx.poll();
    {
        auto val = fut.get();
        EXPECT_FALSE(val->getTimeout());
        auto status = val->getStatus();
        EXPECT_EQ(status->getId(), "id");
        EXPECT_EQ(status->getError(), false);
        EXPECT_EQ(status->getState(), daqif::StateAcquiring);
        EXPECT_EQ(status->getSubState(), daqif::Acquiring);
    }
}
 
TEST_F(TestOcmDaqServiceAbandoned, StartDaq) {
    // Setup
 
    // This is the promise used to create result from daq::Manager
    boost::promise<daq::State> mgr_promise;
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).WillOnce(Return(false));
    EXPECT_CALL(m_mgr_mock, MakeDaqId(_)).WillOnce(Return("fileid"));
    EXPECT_CALL(m_mgr_mock, StartDaqAsync(_)).WillOnce(Return(ByMove(mgr_promise.get_future())));
 
    // Run
    auto reply_fut = m_daq_impl->StartDaq("id", "prefix", m_prim_sources, m_metadata_sources, "");
    // Abandon the service
    m_daq_impl.reset();
 
    // Fulfull promise from daq::Manager:
    mgr_promise.set_value(daq::State::Acquiring);
 
    // Trigger handlers
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqServiceAbandoned, StopDaq) {
    // Setup
 
    // This is the promise used to create result from daq::Manager
    boost::promise<daq::Status> mgr_promise;
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, StopDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(mgr_promise.get_future())));
 
    // Run
    auto reply_fut = m_daq_impl->StopDaq("id");
    // Abandon the service
    m_daq_impl.reset();
 
    // Fulfull promise from daq::Manager:
    mgr_promise.set_value(
        daq::Status("id", "fileid", daq::State::Stopped, true, daq::Status::TimePoint()));
 
    // Trigger handlers
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}
 
TEST_F(TestOcmDaqServiceAbandoned, AbortDaq) {
    // Setup
 
    // This is the promise used to create result from daq::Manager
    boost::promise<daq::Status> mgr_promise;
 
    auto reply_status = daq::Status("id", "fileid");
    reply_status.state = daq::State::Aborted;
 
    // Implementation might, or might not check if ID exist before issuing request to stop
    EXPECT_CALL(m_mgr_mock, HaveDaq("id"sv, _)).Times(Between(0, 1)).WillRepeatedly(Return(true));
    EXPECT_CALL(m_mgr_mock, AbortDaqAsync("id"sv, daq::ErrorPolicy::Strict))
        .WillOnce(Return(ByMove(mgr_promise.get_future())));
 
    // Run
    auto reply_fut = m_daq_impl->AbortDaq("id");
    // Abandon the service
    m_daq_impl.reset();
 
    // Fulfull promise from daq::Manager:
    mgr_promise.set_value(reply_status);
 
    // Trigger handlers
    m_io_ctx.poll();
    ASSERT_TRUE(reply_fut.is_ready());
    EXPECT_THROW(reply_fut.get(), daqif::DaqException);
}