Add endpoint_config_connection_error test

bin/loop_sequences

@@ -21,7 +21,13 @@ fi
 
 export UDMI_VERSION=`git describe --always --dirty`
 
+if [[ ! -f $VALIDATOR_CONFIG ]]; then
+  echo Missing $VALIDATOR_CONFIG
+  false
+fi
+
 echo Parsing $VALIDATOR_CONFIG:
+
 cat $VALIDATOR_CONFIG
 project_id=`jq -r .project_id $VALIDATOR_CONFIG`
 site_model=`jq -r .site_model $VALIDATOR_CONFIG`

bin/pubber

@@ -26,26 +26,23 @@ $ROOT_DIR/pubber/bin/build
 
 echo Running tools version `(cd $ROOT_DIR; git describe)`
 
-if (($# == 0)); then  
-    $ROOT_DIR/pubber/bin/run $project_id $site_path $device_id $serial_no
-else
-    declare -A options
-    for option in $*; do
-        if [[ $option == *"="* ]]; then
-            k=$(echo $option | cut -d'=' -f1)
-            v="\"$(echo $option | cut -d'=' -f2)\""
-        else
-            k=$option
-            v=true
-        fi
-        printf -v options_json '%s"%s":%s,' "$options_json" "$k" "$v"
-    done
-    options_json="{${options_json%,}}" 
-
-    registry_id=`jq -r .registry_id $site_path/cloud_iot_config.json`
-    cloud_region=`jq -r .cloud_region $site_path/cloud_iot_config.json`
-
-    cat <<EOF > /tmp/pubber_config.json
+declare -A options
+for option in $*; do
+    if [[ $option == *"="* ]]; then
+        k=$(echo $option | cut -d'=' -f1)
+        v="\"$(echo $option | cut -d'=' -f2)\""
+    else
+        k=$option
+        v=true
+    fi
+    printf -v options_json '%s"%s":%s,' "$options_json" "$k" "$v"
+done
+options_json="{${options_json%,}}"
+
+registry_id=`jq -r .registry_id $site_path/cloud_iot_config.json`
+cloud_region=`jq -r .cloud_region $site_path/cloud_iot_config.json`
+
+cat <<EOF > /tmp/pubber_config.json
 {
     "endpoint": {
       "protocol": "mqtt",
@@ -58,5 +55,5 @@ else
     "options": $options_json
 }
 EOF
-    $ROOT_DIR/pubber/bin/run /tmp/pubber_config.json
-fi
+
+$ROOT_DIR/pubber/bin/run /tmp/pubber_config.json

docs/specs/sequences/generated.md

@@ -29,6 +29,7 @@ Some caveats:
 <!-- START GENERATED, do not edit anything after this line! -->
 * [broken_config](#broken_config): Check that the device correctly handles a broken (non-json) config message.
 * [device_config_acked](#device_config_acked): Check that the device MQTT-acknowledges a sent config.
+* [endpoint_config_connection_error](#endpoint_config_connection_error): Push endpoint config message to device that results in a connection error.
 * [extra_config](#extra_config): Check that the device correctly handles an extra out-of-schema field
 * [periodic_scan](#periodic_scan)
 * [self_enumeration](#self_enumeration)
@@ -65,6 +66,15 @@ Check that the device MQTT-acknowledges a sent config.
 
 1. Wait for config acked
 
+## endpoint_config_connection_error
+
+Push endpoint config message to device that results in a connection error.
+
+1. Update config:
+    * Add `blobset` = { "blobs": {  } }
+1. Wait for device tried endpoint config which resulted in connection error
+1. Test failed: timeout waiting for device tried endpoint config which resulted in connection error
+
 ## extra_config
 
 Check that the device correctly handles an extra out-of-schema field

pubber/src/main/java/daq/pubber/Pubber.java

@@ -165,6 +165,7 @@ public class Pubber {
   private EndpointConfiguration extractedEndpoint;
   private SiteModel siteModel;
   private PrintStream logPrintWriter;
+  private Entry endpointStatus;
 
   /**
    * Start an instance from a configuration file.
@@ -829,16 +830,13 @@ private void processConfigUpdate(Config config) {
 
   private void extractEndpointBlobConfig() {
     if (deviceConfig.blobset == null) {
-      deviceState.blobset = null;
+      extractedEndpoint = null;
       return;
     }
     try {
       String iotConfig = extractConfigBlob(IOT_ENDPOINT_CONFIG.value());
-      if (iotConfig == null) {
-        removeBlobsetBlobState(IOT_ENDPOINT_CONFIG);
-        return;
-      }
-      extractedEndpoint = OBJECT_MAPPER.readValue(iotConfig, EndpointConfiguration.class);
+      extractedEndpoint = iotConfig == null ? null
+          : OBJECT_MAPPER.readValue(iotConfig, EndpointConfiguration.class);
     } catch (Exception e) {
       throw new RuntimeException("While extracting endpoint blob config", e);
     }
@@ -849,6 +847,10 @@ private void removeBlobsetBlobState(SystemBlobsets blobId) {
       return;
     }
     deviceState.blobset.blobs.remove(blobId.value());
+    if (deviceState.blobset.blobs.isEmpty()) {
+      deviceState.blobset = null;
+    }
+    markStateDirty(0);
   }
 
   private void maybeRedirectEndpoint() {
@@ -857,25 +859,33 @@ private void maybeRedirectEndpoint() {
     String extractedSignature =
         redirectRegistry == null ? toJson(extractedEndpoint) : redirectedEndpoint(redirectRegistry);
 
-    if (extractedSignature == null || extractedSignature.equals(
-        currentSignature) || extractedSignature.equals(attemptedEndpoint)) {
+    if (extractedSignature == null) {
+      attemptedEndpoint = null;
+      removeBlobsetBlobState(IOT_ENDPOINT_CONFIG);
+      return;
+    }
+
+    BlobBlobsetState endpointState = ensureBlobsetState(IOT_ENDPOINT_CONFIG);
+
+    if (extractedSignature.equals(currentSignature) 
+        || extractedSignature.equals(attemptedEndpoint)) {
       return; // No need to redirect anything!
     }
 
     info("New config blob endpoint detected");
-    BlobBlobsetState endpointState = ensureBlobsetState(IOT_ENDPOINT_CONFIG);
 
     try {
+      attemptedEndpoint = extractedSignature;
       endpointState.phase = BlobPhase.APPLY;
       endpointState.status = null;
       publishSynchronousState();
-      attemptedEndpoint = extractedSignature;
       resetConnection(extractedSignature);
       endpointState.phase = BlobPhase.FINAL;
       appliedEndpoint = null;
     } catch (Exception e) {
       try {
         error("Reconfigure failed, attempting connection to last working endpoint", e);
+        endpointState.phase = BlobPhase.FINAL;
         endpointState.status = exceptionStatus(e, Category.BLOBSET_BLOB_APPLY);
         resetConnection(workingEndpoint);
         publishAsynchronousState();

validator/src/main/java/com/google/daq/mqtt/sequencer/sequences/BlobsetSequences.java

@@ -0,0 +1,48 @@
+package com.google.daq.mqtt.sequencer.sequences;
+
+import static udmi.schema.Category.BLOBSET_BLOB_APPLY;
+
+import com.google.daq.mqtt.sequencer.SequenceRunner;
+import java.util.Base64;
+import java.util.HashMap;
+import org.junit.Test;
+import udmi.schema.BlobBlobsetConfig;
+import udmi.schema.BlobBlobsetConfig.BlobPhase;
+import udmi.schema.BlobsetConfig;
+import udmi.schema.BlobsetConfig.SystemBlobsets;
+import udmi.schema.Entry;
+import udmi.schema.Level;
+
+/**
+ * Validation tests for instances that involve blobset config messages.
+ */
+
+public class BlobsetSequences extends SequenceRunner {
+
+  private static final String ENDPOINT_CONFIG_CONNECTION_ERROR_PAYLOAD =
+      "{ "
+      + "  \"protocol\": \"mqtt\",\n"
+      + "  \"client_id\": \"test_project/device\",\n"
+      + "  \"hostname\": \"localhost\"\n"
+      + "}";
+
+  @Test
+  @Description("Push endpoint config message to device that results in a connection error.")
+  public void endpoint_config_connection_error() {
+    BlobBlobsetConfig config = new BlobBlobsetConfig();
+    config.phase = BlobPhase.FINAL;
+    config.base64 = String.valueOf(
+        Base64.getEncoder().encode(ENDPOINT_CONFIG_CONNECTION_ERROR_PAYLOAD.getBytes()));
+    config.content_type = "application/json";
+    deviceConfig.blobset = new BlobsetConfig();
+    deviceConfig.blobset.blobs = new HashMap<String, BlobBlobsetConfig>();
+    deviceConfig.blobset.blobs.put(SystemBlobsets.IOT_ENDPOINT_CONFIG.value(), config);
+
+    untilTrue("device tried endpoint config which resulted in connection error", () -> {
+      Entry stateStatus = deviceState.blobset.blobs.get(SystemBlobsets.IOT_ENDPOINT_CONFIG).status;
+      return stateStatus.category.equals(BLOBSET_BLOB_APPLY)
+          && stateStatus.level == Level.ERROR.value();
+    });
+  }
+
+}