E2E: Add regression test for custom mqtt topics (Azure#4857)

This PR is adding an E2E test for custom mqtt topics using the existing ```genericMqttTester``` image we run in longhaul. 

This test will deploy edgehub with the broker enabled and two genericMqttTester modules. One genericMqttTester module will be in initiate mode and the other will be in relay mode. The initiate mode module will publish on an mqtt topic ```initiate/1``` and the relaying module will subscribe to this topic. When the relaying module receives a message, it will publish it back on ```relay/1```. The initiating module will be subscribed on this ```relay/1``` topic and will report to the TRC when it receives the message.

In addition to adding the test it:
1. Refactors shared logic into util abstractions for cleaner code. Specifically adding the broker/TRC to the deployment.
2. Introduces a configuration for defining the restartPolicy for test module deployed in the test. This is done so that the genericMqttTester module will no start back up after it sends its defined amount of messages.
3. Improves the genericMqttTester module to use the module name when reporting to the trc rather than some hardcoded string "genericMqttTester". This allows the dotnet test code we right here to be more understandable.
3. Updates the documentation to include new parameters required for this test

builds/e2e/templates/e2e-setup.yaml

@@ -165,6 +165,7 @@ steps:
       metricsValidatorImage = "$imagePrefix-metrics-validator:$imageTag";
       numberLoggerImage = "$imagePrefix-number-logger:$imageTag";
       edgeAgentBootstrapImage = "$imagePrefix-agent-bootstrap-e2e-$(os)-$(arch)";
+      genericMqttTesterImage = "$imagePrefix-generic-mqtt-tester:$imageTag"
       registries = @(
         @{
           address = '$(cr.address)';

test/Microsoft.Azure.Devices.Edge.Test.Common/config/EdgeConfigBuilder.cs

@@ -45,9 +45,9 @@ public IModuleConfigBuilder AddEdgeHub(string image = null, bool optimizeForPerf
             return builder;
         }
 
-        public IModuleConfigBuilder AddModule(string name, string image)
+        public IModuleConfigBuilder AddModule(string name, string image, bool shouldRestart = true)
         {
-            var builder = new ModuleConfigBuilder(name, image);
+            var builder = new ModuleConfigBuilder(name, image, shouldRestart);
             this.moduleBuilders.Add(builder.Name, builder);
             return builder;
         }

test/Microsoft.Azure.Devices.Edge.Test.Common/config/HubModuleConfigBuilder.cs

@@ -10,7 +10,7 @@ public class HubModuleConfigBuilder : ModuleConfigBuilder
         const string HubCreateOptions = "{\"HostConfig\":{\"PortBindings\":{\"8883/tcp\":[{\"HostPort\":\"8883\"}],\"443/tcp\":[{\"HostPort\":\"443\"}],\"5671/tcp\":[{\"HostPort\":\"5671\"}]}}}";
 
         public HubModuleConfigBuilder(Option<string> image, bool optimizeForPerformance)
-            : base(ModuleName.EdgeHub, image.GetOrElse(DefaultImage), Option.Some(HubCreateOptions))
+            : base(ModuleName.EdgeHub, image.GetOrElse(DefaultImage), Option.Some(HubCreateOptions), true)
         {
             this.WithDesiredProperties(
                 new Dictionary<string, object>

test/Microsoft.Azure.Devices.Edge.Test.Common/config/ModuleConfigBuilder.cs

@@ -5,18 +5,28 @@ namespace Microsoft.Azure.Devices.Edge.Test.Common.Config
 
     public class ModuleConfigBuilder : BaseModuleConfigBuilder
     {
-        public ModuleConfigBuilder(string name, string image)
-            : this(name, image, Option.None<string>())
+        public ModuleConfigBuilder(string name, string image, bool shouldRestart)
+            : this(name, image, Option.None<string>(), shouldRestart)
         {
         }
 
-        public ModuleConfigBuilder(string name, string image, Option<string> createOptions)
+        public ModuleConfigBuilder(string name, string image, Option<string> createOptions, bool shouldRestart)
             : base(name, image)
         {
+            string restartPolicy = string.Empty;
+            if (shouldRestart)
+            {
+                restartPolicy = "always";
+            }
+            else
+            {
+                restartPolicy = "never";
+            }
+
             this.WithDeployment(
                 new[]
                 {
-                    ("restartPolicy", "always"),
+                    ("restartPolicy", restartPolicy),
                     ("status", "running")
                 });
             createOptions.ForEach(s => this.WithSettings(new[] { ("createOptions", s) }));

test/Microsoft.Azure.Devices.Edge.Test/GenericMqtt.cs

@@ -0,0 +1,82 @@
+// Copyright (c) Microsoft. All rights reserved.
+namespace Microsoft.Azure.Devices.Edge.Test
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Azure.Devices.Edge.Test.Common;
+    using Microsoft.Azure.Devices.Edge.Test.Common.Config;
+    using Microsoft.Azure.Devices.Edge.Test.Helpers;
+    using Microsoft.Azure.Devices.Edge.Util.Test.Common.NUnit;
+    using NUnit.Framework;
+
+    [EndToEnd]
+    public class GenericMqtt : SasManualProvisioningFixture
+    {
+        const string NetworkControllerModuleName = "networkController";
+        const string GenericMqttInitiatorModuleName = "GenericMqttInitiator";
+        const string GenericMqttRelayerModuleName = "GenericMqttRelayer";
+        const string GenericMqttTesterMaxMessages = "5";
+        const string GenericMqttTesterTestStartDelay = "10s";
+        const int SecondsBeforeVerification = 45;
+
+        /// <summary>
+        /// Scenario:
+        /// - Deploy Edge Hub with the broker enabled and two genericMqttTester
+        ///   modules.
+        /// - One genericMqttTester module will be in initiate mode and the other
+        ///   will be in relay mode.
+        /// - The initiate mode module will publish on an mqtt topic initiate/1
+        ///   which the relaying module will be subscribed to.
+        /// - When the relaying module receives a message, it will publish it
+        ///   back on relay/1.
+        /// - The initiating module will be subscribed on this relay/1 topic
+        ///   and will report to the TRC when it receives the message.
+        /// </summary>
+        /// <returns><see cref="Task"/> representing the asynchronous unit test.</returns>
+        [Test]
+        public async Task GenericMqttTelemetry()
+        {
+            CancellationToken token = this.TestToken;
+            string networkControllerImage = Context.Current.NetworkControllerImage.Expect(() => new ArgumentException("networkControllerImage parameter is required for Generic Mqtt test"));
+            string trcImage = Context.Current.TestResultCoordinatorImage.Expect(() => new ArgumentException("testResultCoordinatorImage parameter is required for Generic Mqtt test"));
+            string genericMqttTesterImage = Context.Current.GenericMqttTesterImage.Expect(() => new ArgumentException("genericMqttTesterImage parameter is required for Generic Mqtt test"));
+            string trackingId = Guid.NewGuid().ToString();
+
+            Action<EdgeConfigBuilder> addMqttBrokerConfig = MqttBrokerUtil.BuildAddBrokerToDeployment(false);
+            Action<EdgeConfigBuilder> addNetworkControllerConfig = TestResultCoordinatorUtil.BuildAddNetworkControllerConfig(trackingId, networkControllerImage);
+            Action<EdgeConfigBuilder> addTestResultCoordinatorConfig = TestResultCoordinatorUtil.BuildAddTestResultCoordinatorConfig(trackingId, trcImage, GenericMqttInitiatorModuleName, GenericMqttInitiatorModuleName);
+            Action<EdgeConfigBuilder> addGenericMqttTesterConfig = this.BuildAddGenericMqttTesterConfig(trackingId, trcImage, genericMqttTesterImage);
+            Action<EdgeConfigBuilder> config = addMqttBrokerConfig + addNetworkControllerConfig + addTestResultCoordinatorConfig + addGenericMqttTesterConfig;
+            EdgeDeployment deployment = await this.runtime.DeployConfigurationAsync(config, token, Context.Current.NestedEdge);
+
+            await Task.Delay(TimeSpan.FromSeconds(SecondsBeforeVerification));
+
+            await TestResultCoordinatorUtil.ValidateResultsAsync();
+        }
+
+        private Action<EdgeConfigBuilder> BuildAddGenericMqttTesterConfig(string trackingId, string trcImage, string genericMqttTesterImage)
+        {
+            return new Action<EdgeConfigBuilder>(
+                builder =>
+                {
+                    builder.AddModule(GenericMqttInitiatorModuleName, genericMqttTesterImage, false)
+                        .WithEnvironment(new[]
+                        {
+                            ("TEST_SCENARIO", "Initiate"),
+                            ("TRACKING_ID", trackingId),
+                            ("TEST_START_DELAY", GenericMqttTesterTestStartDelay),
+                            ("MESSAGES_TO_SEND", GenericMqttTesterMaxMessages),
+                        });
+
+                    builder.AddModule(GenericMqttRelayerModuleName, genericMqttTesterImage, false)
+                        .WithEnvironment(new[]
+                        {
+                            ("TEST_SCENARIO", "Relay"),
+                            ("TRACKING_ID", trackingId),
+                            ("TEST_START_DELAY", GenericMqttTesterTestStartDelay),
+                        });
+                });
+        }
+    }
+}

test/Microsoft.Azure.Devices.Edge.Test/PlugAndPlay.cs

@@ -2,26 +2,17 @@
 namespace Microsoft.Azure.Devices.Edge.Test
 {
     using System;
-    using System.Collections.Generic;
-    using System.Globalization;
     using System.Net;
-    using System.Net.Http;
-    using System.Net.Http.Headers;
-    using System.Security.Cryptography;
-    using System.Text;
     using System.Threading;
     using System.Threading.Tasks;
     using Microsoft.Azure.Devices.Edge.Test.Common;
-    using Microsoft.Azure.Devices.Edge.Test.Common.Certs;
     using Microsoft.Azure.Devices.Edge.Test.Common.Config;
     using Microsoft.Azure.Devices.Edge.Test.Helpers;
     using Microsoft.Azure.Devices.Edge.Util;
     using Microsoft.Azure.Devices.Edge.Util.Test.Common.NUnit;
     using Microsoft.Azure.Devices.Shared;
     using Microsoft.VisualStudio.TestPlatform.ObjectModel;
-    using Newtonsoft.Json.Linq;
     using NUnit.Framework;
-    using Serilog;
 
     [EndToEnd]
     public class PlugAndPlay : SasManualProvisioningFixture
@@ -44,16 +35,14 @@ public async Task PlugAndPlayDeviceClient(Protocol protocol, bool brokerOn)
                 Assert.Ignore();
             }
 
-            EdgeDeployment deployment = await this.runtime.DeployConfigurationAsync(
-                builder =>
-                {
-                    if (brokerOn)
-                    {
-                        this.AddBrokerToDeployment(builder);
-                    }
+            Action<EdgeConfigBuilder> config = this.BuildAddEdgeHubConfig(protocol);
+            if (brokerOn)
+            {
+                config += MqttBrokerUtil.BuildAddBrokerToDeployment(true);
+            }
 
-                    builder.GetModule(ModuleName.EdgeHub).WithEnvironment(new[] { ("UpstreamProtocol", protocol.ToString()) });
-                },
+            EdgeDeployment deployment = await this.runtime.DeployConfigurationAsync(
+                config,
                 token,
                 Context.Current.NestedEdge);
 
@@ -100,24 +89,14 @@ public async Task PlugAndPlayModuleClient(Protocol protocol, bool brokerOn)
             }
 
             string loadGenImage = Context.Current.LoadGenImage.Expect(() => new ArgumentException("loadGenImage parameter is required for Priority Queues test"));
-            EdgeDeployment deployment = await this.runtime.DeployConfigurationAsync(
-                builder =>
-                {
-                    if (brokerOn)
-                    {
-                        this.AddBrokerToDeployment(builder);
-                    }
+            Action<EdgeConfigBuilder> config = this.BuildAddEdgeHubConfig(protocol) + this.BuildAddLoadGenConfig(protocol, loadGenImage);
+            if (brokerOn)
+            {
+                config += MqttBrokerUtil.BuildAddBrokerToDeployment(true);
+            }
 
-                    builder.GetModule(ModuleName.EdgeHub).WithEnvironment(new[] { ("UpstreamProtocol", protocol.ToString()) });
-                    builder.AddModule(LoadGenModuleName, loadGenImage)
-                    .WithEnvironment(new[]
-                    {
-                            ("testStartDelay", "00:00:00"),
-                            ("messageFrequency", "00:00:00.5"),
-                            ("transportType", protocol.ToString()),
-                            ("modelId", TestModelId)
-                    });
-                },
+            EdgeDeployment deployment = await this.runtime.DeployConfigurationAsync(
+                config,
                 token,
                 Context.Current.NestedEdge);
 
@@ -126,42 +105,36 @@ public async Task PlugAndPlayModuleClient(Protocol protocol, bool brokerOn)
             await this.ValidateIdentity(this.runtime.DeviceId, Option.Some(LoadGenModuleName), TestModelId, token);
         }
 
-        EdgeConfigBuilder AddBrokerToDeployment(EdgeConfigBuilder builder)
+        async Task ValidateIdentity(string deviceId, Option<string> moduleId, string expectedModelId, CancellationToken token)
         {
-            builder.GetModule(ModuleName.EdgeHub)
-                .WithEnvironment(new[]
-                {
-                    ("experimentalFeatures__enabled", "true"),
-                    ("experimentalFeatures__mqttBrokerEnabled", "true"),
-                })
-                .WithDesiredProperties(new Dictionary<string, object>
+            Twin twin = await this.IotHub.GetTwinAsync(deviceId, moduleId, token);
+            string actualModelId = twin.ModelId;
+            Assert.AreEqual(expectedModelId, actualModelId);
+        }
+
+        private Action<EdgeConfigBuilder> BuildAddLoadGenConfig(Protocol protocol, string loadGenImage)
+        {
+            return new Action<EdgeConfigBuilder>(
+                builder =>
                 {
-                    ["mqttBroker"] = new
+                    builder.AddModule(LoadGenModuleName, loadGenImage)
+                    .WithEnvironment(new[]
                     {
-                        authorizations = new[]
-                        {
-                            new
-                            {
-                                 identities = new[] { "{{iot:identity}}" },
-                                 allow = new[]
-                                 {
-                                     new
-                                     {
-                                         operations = new[] { "mqtt:connect" }
-                                     }
-                                 }
-                            }
-                        }
-                    }
+                            ("testStartDelay", "00:00:00"),
+                            ("messageFrequency", "00:00:00.5"),
+                            ("transportType", protocol.ToString()),
+                            ("modelId", TestModelId)
+                    });
                 });
-            return builder;
         }
 
-        async Task ValidateIdentity(string deviceId, Option<string> moduleId, string expectedModelId, CancellationToken token)
+        private Action<EdgeConfigBuilder> BuildAddEdgeHubConfig(Protocol protocol)
         {
-            Twin twin = await this.IotHub.GetTwinAsync(deviceId, moduleId, token);
-            string actualModelId = twin.ModelId;
-            Assert.AreEqual(expectedModelId, actualModelId);
+            return new Action<EdgeConfigBuilder>(
+                builder =>
+                {
+                    builder.GetModule(ModuleName.EdgeHub).WithEnvironment(new[] { ("UpstreamProtocol", protocol.ToString()) });
+                });
         }
     }
 }