add python functions, add tests

common/cpp/include/asapo/common/networking.h

@@ -30,6 +30,7 @@ enum Opcode : uint8_t {
     kOpcodeAuthorize,
     kOpcodeTransferMetaData,
     kOpcodeDeleteStream,
+    kOpcodeGetMeta,
     kOpcodeCount,
 };
 

common/cpp/src/database/mongodb_client.cpp

@@ -203,6 +203,17 @@ Error MongoDBClient::InsertBsonDocument(const bson_p& document, bool ignore_dupl
     return nullptr;
 }
 
+bool documentWasChanged(bson_t* reply) {
+    bson_iter_t iter;
+    bson_iter_init_find(&iter, reply, "upsertedCount");
+    auto n_upsert = bson_iter_int32(&iter);
+    bson_iter_init_find(&iter, reply, "modifiedCount");
+    auto n_mod = bson_iter_int32(&iter);
+    bson_iter_init_find(&iter, reply, "matchedCount");
+    auto n_matched = bson_iter_int32(&iter);
+    return n_mod + n_upsert + n_matched > 0;
+}
+
 Error MongoDBClient::ReplaceBsonDocument(const std::string& id, const bson_p& document, bool upsert) const {
     bson_error_t mongo_err;
 
@@ -210,20 +221,13 @@ Error MongoDBClient::ReplaceBsonDocument(const std::string& id, const bson_p& do
     bson_t* selector = BCON_NEW ("_id", BCON_UTF8(id.c_str()));
     bson_t reply;
     Error err = nullptr;
-    bson_iter_t iter;
 
     if (!mongoc_collection_replace_one(current_collection_, selector, document.get(), opts, &reply, &mongo_err)) {
         err = DBErrorTemplates::kInsertError.Generate(mongo_err.message);
     }
 
-    if (err == nullptr) {
-        bson_iter_init_find(&iter, &reply, "upsertedCount");
-        auto n_upsert = bson_iter_int32(&iter);
-        bson_iter_init_find(&iter, &reply, "modifiedCount");
-        auto n_mod = bson_iter_int32(&iter);
-        if (n_mod + n_upsert != 1) {
-            err = DBErrorTemplates::kInsertError.Generate("metadata does not exist");
-        }
+    if (err == nullptr && !documentWasChanged(&reply)) {
+        err = DBErrorTemplates::kWrongInput.Generate("cannot replace: metadata does not exist");
     }
 
     bson_free(opts);
@@ -242,20 +246,12 @@ Error MongoDBClient::UpdateBsonDocument(const std::string& id, const bson_p& doc
 
     bson_t reply;
     Error err = nullptr;
-    bson_iter_t iter;
-
     if (!mongoc_collection_update_one(current_collection_, selector, update, opts, &reply, &mongo_err)) {
         err = DBErrorTemplates::kInsertError.Generate(mongo_err.message);
     }
 
-    if (err == nullptr) {
-        bson_iter_init_find(&iter, &reply, "upsertedCount");
-        auto n_upsert = bson_iter_int32(&iter);
-        bson_iter_init_find(&iter, &reply, "modifiedCount");
-        auto n_mod = bson_iter_int32(&iter);
-        if (n_mod + n_upsert != 1) {
-            err = DBErrorTemplates::kInsertError.Generate("metadata does not exist");
-        }
+    if (err == nullptr && !documentWasChanged(&reply)) {
+        err = DBErrorTemplates::kWrongInput.Generate("cannot update: metadata does not exist");
     }
 
     bson_free(opts);

producer/api/cpp/src/producer_impl.cpp

@@ -507,7 +507,13 @@ std::string ProducerImpl::GetBeamtimeMeta(uint64_t timeout_ms, Error* err) const
 }
 
 std::string ProducerImpl::GetMeta(const std::string& stream, uint64_t timeout_ms, Error* err) const {
-    return std::string();
+    auto header =  GenericRequestHeader{kOpcodeGetMeta, 0, 0, 0, "", stream};
+    auto response = BlockingRequest(std::move(header), timeout_ms, err);
+    if (*err) {
+        return "";
+    }
+    *err = nullptr;
+    return response;
 }
 
 }
\ No newline at end of file

producer/api/cpp/unittests/test_producer_impl.cpp

@@ -501,7 +501,7 @@ TEST_F(ProducerImplTests, WaitRequestsFinished) {
     ASSERT_THAT(err, Eq(asapo::ProducerErrorTemplates::kTimeout));
 }
 
-MATCHER_P3(M_CheckGetStreamInfoRequest, op_code, source_credentials, stream,
+MATCHER_P3(M_CheckGetRequest, op_code, source_credentials, stream,
            "Checks if a valid GenericRequestHeader was Send") {
     auto request = static_cast<ProducerRequest*>(arg);
     return ((asapo::GenericRequestHeader) (arg->header)).op_code == op_code
@@ -509,9 +509,9 @@ MATCHER_P3(M_CheckGetStreamInfoRequest, op_code, source_credentials, stream,
            && strcmp(((asapo::GenericRequestHeader) (arg->header)).stream, stream) == 0;
 }
 
-TEST_F(ProducerImplTests, GetStreamInfoMakesCorerctRequest) {
+TEST_F(ProducerImplTests, GetStreamInfoMakesCorrectRequest) {
     producer.SetCredentials(expected_credentials);
-    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetStreamInfoRequest(asapo::kOpcodeStreamInfo,
+    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetRequest(asapo::kOpcodeStreamInfo,
                                         expected_credentials_str,
                                         expected_stream), true)).WillOnce(
                                             Return(nullptr));
@@ -537,9 +537,9 @@ TEST(GetStreamInfoTest, GetStreamInfoTimeout) {
     ASSERT_THAT(err->Explain(), HasSubstr("opcode: 4"));
 }
 
-TEST_F(ProducerImplTests, GetLastStreamMakesCorerctRequest) {
+TEST_F(ProducerImplTests, GetLastStreamMakesCorrectRequest) {
     producer.SetCredentials(expected_credentials);
-    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetStreamInfoRequest(asapo::kOpcodeLastStream,
+    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetRequest(asapo::kOpcodeLastStream,
                                         expected_credentials_str,
                                         ""), true)).WillOnce(
                                             Return(nullptr));
@@ -605,7 +605,7 @@ MATCHER_P4(M_CheckDeleteStreamRequest, op_code, source_credentials, stream, flag
            && strcmp(((asapo::GenericRequestHeader) (arg->header)).stream, stream) == 0;
 }
 
-TEST_F(ProducerImplTests, DeleteStreamMakesCorerctRequest) {
+TEST_F(ProducerImplTests, DeleteStreamMakesCorrectRequest) {
     producer.SetCredentials(expected_credentials);
     asapo::DeleteStreamOptions expected_options{};
     expected_options.delete_meta = true;
@@ -622,4 +622,29 @@ TEST_F(ProducerImplTests, DeleteStreamMakesCorerctRequest) {
     ASSERT_THAT(err, Eq(asapo::ProducerErrorTemplates::kTimeout));
 }
 
+TEST_F(ProducerImplTests, GetStreamMetaMakesCorrectRequest) {
+    producer.SetCredentials(expected_credentials);
+    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetRequest(asapo::kOpcodeGetMeta,
+                                        expected_credentials_str,
+                                        expected_stream), true)).WillOnce(
+                                            Return(nullptr));
+
+    asapo::Error err;
+    producer.GetStreamMeta(expected_stream, 1000, &err);
+    ASSERT_THAT(err, Eq(asapo::ProducerErrorTemplates::kTimeout));
+}
+
+
+TEST_F(ProducerImplTests, GetBeamtimeMetaMakesCorrectRequest) {
+    producer.SetCredentials(expected_credentials);
+    EXPECT_CALL(mock_pull, AddRequest_t(M_CheckGetRequest(asapo::kOpcodeGetMeta,
+                                        expected_credentials_str,
+                                        ""), true)).WillOnce(
+                                            Return(nullptr));
+
+    asapo::Error err;
+    producer.GetBeamtimeMeta(1000, &err);
+    ASSERT_THAT(err, Eq(asapo::ProducerErrorTemplates::kTimeout));
+}
+
 }

producer/api/python/asapo_producer.pxd

@@ -121,7 +121,8 @@ cdef extern from "asapo/asapo_producer.h" namespace "asapo" nogil:
         Error DeleteStream(string stream, uint64_t timeout_ms, DeleteStreamOptions options)
         Error SendBeamtimeMetadata(string metadata, MetaIngestMode mode, RequestCallback callback)
         Error SendStreamMetadata(string metadata, MetaIngestMode mode, string stream, RequestCallback callback)
-
+        string GetStreamMeta(string stream, uint64_t timeout_ms, Error* err)
+        string GetBeamtimeMeta(uint64_t timeout_ms, Error* err)
 
 cdef extern from "asapo/asapo_producer.h" namespace "asapo":
     uint64_t kDefaultIngestMode

producer/api/python/asapo_producer.pyx.in

@@ -300,6 +300,41 @@ cdef class PyProducer:
             err = self.c_producer.get().DeleteStream(b_stream,timeout_ms,opts)
         if err:
             throw_exception(err)
+    def get_stream_meta(self,stream = 'default', uint64_t timeout_ms = 1000):
+        """
+         :param stream: stream name
+         :type stream: string
+         :param timeout_ms: timeout in milliseconds
+         :type timeout_ms: int
+         :raises:
+            AsapoWrongInputError: wrong input (authorization, ...)
+            AsapoTimeoutError: request not finished for a given timeout
+            AsapoProducerError: other errors
+        """
+        cdef Error err
+        cdef string res
+        cdef string b_stream = _bytes(stream)
+        with nogil:
+            res = self.c_producer.get().GetStreamMeta(b_stream,timeout_ms,&err)
+        if err:
+            throw_exception(err)
+        return json.loads(_str(res) or 'null')
+    def get_beamtime_meta(self, uint64_t timeout_ms = 1000):
+        """
+         :param timeout_ms: timeout in milliseconds
+         :type timeout_ms: int
+         :raises:
+            AsapoWrongInputError: wrong input (authorization, ...)
+            AsapoTimeoutError: request not finished for a given timeout
+            AsapoProducerError: other errors
+        """
+        cdef Error err
+        cdef string res
+        with nogil:
+            res = self.c_producer.get().GetBeamtimeMeta(timeout_ms,&err)
+        if err:
+            throw_exception(err)
+        return json.loads(_str(res) or 'null')
     def stream_info(self, stream = 'default', uint64_t timeout_ms = 1000):
         """
          :param stream: stream name

receiver/CMakeLists.txt

@@ -22,6 +22,7 @@ set(RECEIVER_CORE_FILES
         src/request_handler/request_handler_receive_metadata.cpp
         src/request_handler/request_handler_db_check_request.cpp
         src/request_handler/request_handler_db_delete_stream.cpp
+        src/request_handler/request_handler_db_get_meta.cpp
         src/request_handler/request_factory.cpp
         src/request_handler/request_handler_db.cpp
         src/file_processors/write_file_processor.cpp
@@ -99,6 +100,7 @@ set(TEST_SOURCE_FILES
         unittests/request_handler/test_request_handler_receive_data.cpp
         unittests/request_handler/test_request_handler_receive_metadata.cpp
         unittests/request_handler/test_request_handler_delete_stream.cpp
+        unittests/request_handler/test_request_handler_db_get_meta.cpp
         unittests/statistics/test_statistics_sender_influx_db.cpp
         unittests/statistics/test_statistics_sender_fluentd.cpp
         unittests/mock_receiver_config.cpp

receiver/src/request_handler/request_factory.cpp

@@ -91,6 +91,10 @@ Error RequestFactory::AddHandlersToRequest(std::unique_ptr<Request>& request,
         request->AddHandler(&request_handler_db_last_stream_);
         break;
     }
+    case Opcode::kOpcodeGetMeta: {
+        request->AddHandler(&request_handler_db_get_meta_);
+        break;
+    }
     default:
         return ReceiverErrorTemplates::kInvalidOpCode.Generate();
     }

receiver/src/request_handler/request_factory.h

@@ -7,6 +7,7 @@
 #include "request_handler_db_stream_info.h"
 #include "request_handler_db_last_stream.h"
 #include "request_handler_db_delete_stream.h"
+#include "request_handler_db_get_meta.h"
 
 namespace asapo {
 
@@ -29,6 +30,7 @@ class RequestFactory {
     RequestHandlerDbDeleteStream request_handler_delete_stream_{kDBDataCollectionNamePrefix};
     RequestHandlerDbLastStream request_handler_db_last_stream_{kDBDataCollectionNamePrefix};
     RequestHandlerDbMetaWrite request_handler_db_meta_write_{kDBMetaCollectionName};
+    RequestHandlerDbGetMeta request_handler_db_get_meta_{kDBMetaCollectionName};
     RequestHandlerAuthorize request_handler_authorize_;
     RequestHandlerDbCheckRequest request_handler_db_check_{kDBDataCollectionNamePrefix};;
     SharedCache cache_;

receiver/src/request_handler/request_handler_db_get_meta.cpp

+#include "request_handler_db_get_meta.h"
+#include "../receiver_config.h"
+#include <asapo/database/db_error.h>
+
+namespace asapo {
+
+RequestHandlerDbGetMeta::RequestHandlerDbGetMeta(std::string collection_name_prefix) : RequestHandlerDb(
+        std::move(collection_name_prefix)) {
+}
+
+Error RequestHandlerDbGetMeta::ProcessRequest(Request* request) const {
+    if (auto err = RequestHandlerDb::ProcessRequest(request) ) {
+        return err;
+    }
+
+    auto stream_name = request->GetStream();
+
+    std::string metaid = stream_name.empty() ? "bt" : "st_" + stream_name;
+    std::string meta;
+    auto err =  db_client__->GetMetaFromDb(kDBMetaCollectionName, metaid, &meta);
+
+    bool no_error = err == nullptr;
+    if (err == DBErrorTemplates::kNoRecord) {
+        no_error = true;
+    }
+
+    if (no_error) {
+        log__->Debug(std::string{"get meta for "} + (stream_name.empty() ? "beamtime" : stream_name) + " in " +
+                     db_name_ + " at " + GetReceiverConfig()->database_uri);
+        request->SetResponseMessage(meta, ResponseMessageType::kInfo);
+        return nullptr;
+    }
+
+    return DBErrorToReceiverError(err);
+}
+
+
+}
\ No newline at end of file

receiver/src/request_handler/request_handler_db_get_meta.h

+#include "request_handler_db.h"
+#include "../request.h"
+
+#ifndef ASAPO_RECEIVER_SRC_REQUEST_HANDLER_REQUEST_HANDLER_DB_GET_META_H_
+#define ASAPO_RECEIVER_SRC_REQUEST_HANDLER_REQUEST_HANDLER_DB_GET_META_H_
+
+namespace asapo {
+
+class RequestHandlerDbGetMeta final: public RequestHandlerDb {
+  public:
+    RequestHandlerDbGetMeta(std::string collection_name_prefix);
+    Error ProcessRequest(Request* request) const override;
+};
+
+}
+
+
+#endif //ASAPO_RECEIVER_SRC_REQUEST_HANDLER_REQUEST_HANDLER_DB_GET_META_H_

receiver/unittests/request_handler/test_request_factory.cpp

@@ -16,6 +16,7 @@
 #include "../../src/request_handler/request_handler_db_stream_info.h"
 #include "../../src/request_handler/request_handler_db_last_stream.h"
 #include "../../src/request_handler/request_handler_db_delete_stream.h"
+#include "../../src/request_handler/request_handler_db_get_meta.h"
 
 #include "../../src/request_handler/request_handler_receive_data.h"
 #include "../../src/request_handler/request_handler_receive_metadata.h"
@@ -225,5 +226,16 @@ TEST_F(FactoryTests, DeleteStreamRequest) {
     ASSERT_THAT(dynamic_cast<const asapo::RequestHandlerDbDeleteStream*>(request->GetListHandlers()[1]), Ne(nullptr));
 }
 
+TEST_F(FactoryTests, GetMetamRequest) {
+    generic_request_header.op_code = asapo::Opcode::kOpcodeGetMeta;
+    auto request = factory.GenerateRequest(generic_request_header, 1, origin_uri, &err);
+    ASSERT_THAT(err, Eq(nullptr));
+    ASSERT_THAT(request->GetListHandlers().size(), Eq(2));
+    ASSERT_THAT(dynamic_cast<const asapo::RequestHandlerAuthorize*>(request->GetListHandlers()[0]), Ne(nullptr));
+    ASSERT_THAT(dynamic_cast<const asapo::RequestHandlerDbGetMeta*>(request->GetListHandlers()[1]), Ne(nullptr));
+}
+
+
+
 
 }

receiver/unittests/request_handler/test_request_handler_db_get_meta.cpp

+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+#include <asapo/database/db_error.h>
+
+#include "asapo/unittests/MockIO.h"
+#include "asapo/unittests/MockDatabase.h"
+#include "asapo/unittests/MockLogger.h"
+
+#include "../../src/receiver_error.h"
+#include "../../src/request.h"
+#include "../../src/request_handler/request_factory.h"
+#include "../../src/request_handler/request_handler.h"
+#include "../../src/request_handler/request_handler_db_get_meta.h"
+#include "../../../common/cpp/src/database/mongodb_client.h"
+
+#include "../mock_receiver_config.h"
+#include "asapo/common/data_structs.h"
+#include "asapo/common/networking.h"
+#include "../receiver_mocking.h"
+
+using asapo::MockRequest;
+using asapo::MessageMeta;
+using ::testing::Test;
+using ::testing::Return;
+using ::testing::ReturnRef;
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::SetArgReferee;
+using ::testing::Gt;
+using ::testing::Eq;
+using ::testing::Ne;
+using ::testing::Mock;
+using ::testing::NiceMock;
+using ::testing::InSequence;
+using ::testing::SetArgPointee;
+using ::testing::AllOf;
+using ::testing::HasSubstr;
+
+
+using ::asapo::Error;
+using ::asapo::ErrorInterface;
+using ::asapo::FileDescriptor;
+using ::asapo::SocketDescriptor;
+using ::asapo::MockIO;
+using asapo::Request;
+using asapo::RequestHandlerDbGetMeta;
+using ::asapo::GenericRequestHeader;
+
+using asapo::MockDatabase;
+using asapo::RequestFactory;
+using asapo::SetReceiverConfig;
+using asapo::ReceiverConfig;
+
+
+namespace {
+
+class DbMetaGetMetaTests : public Test {
+  public:
+    RequestHandlerDbGetMeta handler{asapo::kDBDataCollectionNamePrefix};
+    std::unique_ptr<NiceMock<MockRequest>> mock_request;
+    NiceMock<MockDatabase> mock_db;
+    NiceMock<asapo::MockLogger> mock_logger;
+    ReceiverConfig config;
+    std::string expected_beamtime_id = "beamtime_id";
+    std::string expected_data_source = "source";
+    std::string expected_stream = "stream";
+    std::string expected_meta = "meta";
+    void SetUp() override {
+        GenericRequestHeader request_header;
+        handler.db_client__ = std::unique_ptr<asapo::Database> {&mock_db};
+        handler.log__ = &mock_logger;
+        mock_request.reset(new NiceMock<MockRequest> {request_header, 1, "", nullptr});
+        ON_CALL(*mock_request, GetBeamtimeId()).WillByDefault(ReturnRef(expected_beamtime_id));
+    }
+    void TearDown() override {
+        handler.db_client__.release();
+    }
+    void ExpectGet(bool stream, const asapo::DBErrorTemplate* errorTemplate) {
+        SetReceiverConfig(config, "none");
+        EXPECT_CALL(*mock_request, GetDataSource()).WillOnce(ReturnRef(expected_data_source));
+        if (stream) {
+            EXPECT_CALL(*mock_request, GetStream()).WillOnce(Return(expected_stream));
+        }
+
+        EXPECT_CALL(mock_db, Connect_t(config.database_uri, expected_beamtime_id + "_" + expected_data_source)).
+        WillOnce(testing::Return(nullptr));
+        EXPECT_CALL(mock_db, GetMetaFromDb_t("meta", stream ? "st_" + expected_stream : "bt", _)).
+        WillOnce(DoAll(
+                     SetArgPointee<2>(expected_meta),
+                     testing::Return(errorTemplate == nullptr ? nullptr : errorTemplate->Generate().release())
+                 ));
+        if (errorTemplate == nullptr) {
+            EXPECT_CALL(*mock_request, SetResponseMessage(expected_meta, asapo::ResponseMessageType::kInfo));
+            EXPECT_CALL(mock_logger, Debug(AllOf(HasSubstr("meta"),
+                                                 HasSubstr(config.database_uri),
+                                                 HasSubstr(expected_data_source),
+                                                 HasSubstr(stream ? expected_stream : "beamtime"),
+                                                 HasSubstr(expected_beamtime_id)
+                                                )
+                                          )
+                       );
+        }
+
+    }
+};
+
+
+
+TEST_F(DbMetaGetMetaTests, GetBeamtimeMetaOk) {
+    ExpectGet(false, nullptr);
+    auto err = handler.ProcessRequest(mock_request.get());
+    ASSERT_THAT(err, Eq(nullptr));
+}
+
+
+TEST_F(DbMetaGetMetaTests, GetStreamMetaOk) {
+    ExpectGet(true, nullptr);
+    auto err = handler.ProcessRequest(mock_request.get());
+    ASSERT_THAT(err, Eq(nullptr));
+}
+
+TEST_F(DbMetaGetMetaTests, GetStreamMetaError) {
+    ExpectGet(true, &asapo::DBErrorTemplates::kDBError);
+    auto err = handler.ProcessRequest(mock_request.get());
+    ASSERT_THAT(err, Ne(nullptr));
+}
+
+}

tests/automatic/consumer/consumer_api/CMakeLists.txt

@@ -11,6 +11,5 @@ target_link_libraries(${TARGET_NAME} test_common asapo-consumer)
 ################################
 # Testing
 ################################
-add_script_test("${TARGET_NAME}" "$<TARGET_FILE:${TARGET_NAME}>"
-        )
+add_script_test("${TARGET_NAME}" "$<TARGET_FILE:${TARGET_NAME}>")
 

tests/automatic/mongo_db/meta/meta_mongodb.cpp

@@ -75,7 +75,7 @@ int main(int argc, char* argv[]) {
             db.InsertMeta("meta", stream_name, reinterpret_cast<const uint8_t*>(meta.c_str()), meta.size(), mode);
         M_AssertEq(nullptr, err);
         err = db.InsertMeta("meta", "notexist", reinterpret_cast<const uint8_t*>(meta.c_str()), meta.size(), mode);
-        M_AssertTrue(err == asapo::DBErrorTemplates::kInsertError);
+        M_AssertTrue(err == asapo::DBErrorTemplates::kWrongInput);
 
         std::string meta_res;
         err = db.GetMetaFromDb("meta", "0", &meta_res);
@@ -87,6 +87,9 @@ int main(int argc, char* argv[]) {
         mode.op = asapo::MetaIngestOp::kUpdate;
         err = db.InsertMeta("meta", stream_name, reinterpret_cast<const uint8_t*>(mod_meta.c_str()), mod_meta.size(), mode);
         M_AssertEq(nullptr, err);
+        err = db.InsertMeta("meta", stream_name, reinterpret_cast<const uint8_t*>(mod_meta.c_str()), mod_meta.size(), mode);
+        M_AssertEq(nullptr, err);
+
         err = db.GetMetaFromDb("meta", stream_name, &meta_res);
         M_AssertEq(nullptr, err);
         M_AssertEq(expected_meta, meta_res);

tests/automatic/producer/cpp_api/CMakeLists.txt

@@ -11,6 +11,5 @@ target_link_libraries(${TARGET_NAME} test_common asapo-producer)
 ################################
 # Testing
 ################################
-add_script_test("${TARGET_NAME}" "$<TARGET_FILE:${TARGET_NAME}>"
-        )
+add_script_test("${TARGET_NAME}" "$<TARGET_FILE:${TARGET_NAME}>")
 

tests/automatic/producer/cpp_api/check_linux.sh

@@ -23,4 +23,4 @@ mkdir -p ${receiver_folder}
 
 echo test > file1
 
-$1 "127.0.0.1:8400" $data_source $beamtime_id
+$@ "127.0.0.1:8400" $data_source $beamtime_id

tests/automatic/producer/cpp_api/producer_api.cpp

@@ -23,6 +23,21 @@ Args GetArgs(int argc, char* argv[]) {
     return Args{server, source, beamtime};
 }
 
+void TestMeta(const std::unique_ptr<asapo::Producer>& producer) {
+    asapo::Error err;
+    std::string meta = R"({"data":"test","embedded":{"edata":2}})";
+    producer->SendBeamtimeMetadata(meta, asapo::MetaIngestMode{asapo::MetaIngestOp::kInsert, true}, nullptr);
+    producer->WaitRequestsFinished(5000);
+    auto meta_received = producer->GetBeamtimeMeta(5000, &err);
+    M_AssertTrue(meta_received == meta);
+    std::string meta_update = R"({"embedded":{"edata":3}})";
+    std::string meta_updated = R"({"data":"test","embedded":{"edata":3}})";
+    producer->SendBeamtimeMetadata(meta_update, asapo::MetaIngestMode{asapo::MetaIngestOp::kUpdate, false}, nullptr);
+    producer->WaitRequestsFinished(5000);
+    meta_received = producer->GetBeamtimeMeta(5000, &err);
+    M_AssertTrue(meta_received == meta_updated);
+}
+
 
 void Test(const std::unique_ptr<asapo::Producer>& producer) {
     asapo::MessageMeta fi;
@@ -37,9 +52,11 @@ void Test(const std::unique_ptr<asapo::Producer>& producer) {
     M_AssertTrue(!server.empty(), "server version");
 
 
-    producer->GetStreamInfo("default",5000,&err);
+    TestMeta(producer);
+
+    producer->GetStreamInfo("default", 5000, &err);
     if (err) {
-        printf("%s\n",err->Explain().c_str());
+        printf("%s\n", err->Explain().c_str());
     }
     M_AssertTrue(err == nullptr, "stream info");
 
@@ -51,7 +68,7 @@ std::unique_ptr<asapo::Producer> CreateProducer(const Args& args) {
     auto producer = asapo::Producer::Create(args.server, 2,
                                             asapo::RequestHandlerType::kTcp,
                                             asapo::SourceCredentials{asapo::SourceType::kProcessed, args.beamtime,
-                                                                     "", args.source, ""}, 60000, &err);
+                                                    "", args.source, ""}, 60000, &err);
     if (err) {
         std::cerr << "Cannot start producer. ProducerError: " << err << std::endl;
         exit(EXIT_FAILURE);
@@ -66,7 +83,7 @@ std::unique_ptr<asapo::Producer> CreateProducer(const Args& args) {
 void TestAll(const Args& args) {
     asapo::Error err;
     auto producer = CreateProducer(args);
-    if (producer==nullptr) {
+    if (producer == nullptr) {
         std::cout << "Error CreateProducer: " << err << std::endl;
         exit(EXIT_FAILURE);
     }

tests/automatic/producer/python_api/check_linux.sh

@@ -27,14 +27,19 @@ echo test > file1
 
 $1 $3 $data_source $beamtime_id  "127.0.0.1:8400" &> out || cat out
 cat out
-echo count successfully send, expect 15
-cat out | grep "successfuly sent" | wc -l | tee /dev/stderr | grep 15
+echo count successfully send, expect 17
+cat out | grep "successfuly sent" | wc -l | tee /dev/stderr | grep 17
+echo count wrong input, expect 11
+cat out | grep "wrong input" | wc -l | tee /dev/stderr | grep 11
+
+echo count wrong json, expect 2
+cat out | grep "JSON parse error" | wc -l | tee /dev/stderr | grep 2
 echo count same id, expect 4
 cat out | grep "already have record with same id" | wc -l | tee /dev/stderr | grep 4
 echo count duplicates, expect 6
 cat out | grep "duplicate" | wc -l | tee /dev/stderr | grep 6
-echo count data in callback, expect 3
-cat out | grep "'data':" | wc -l  | tee /dev/stderr | grep 3
+echo count data in callback, expect 6
+cat out | grep "'data':" | wc -l  | tee /dev/stderr | grep 6
 echo check found local io error
 cat out | grep "local i/o error"
 cat out | grep "Finished successfully"

tests/automatic/producer/python_api/check_windows.bat

@@ -18,7 +18,7 @@ set PYTHONPATH=%2
 type out
 set NUM=0
 for /F %%N in ('find /C "successfuly sent" ^< "out"') do set NUM=%%N
-echo %NUM% | findstr 15 || goto error
+echo %NUM% | findstr 17 || goto error
 
 for /F %%N in ('find /C "} wrong input: Bad request: already have record with same id" ^< "out"') do set NUM=%%N
 echo %NUM% | findstr 2 || goto error