Add test for multiple function hosts

JatinSanghvi · JatinSanghvi · commit 7150d2b6e1c6 · 2022-09-10T22:08:26.000+05:30
diff --git a/test/Integration/IntegrationTestBase.cs b/test/Integration/IntegrationTestBase.cs
@@ -1,29 +1,36 @@
 ﻿// Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
-using Microsoft.Data.SqlClient;
-using Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Common;
 using System;
+using System.Collections.Generic;
 using System.Data.Common;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Net.Http;
 using System.Reflection;
 using System.Runtime.InteropServices;
 using System.Text;
 using System.Threading.Tasks;
+using Microsoft.Azure.WebJobs.Extensions.Sql.Samples.Common;
+using Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Common;
+using Microsoft.Data.SqlClient;
 using Xunit;
 using Xunit.Abstractions;
-using Microsoft.Azure.WebJobs.Extensions.Sql.Samples.Common;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Integration
 {
     public class IntegrationTestBase : IDisposable
     {
         /// <summary>
-        /// Host process for Azure Function CLI
+        /// Host process for Azure Function CLI. Useful when only one host process is involved.
         /// </summary>
-        protected Process FunctionHost { get; private set; }
+        protected Process FunctionHost => this.FunctionHostList.FirstOrDefault();
+
+        /// <summary>
+        /// Host processes for Azure Function CLI.
+        /// </summary>
+        protected List<Process> FunctionHostList { get; } = new List<Process>();
 
         /// <summary>
         /// Host process for Azurite local storage emulator. This is required for non-HTTP trigger functions:
@@ -164,44 +171,58 @@ protected void StartAzurite()
         /// - The functionName is different than its route.<br/>
         /// - You can start multiple functions by passing in a space-separated list of function names.<br/>
         /// </remarks>
-        protected void StartFunctionHost(string functionName, SupportedLanguages language, bool useTestFolder = false, DataReceivedEventHandler customOutputHandler = null)
+        protected void StartFunctionHost(string functionName, SupportedLanguages language, bool useTestFolder = false, DataReceivedEventHandler[] customOutputHandlers = null)
         {
             string workingDirectory = useTestFolder ? GetPathToBin() : Path.Combine(GetPathToBin(), "SqlExtensionSamples", Enum.GetName(typeof(SupportedLanguages), language));
             if (!Directory.Exists(workingDirectory))
             {
                 throw new FileNotFoundException("Working directory not found at " + workingDirectory);
             }
+
+            // Use a different port for each new host process, starting with the default port number: 7071.
+            int port = this.Port + this.FunctionHostList.Count;
+
             var startInfo = new ProcessStartInfo
             {
                 // The full path to the Functions CLI is required in the ProcessStartInfo because UseShellExecute is set to false.
                 // We cannot both use shell execute and redirect output at the same time: https://docs.microsoft.com//dotnet/api/system.diagnostics.processstartinfo.redirectstandardoutput#remarks
                 FileName = GetFunctionsCoreToolsPath(),
-                Arguments = $"start --verbose --port {this.Port} --functions {functionName}",
+                Arguments = $"start --verbose --port {port} --functions {functionName}",
                 WorkingDirectory = workingDirectory,
                 WindowStyle = ProcessWindowStyle.Hidden,
                 RedirectStandardOutput = true,
                 RedirectStandardError = true,
                 UseShellExecute = false
             };
+
             this.TestOutput.WriteLine($"Starting {startInfo.FileName} {startInfo.Arguments} in {startInfo.WorkingDirectory}");
-            this.FunctionHost = new Process
+
+            var functionHost = new Process
             {
                 StartInfo = startInfo
             };
 
+            this.FunctionHostList.Add(functionHost);
+
             // Register all handlers before starting the functions host process.
             var taskCompletionSource = new TaskCompletionSource<bool>();
-            this.FunctionHost.OutputDataReceived += this.TestOutputHandler;
-            this.FunctionHost.OutputDataReceived += SignalStartupHandler;
-            this.FunctionHost.OutputDataReceived += customOutputHandler;
+            functionHost.OutputDataReceived += SignalStartupHandler;
 
-            this.FunctionHost.ErrorDataReceived += this.TestOutputHandler;
+            if (customOutputHandlers != null)
+            {
+                foreach (DataReceivedEventHandler handler in customOutputHandlers)
+                {
+                    functionHost.OutputDataReceived += handler;
+                }
+            }
 
-            this.FunctionHost.Start();
-            this.FunctionHost.BeginOutputReadLine();
-            this.FunctionHost.BeginErrorReadLine();
+            functionHost.Start();
+            functionHost.OutputDataReceived += this.GetTestOutputHandler(functionHost.Id);
+            functionHost.ErrorDataReceived += this.GetTestOutputHandler(functionHost.Id);
+            functionHost.BeginOutputReadLine();
+            functionHost.BeginErrorReadLine();
 
-            this.TestOutput.WriteLine($"Waiting for Azure Function host to start...");
+            this.TestOutput.WriteLine("Waiting for Azure Function host to start...");
 
             const int FunctionHostStartupTimeoutInSeconds = 60;
             bool isCompleted = taskCompletionSource.Task.Wait(TimeSpan.FromSeconds(FunctionHostStartupTimeoutInSeconds));
@@ -212,8 +233,8 @@ protected void StartFunctionHost(string functionName, SupportedLanguages languag
             const int BufferTimeInSeconds = 5;
             Task.Delay(TimeSpan.FromSeconds(BufferTimeInSeconds)).Wait();
 
-            this.TestOutput.WriteLine($"Azure Function host started!");
-            this.FunctionHost.OutputDataReceived -= SignalStartupHandler;
+            this.TestOutput.WriteLine("Azure Function host started!");
+            functionHost.OutputDataReceived -= SignalStartupHandler;
 
             void SignalStartupHandler(object sender, DataReceivedEventArgs e)
             {
@@ -259,11 +280,17 @@ private static string GetFunctionsCoreToolsPath()
             return funcPath;
         }
 
-        private void TestOutputHandler(object sender, DataReceivedEventArgs e)
+
+        private DataReceivedEventHandler GetTestOutputHandler(int processId)
         {
-            if (e != null && !string.IsNullOrEmpty(e.Data))
+            return TestOutputHandler;
+
+            void TestOutputHandler(object sender, DataReceivedEventArgs e)
             {
-                this.TestOutput.WriteLine(e.Data);
+                if (e != null && !string.IsNullOrEmpty(e.Data))
+                {
+                    this.TestOutput.WriteLine($"[{processId}] {e.Data}");
+                }
             }
         }
 
@@ -343,14 +370,17 @@ public void Dispose()
                 this.TestOutput.WriteLine($"Failed to close connection. Error: {e1.Message}");
             }
 
-            try
-            {
-                this.FunctionHost?.Kill();
-                this.FunctionHost?.Dispose();
-            }
-            catch (Exception e2)
+            foreach (Process functionHost in this.FunctionHostList)
             {
-                this.TestOutput.WriteLine($"Failed to stop function host, Error: {e2.Message}");
+                try
+                {
+                    functionHost?.Kill();
+                    functionHost?.Dispose();
+                }
+                catch (Exception e2)
+                {
+                    this.TestOutput.WriteLine($"Failed to stop function host, Error: {e2.Message}");
+                }
             }
 
             try
diff --git a/test/Integration/SqlTriggerBindingIntegrationTests.cs b/test/Integration/SqlTriggerBindingIntegrationTests.cs
@@ -30,10 +30,10 @@ public SqlTriggerBindingIntegrationTests(ITestOutputHelper output) : base(output
         public async Task SingleOperationTriggerTest()
         {
             this.EnableChangeTrackingForTable("Products");
-            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp);
 
             var changes = new List<SqlChange<Product>>();
-            this.MonitorProductChanges(changes);
+            DataReceivedEventHandler[] changeHandlers = new[] { this.GetProductChangeHandler(changes, "SQL Changes: ") };
+            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp, useTestFolder: false, changeHandlers);
 
             // Considering the polling interval of 5 seconds and batch-size of 10, it should take around 15 seconds to
             // process 30 insert operations. Similar reasoning is used to set delays for update and delete operations.
@@ -65,10 +65,10 @@ public async Task SingleOperationTriggerTest()
         public async Task MultiOperationTriggerTest()
         {
             this.EnableChangeTrackingForTable("Products");
-            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp);
 
             var changes = new List<SqlChange<Product>>();
-            this.MonitorProductChanges(changes);
+            DataReceivedEventHandler[] changeHandlers = new[] { this.GetProductChangeHandler(changes, "SQL Changes: ") };
+            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp, useTestFolder: false, changeHandlers);
 
             // Insert + multiple updates to a row are treated as single insert with latest row values.
             this.InsertProducts(1, 5);
@@ -97,13 +97,51 @@ public async Task MultiOperationTriggerTest()
             changes.Clear();
         }
 
+        /// <summary>
+        /// Ensures correct functionality with user functions running across multiple functions host processes.
+        /// </summary>
+        [Fact]
+        public async Task MultiHostTriggerTest()
+        {
+            this.EnableChangeTrackingForTable("Products");
+
+            var changes = new List<SqlChange<Product>>();
+            DataReceivedEventHandler[] changeHandlers = new[] { this.GetProductChangeHandler(changes, "SQL Changes: ") };
+
+            // Prepare three function host processes
+            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp, useTestFolder: false, changeHandlers);
+            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp, useTestFolder: false, changeHandlers);
+            this.StartFunctionHost(nameof(ProductsTrigger), Common.SupportedLanguages.CSharp, useTestFolder: false, changeHandlers);
+
+            // Considering the polling interval of 5 seconds and batch-size of 10, it should take around 15 seconds to
+            // process 90 insert operations across all functions host processes. Similar reasoning is used to set delays
+            // for update and delete operations.
+            this.InsertProducts(1, 90);
+            await Task.Delay(TimeSpan.FromSeconds(20));
+            ValidateProductChanges(changes, 1, 90, SqlChangeOperation.Insert, id => $"Product {id}", id => id * 100);
+            changes.Clear();
+
+            // All table columns (not just the columns that were updated) would be returned for update operation.
+            this.UpdateProducts(1, 60);
+            await Task.Delay(TimeSpan.FromSeconds(15));
+            ValidateProductChanges(changes, 1, 60, SqlChangeOperation.Update, id => $"Updated Product {id}", id => id * 100);
+            changes.Clear();
+
+            // The properties corresponding to non-primary key columns would be set to the C# type's default values
+            // (null and 0) for delete operation.
+            this.DeleteProducts(31, 90);
+            await Task.Delay(TimeSpan.FromSeconds(15));
+            ValidateProductChanges(changes, 31, 90, SqlChangeOperation.Delete, _ => null, _ => 0);
+            changes.Clear();
+        }
+
         /// <summary>
         /// Tests the error message when the user table is not present in the database.
         /// </summary>
         [Fact]
         public void TableNotPresentTriggerTest()
         {
-            this.StartFunctionsHostAndWaitForError(
+            this.StartFunctionHostAndWaitForError(
                 nameof(TableNotPresentTrigger),
                 true,
                 "Could not find table: 'dbo.TableNotPresent'.");
@@ -115,7 +153,7 @@ public void TableNotPresentTriggerTest()
         [Fact]
         public void PrimaryKeyNotCreatedTriggerTest()
         {
-            this.StartFunctionsHostAndWaitForError(
+            this.StartFunctionHostAndWaitForError(
                 nameof(PrimaryKeyNotPresentTrigger),
                 true,
                 "Could not find primary key created in table: 'dbo.ProductsWithoutPrimaryKey'.");
@@ -128,7 +166,7 @@ public void PrimaryKeyNotCreatedTriggerTest()
         [Fact]
         public void ReservedPrimaryKeyColumnNamesTriggerTest()
         {
-            this.StartFunctionsHostAndWaitForError(
+            this.StartFunctionHostAndWaitForError(
                 nameof(ReservedPrimaryKeyColumnNamesTrigger),
                 true,
                 "Found reserved column name(s): '_az_func_ChangeVersion', '_az_func_AttemptCount', '_az_func_LeaseExpirationTime' in table: 'dbo.ProductsWithReservedPrimaryKeyColumnNames'." +
@@ -141,7 +179,7 @@ public void ReservedPrimaryKeyColumnNamesTriggerTest()
         [Fact]
         public void UnsupportedColumnTypesTriggerTest()
         {
-            this.StartFunctionsHostAndWaitForError(
+            this.StartFunctionHostAndWaitForError(
                 nameof(UnsupportedColumnTypesTrigger),
                 true,
                 "Found column(s) with unsupported type(s): 'Location' (type: geography), 'Geometry' (type: geometry), 'Organization' (type: hierarchyid)" +
@@ -154,7 +192,7 @@ public void UnsupportedColumnTypesTriggerTest()
         [Fact]
         public void ChangeTrackingNotEnabledTriggerTest()
         {
-            this.StartFunctionsHostAndWaitForError(
+            this.StartFunctionHostAndWaitForError(
                 nameof(ProductsTrigger),
                 false,
                 "Could not find change tracking enabled for table: 'dbo.Products'.");
@@ -177,17 +215,22 @@ ALTER TABLE [dbo].[{tableName}]
             ");
         }
 
-        private void MonitorProductChanges(List<SqlChange<Product>> changes)
+        private DataReceivedEventHandler GetProductChangeHandler(List<SqlChange<Product>> changes, string messagePrefix)
         {
-            int index = 0;
-            string prefix = "SQL Changes: ";
+            return ProductChangeHandler;
 
-            this.FunctionHost.OutputDataReceived += (sender, e) =>
+            void ProductChangeHandler(object sender, DataReceivedEventArgs e)
             {
-                if (e.Data != null && (index = e.Data.IndexOf(prefix, StringComparison.Ordinal)) >= 0)
+                int index = 0;
+
+                if (e.Data != null && (index = e.Data.IndexOf(messagePrefix, StringComparison.Ordinal)) >= 0)
                 {
-                    string json = e.Data[(index + prefix.Length)..];
-                    changes.AddRange(JsonConvert.DeserializeObject<IReadOnlyList<SqlChange<Product>>>(json));
+                    string json = e.Data[(index + messagePrefix.Length)..];
+
+                    lock (changes)
+                    {
+                        changes.AddRange(JsonConvert.DeserializeObject<IReadOnlyList<SqlChange<Product>>>(json));
+                    }
                 }
             };
         }
@@ -248,7 +291,7 @@ private static void ValidateProductChanges(List<SqlChange<Product>> changes, int
         /// <param name="functionName">Name of the user function that should cause error in trigger listener</param>
         /// <param name="useTestFolder">Whether the functions host should be launched from test folder</param>
         /// <param name="expectedErrorMessage">Expected error message string</param>
-        private void StartFunctionsHostAndWaitForError(string functionName, bool useTestFolder, string expectedErrorMessage)
+        private void StartFunctionHostAndWaitForError(string functionName, bool useTestFolder, string expectedErrorMessage)
         {
             string errorMessage = null;
             var tcs = new TaskCompletionSource<bool>();
@@ -268,7 +311,7 @@ void OutputHandler(object sender, DataReceivedEventArgs e)
             };
 
             // All trigger integration tests are only using C# functions for testing at the moment.
-            this.StartFunctionHost(functionName, Common.SupportedLanguages.CSharp, useTestFolder, OutputHandler);
+            this.StartFunctionHost(functionName, Common.SupportedLanguages.CSharp, useTestFolder, new DataReceivedEventHandler[] { OutputHandler });
             this.FunctionHost.OutputDataReceived -= OutputHandler;
             this.FunctionHost.Kill();
 
