Add ECMWF reproducer test

This is for JSC issue #16

configure.ac

@@ -557,6 +557,7 @@ m4_define([TESTSCRIPTS],
    tests/check_pm_declare_group.sh \
    tests/check_pm_interlock.sh \
    tests/check_subscribe.sh \
+   tests/check_ecmwf_selector.sh \
   ])
 
 m4_foreach_w([TESTSCRIPT],

tests/.gitignore

@@ -22,3 +22,6 @@ simple_injector
 simple_group_injector
 simple_telemetry_listener
 coverage
+check_ecmwf_selector.sh
+ecmwf_consumer
+ecmwf_producer

tests/Makefile.am

@@ -76,7 +76,8 @@ TESTS = check_version check_init check_uuid \
 	check_pm_declare.sh \
 	check_pm_interlock.sh \
 	check_subscribe.sh \
-	check_pm_declare_group.sh
+	check_pm_declare_group.sh \
+	check_ecmwf_selector.sh
 
 
 XFAIL_TESTS = \
@@ -117,7 +118,9 @@ check_PROGRAMS = check_version check_init check_uuid \
 		 simple_injector \
 		 simple_archiver \
 		 simple_telemetry_listener \
-		 check_events
+		 check_events \
+		 ecmwf_producer \
+                 ecmwf_consumer
 
 
 if WITH_MIO
@@ -200,7 +203,7 @@ clean-local-check:
 all:
 
 if WITH_MIO
-check_PROGRAMS += mio-config-default.yaml mio-config-C1.yaml mio-config-C2.yaml mio-config-C1a.yaml mio-config-C2a.yaml mio-config-PM1.yaml mio-config-PM2.yaml mio-config-PM3.yaml mio-config-CINJ.yaml mio-config-CARCH.yaml
+check_PROGRAMS += mio-config-default.yaml mio-config-C1.yaml mio-config-C2.yaml mio-config-C1a.yaml mio-config-C2a.yaml mio-config-PM1.yaml mio-config-PM2.yaml mio-config-PM3.yaml mio-config-CINJ.yaml mio-config-CARCH.yaml mio-config-PMECMWF.yaml mio-config-PECMWF.yaml mio-config-CECMWF.yaml
 
 # avoid getting weird default values from automake for these
 mio_config_default_yaml_SOURCES =# empty
@@ -213,6 +216,9 @@ mio_config_C2_yaml_SOURCES =# empty
 mio_config_C2a_yaml_SOURCES =# empty
 mio_config_CINJ_yaml_SOURCES =# empty
 mio_config_CARCH_yaml_SOURCES =# empty
+mio_config_PMECMWF_yaml_SOURCE =# empty
+mio_config_PECMWF_yaml_SOURCE =# empty
+mio_config_CECMWF_yaml_SOURCE =# empty
 
 mio-config-default.yaml: generate-mio-config.sh
 	./generate-mio-config.sh 101 && mv mio-config.yaml $@
@@ -234,6 +240,12 @@ mio-config-CINJ.yaml: generate-mio-config.sh
 	./generate-mio-config.sh 109 && mv mio-config.yaml $@
 mio-config-CARCH.yaml: generate-mio-config.sh
 	./generate-mio-config.sh 110 && mv mio-config.yaml $@
+mio-config-PMECMWF.yaml: generate-mio-config.sh
+	./generate-mio-config.sh 111 && mv mio-config.yaml $@
+mio-config-PECMWF.yaml: generate-mio-config.sh
+	./generate-mio-config.sh 112 && mv mio-config.yaml $@
+mio-config-CECMWF.yaml: generate-mio-config.sh
+	./generate-mio-config.sh 113 && mv mio-config.yaml $@
 
 # inject mio config location into test:
 AM_TESTS_ENVIRONMENT=  MSTRO_MIO_CONFIG=mio-config-default.yaml; export MSTRO_MIO_CONFIG;

tests/check_ecmwf_selector.sh.in

+#!/usr/bin/env bash
+#
+# Run pool manager, a CDO producer task, and an consumer task
+# that tries to subscribe using a 'complicated' ECMWF selector
+#
+# Copyright (C) 2021 HPE Switzerland GmbH
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+# 1. Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#
+# 3. Neither the name of the copyright holder nor the names of its
+#    contributors may be used to endorse or promote products derived from
+#    this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+# PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+# aggressive error tracing
+set -euo pipefail
+
+# set a workflow name
+MSTRO_WORKFLOW_NAME="check_ecmwf_selector_wf_$$"
+export MSTRO_WORKFLOW_NAME
+
+# aggressive error tracing
+set -euo pipefail
+# ensure error and warnings get colors
+MSTRO_LOG_COLOR_ERRORS=1
+export MSTRO_LOG_COLOR_ERRORS
+
+# start pool manager as named pipe
+PM_CMD="@top_builddir@/tests/simple_pool_manager"
+
+PRODUCER_CMD="@top_builddir@/tests/ecmwf_producer"
+CONSUMER_CMD="@top_builddir@/tests/ecmwf_consumer"
+
+# start pool manager, connect its output to fd 3:
+# (we need to run in a subshell to start a new process group)
+exec 3< <(env MSTRO_LOG_LEVEL=${MSTRO_LOG_LEVEL:-3}  MSTRO_MIO_CONFIG="./mio-config-PM-ECMWF.yaml"  ${PM_CMD})
+PM_PID=$!
+
+terminate () {
+    #$1 is PM_PID, $2 is desired exit code Note that killing PM_PID is
+    # not enough, the actual PM will likely be a child of this one,
+    # and if we kill it it will be reparented to init(1).
+    echo "exit, stopping pool manager"
+    # close stdout of the pool manager; its heartbeat output on stdout
+    # will make it receive a SIGPIPE eventually, which initiates
+    # termination.
+    exec 3<&-
+    exit ${2:-99}	       
+}
+
+# ensure USR2 is not caught by this shell
+trap '' USR2
+
+# trap error exit of script: close pipe to pool manager and return error
+trap "terminate ${PM_PID} 99" err
+
+# read pm info: It comes as a CSV of MSTRO_POOL_INFO; base64 text;
+# Potentially we might (later) have the pool manager report changed
+# info over time which we could read with the same read command.
+# 
+read -d ';' -u 3 pm_info_varname
+read -d ';' -u 3 pm_info
+echo "PM info: $pm_info"
+
+if test x"${pm_info_varname}" != xMSTRO_POOL_MANAGER_INFO; then
+    exit 99;
+fi
+
+
+MSTRO_POOL_MANAGER_INFO="$pm_info"
+export MSTRO_POOL_MANAGER_INFO
+
+# start consumer
+(env MSTRO_COMPONENT_NAME="CONSUMER" MSTRO_TRANSPORT_DEFAULT="MIO" MSTRO_MIO_CONFIG="./mio-config-CECMWF.yaml"  MSTRO_LOG_COLOR_ERRORS=1 MSTRO_LOG_COLOR="CYAN" ${CONSUMER_CMD} -n) || exit 99  &
+
+# wait for consumer to register subscriptions
+sleep 3
+
+# start producer
+(env MSTRO_COMPONENT_NAME="PRODUCER" MSTRO_TRANSPORT_DEFAULT="MIO" MSTRO_MIO_CONFIG="./mio-config-PECMWF.yaml"  MSTRO_LOG_COLOR_ERRORS=1 MSTRO_LOG_COLOR="BLUE" ${PRODUCER_CMD} -n) || exit 99  &
+
+
+wait %1 || exit 99
+wait %2 || exit 99
+
+# trap normal script termination: close pipe to pool manager
+terminate ${PM_PID} 0
+

tests/check_uuid.c

@@ -115,7 +115,7 @@ CHEAT_TEST(uuid_import_export,
 
        cheat_assert(UUID_RC_OK==uuid_export(uuid1, UUID_FMT_STR, &data2, NULL));
 
-       msg = uuid_import(uuid2, UUID_FMT_STR, data2, &data_len);
+       msg = uuid_import(uuid2, UUID_FMT_STR, data2, data_len);
 
        printf("%s", uuid_error(msg));
        

tests/ecmwf_consumer.c

+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include "maestro.h"
+
+
+int main(int argc, char *argv[])
+{
+    mstro_cdo_selector selector = NULL;
+    mstro_subscription subscription = NULL;
+
+    assert(MSTRO_OK == mstro_init(getenv("MSTRO_WORKFLOW_NAME"), getenv("MSTRO_COMPONENT_NAME"), 0));
+    assert(MSTRO_OK == mstro_cdo_selector_create(NULL, NULL, "(.maestro.ecmwf.step = 2)", &selector));
+    assert(MSTRO_OK == mstro_subscribe(selector, MSTRO_POOL_EVENT_OFFER, false, &subscription));
+    assert(MSTRO_OK == mstro_cdo_selector_dispose(selector));
+
+    mstro_pool_event e = NULL;
+    assert(MSTRO_OK == mstro_subscription_wait(subscription, &e));
+    if (e != NULL) {
+        fprintf(stdout, "Offer event: %p, kind: %d, cdo_name: %s\n", e, e->kind, e->offer.cdo_name);
+        mstro_cdo cdo = NULL;
+        assert(MSTRO_OK == mstro_cdo_declare(e->offer.cdo_name, MSTRO_ATTR_DEFAULT, &cdo));
+        assert(MSTRO_OK == mstro_cdo_require(cdo));
+        assert(MSTRO_OK == mstro_cdo_demand(cdo));
+
+        enum mstro_cdo_attr_value_type ttype;
+        const int64_t* tval = NULL;
+        assert(MSTRO_OK == mstro_cdo_attribute_get(cdo, ".maestro.ecmwf.step", &ttype, (const void**)&tval));
+        assert(ttype == MSTRO_CDO_ATTR_VALUE_int64);
+        fprintf(stdout, "Step = %d\n", *tval);
+
+        assert(MSTRO_OK == mstro_cdo_dispose(cdo));
+    }
+
+    assert(MSTRO_OK == mstro_finalize());
+    return 0;
+}

tests/ecmwf_producer.c

+#include <stdio.h>
+#include <assert.h>
+#include <unistd.h>
+#include "maestro.h"
+
+
+int main(int argc, char *argv[])
+{
+    assert(MSTRO_OK == mstro_init(getenv("MSTRO_WORKFLOW_NAME"), getenv("MSTRO_COMPONENT_NAME"), 0));
+    mstro_cdo cdo = NULL;
+    assert(MSTRO_OK == mstro_cdo_declare("my-cdo-1", MSTRO_ATTR_DEFAULT, &cdo));
+    int64_t step = 2;
+    assert(MSTRO_OK == mstro_cdo_attribute_set(cdo, ".maestro.ecmwf.step", (void **)&step, true));
+    assert(MSTRO_OK == mstro_cdo_seal(cdo));
+    assert(MSTRO_OK == mstro_cdo_offer(cdo));
+    sleep(5); // sleep until we fix acknowledge issue
+    assert(MSTRO_OK == mstro_cdo_withdraw(cdo));
+    assert(MSTRO_OK == mstro_cdo_dispose(cdo));
+    assert(MSTRO_OK == mstro_finalize());
+    return 0;
+}