WIP: Use Docker container to run RabbitMQ integration tests

Source/EventFlow.Elasticsearch.Tests/ElasticsearchRunner.cs

@@ -23,25 +23,17 @@
 
 using System;
 using System.Collections.Generic;
-using System.IO;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using EventFlow.Extensions;
 using EventFlow.TestHelpers;
-using EventFlow.TestHelpers.Installer;
 using NUnit.Framework;
 
 namespace EventFlow.Elasticsearch.Tests
 {
     [Category(Categories.Integration)]
     public class ElasticsearchRunner
     {
-        private static readonly SoftwareDescription SoftwareDescription = SoftwareDescription.Create(
-            "elasticsearch",
-            new Version(5, 3, 2),
-            "https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.3.2.zip");
-
         // ReSharper disable once ClassNeverInstantiated.Local
         private class ClusterHealth
         {
@@ -116,50 +108,30 @@ public async Task TestRunner()
 
         public static async Task<ElasticsearchInstance> StartAsync()
         {
-            if (string.IsNullOrEmpty(Environment.GetEnvironmentVariable("JAVA_HOME")))
-                throw new InvalidOperationException("The 'JAVA_HOME' environment variable is required");
-
-            var installedSoftware = await InstallHelper.InstallAsync(SoftwareDescription).ConfigureAwait(false);
-            var version = SoftwareDescription.Version;
-            var installPath = Path.Combine(installedSoftware.InstallPath,
-                $"elasticsearch-{version.Major}.{version.Minor}.{version.Build}");
-
-            var tcpPort = TcpHelper.GetFreePort();
-            var exePath = Path.Combine(installPath, "bin", "elasticsearch.bat");
-            var nodeName = $"node-{Guid.NewGuid():N}";
+            var disposable = await DockerHelper.StartContainerAsync(
+                    "elasticsearch:5.3.2-alpine",
+                    new[] {9200},
+                    new Dictionary<string, string>
+                    {
+                        {"gateway.expected_nodes", "1"},
+                        {"cluster.routing.allocation.disk.threshold_enabled", "false"}
+                    })
+                .ConfigureAwait(false);
+
+            var elasticsearchInstance = new ElasticsearchInstance(
+                new Uri("http://127.0.0.1:9200"),
+                disposable);
 
-            var settings = new Dictionary<string, string>
-            {
-                {"node.name", nodeName},
-                {"http.port", tcpPort.ToString()},
-                {"gateway.expected_nodes", "1"},
-                {"cluster.routing.allocation.disk.threshold_enabled", "false"}
-            };
-            var configFilePath = Path.Combine(installPath, "config", "elasticsearch.yml");
-            if (!File.Exists(configFilePath))
-            {
-                throw new ApplicationException($"Could not find config file at '{configFilePath}'");
-            }
-            File.WriteAllLines(configFilePath, settings.Select(kv => $"{kv.Key}: {kv.Value}"));
-
-            IDisposable processDisposable = null;
             try
             {
-                processDisposable = ProcessHelper.StartExe(exePath,
-                    $"[${nodeName}] started");
-
-                var elasticsearchInstance = new ElasticsearchInstance(
-                    new Uri($"http://127.0.0.1:{tcpPort}"),
-                    processDisposable);
-
                 await elasticsearchInstance.WaitForGreenStateAsync().ConfigureAwait(false);
                 await elasticsearchInstance.DeleteEverythingAsync().ConfigureAwait(false);
 
                 return elasticsearchInstance;
             }
             catch
             {
-                processDisposable.DisposeSafe("Failed to dispose Elasticsearch process");
+                elasticsearchInstance.DisposeSafe("Failed to dispose Elasticsearch process");
                 throw;
             }
         }

Source/EventFlow.Elasticsearch.Tests/app.config

@@ -9,6 +9,10 @@
         <assemblyIdentity name="Moq" publicKeyToken="69f491c39445e920" culture="neutral"/>
         <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="4.7.99.0"/>
       </dependentAssembly>
+      <dependentAssembly>
+        <assemblyIdentity name="Newtonsoft.Json" publicKeyToken="30ad4fe6b2a6aeed" culture="neutral"/>
+        <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="10.0.0.0"/>
+      </dependentAssembly>
     </assemblyBinding>
   </runtime>
 </configuration>

Source/EventFlow.RabbitMQ.Tests/Integration/RabbitMqTests.cs

@@ -46,25 +46,29 @@ namespace EventFlow.RabbitMQ.Tests.Integration
     [Category(Categories.Integration)]
     public class RabbitMqTests
     {
-        private Uri _uri;
+        private static readonly Uri RabbitMqUri = new Uri("amqp://localhost:5672");
+        private IDisposable _rabbitMq;
 
-        [SetUp]
-        public void SetUp()
+        [OneTimeSetUp]
+        public void OneTimeSetUp()
         {
-            var url = Environment.GetEnvironmentVariable("RABBITMQ_URL");
-            if (string.IsNullOrEmpty(url))
-            {
-                Assert.Inconclusive("The environment variable named 'RABBITMQ_URL' isn't set. Set it to e.g. 'amqp://localhost'");
-            }
+            _rabbitMq = DockerHelper.StartContainerAsync(
+                "library/rabbitmq:3.6-management-alpine",
+                new[] { 5672, 15672 }).Result;
+            Thread.Sleep(TimeSpan.FromSeconds(5)); // let RabbitMQ start TODO, poke for access
+        }
 
-            _uri = new Uri(url);
+        [OneTimeTearDown]
+        public void OneTimeTearDown()
+        {
+            _rabbitMq.DisposeSafe("Failed to dispose RabbitMQ");
         }
 
         [Test, Timeout(10000)]
         public async Task Scenario()
         {
             var exchange = new Exchange($"eventflow-{Guid.NewGuid():N}");
-            using (var consumer = new RabbitMqConsumer(_uri, exchange, new[] { "#" }))
+            using (var consumer = new RabbitMqConsumer(RabbitMqUri, exchange, new[] { "#" }))
             using (var resolver = BuildResolver(exchange))
             {
                 var commandBus = resolver.Resolve<ICommandBus>();
@@ -95,7 +99,7 @@ public async Task PublisherPerformance()
             const int messagesPrThread = 200;
             const int totalMessageCount = taskCount * messagesPrThread;
 
-            using (var consumer = new RabbitMqConsumer(_uri, exchange, new[] { "#" }))
+            using (var consumer = new RabbitMqConsumer(RabbitMqUri, exchange, new[] { "#" }))
             using (var resolver = BuildResolver(exchange, o => o.RegisterServices(sr => sr.Register<ILog, NullLog>())))
             {
                 var rabbitMqPublisher = resolver.Resolve<IRabbitMqPublisher>();
@@ -143,7 +147,7 @@ private IRootResolver BuildResolver(Exchange exchange, Func<IEventFlowOptions, I
             configure = configure ?? (e => e);
 
             return configure(EventFlowOptions.New
-                .PublishToRabbitMq(RabbitMqConfiguration.With(_uri, false, exchange: exchange.Value))
+                .PublishToRabbitMq(RabbitMqConfiguration.With(RabbitMqUri, false, exchange: exchange.Value))
                 .AddDefaults(EventFlowTestHelpers.Assembly))
                 .CreateResolver(false);
         }

Source/EventFlow.RabbitMQ.Tests/app.config

@@ -9,6 +9,10 @@
         <assemblyIdentity name="Moq" publicKeyToken="69f491c39445e920" culture="neutral"/>
         <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="4.7.99.0"/>
       </dependentAssembly>
+      <dependentAssembly>
+        <assemblyIdentity name="Newtonsoft.Json" publicKeyToken="30ad4fe6b2a6aeed" culture="neutral"/>
+        <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="10.0.0.0"/>
+      </dependentAssembly>
     </assemblyBinding>
   </runtime>
 </configuration>

Source/EventFlow.TestHelpers/DockerHelper.cs

@@ -0,0 +1,154 @@
+// The MIT License (MIT)
+//
+// Copyright (c) 2015-2018 Rasmus Mikkelsen
+// Copyright (c) 2015-2018 eBay Software Foundation
+// https://github.com/eventflow/EventFlow
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of
+// this software and associated documentation files (the "Software"), to deal in
+// the Software without restriction, including without limitation the rights to
+// use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+// FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+// COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+// IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+// CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Docker.DotNet;
+using Docker.DotNet.Models;
+using EventFlow.Core;
+using NUnit.Framework;
+
+namespace EventFlow.TestHelpers
+{
+    public static class DockerHelper
+    {
+        private static readonly DockerClient DockerClient = new DockerClientConfiguration(new Uri("npipe://./pipe/docker_engine"))
+            .CreateClient();
+
+        public static async Task<IDisposable> StartContainerAsync(
+            string image,
+            IReadOnlyCollection<int> ports = null,
+            IReadOnlyDictionary<string, string> environment = null)
+        {
+            using (var ts = new CancellationTokenSource(TimeSpan.FromMinutes(5)))
+            {
+                var env = environment
+                    ?.OrderBy(kv => kv.Key)
+                    .Select(kv => $"{kv.Key}={kv.Value}")
+                    .ToList();
+                LogHelper.Log.Information($"Starting container with image '{image}'");
+
+                LogHelper.Log.Information($"Pulling image {image}");
+                var imageAndTag = image.Split(':');
+                await DockerClient.Images.CreateImageAsync(
+                        new ImagesCreateParameters
+                        {
+                            FromImage = imageAndTag[0],
+                            Tag = imageAndTag.Length > 1 ? imageAndTag[1] : null,
+                        },
+                        null,
+                        new Progress<JSONMessage>(m => LogHelper.Log.Verbose($"{m.ProgressMessage} ({m.ID})")),
+                        ts.Token)
+                    .ConfigureAwait(false);
+                while (true)
+                {
+                    var imagesListResponses = await DockerClient.Images.ListImagesAsync(
+                            new ImagesListParameters(),
+                            ts.Token)
+                        .ConfigureAwait(false);
+                    if (imagesListResponses.Any(i =>
+                        i.RepoTags != null &&
+                        i.RepoTags.Any(t => string.Equals(t, image, StringComparison.OrdinalIgnoreCase))))
+                    {
+                        break;
+                    }
+
+                    Thread.Sleep(TimeSpan.FromSeconds(1));
+                }
+
+                var createContainerResponse = await DockerClient.Containers.CreateContainerAsync(
+                        new CreateContainerParameters(
+                            new Config
+                            {
+                                Image = image,
+                                Env = env,
+                                ExposedPorts = ports?
+                                    .ToDictionary(p => $"{p}/tcp", p => new EmptyStruct()),
+                            })
+                        {
+                            HostConfig = new HostConfig
+                            {
+                                PortBindings = ports?.ToDictionary(
+                                    p => $"{p}/tcp",
+                                    p => (IList<PortBinding>) new List<PortBinding>
+                                    {
+                                        new PortBinding {HostPort = p.ToString()}
+                                    })
+                            }
+                        },
+                        ts.Token)
+                    .ConfigureAwait(false);
+                LogHelper.Log.Information($"Successfully created container '{createContainerResponse.ID}'");
+
+                await DockerClient.Containers.StartContainerAsync(
+                        createContainerResponse.ID,
+                        new ContainerStartParameters(),
+                        ts.Token)
+                    .ConfigureAwait(false);
+                LogHelper.Log.Information($"Successfully started container '{createContainerResponse.ID}'");
+
+                return new DisposableAction(() => StopContainer(createContainerResponse.ID));
+            }
+        }
+
+        private static void StopContainer(string id)
+        {
+            try
+            {
+                DockerClient.Containers.StopContainerAsync(
+                    id,
+                    new ContainerStopParameters
+                    {
+                        WaitBeforeKillSeconds = 5
+                    }).Wait();
+                LogHelper.Log.Information($"Stopped container {id}");
+
+                DockerClient.Containers.RemoveContainerAsync(
+                    id,
+                    new ContainerRemoveParameters
+                    {
+                        Force = true
+                    }).Wait();
+                LogHelper.Log.Information($"Removed container {id}");
+            }
+            catch (Exception e)
+            {
+                Console.WriteLine(e);
+            }
+        }
+
+        [Test, Explicit]
+        public static async Task Test()
+        {
+            using (await StartContainerAsync(
+                "rabbitmq:3.6-management-alpine",
+                new []{ 15672 }))
+            {
+                Thread.Sleep(TimeSpan.FromSeconds(5));
+            }
+        }
+    }
+}

Source/EventFlow.TestHelpers/EventFlow.TestHelpers.csproj

@@ -26,6 +26,7 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="AutoFixture.AutoMoq" Version="3.50.6" />
+    <PackageReference Include="Docker.DotNet" Version="3.125.1" />
     <PackageReference Include="FluentAssertions" Version="4.19.4" />
     <PackageReference Include="Moq" Version="4.7.99" />
     <PackageReference Include="NUnit" Version="3.8.1" />

Source/EventFlow.TestHelpers/app.config

@@ -9,6 +9,10 @@
         <assemblyIdentity name="Moq" publicKeyToken="69f491c39445e920" culture="neutral"/>
         <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="4.7.99.0"/>
       </dependentAssembly>
+      <dependentAssembly>
+        <assemblyIdentity name="Newtonsoft.Json" publicKeyToken="30ad4fe6b2a6aeed" culture="neutral"/>
+        <bindingRedirect oldVersion="0.0.0.0-65535.65535.65535.65535" newVersion="10.0.0.0"/>
+      </dependentAssembly>
     </assemblyBinding>
   </runtime>
 </configuration>

appveyor.yml

@@ -14,6 +14,9 @@ environment:
   RABBITMQ_URL:
     secure: es47RGxjYJo4Ms8YR7ZY1R93OayJf6z3CVH2oUZMChaDXtchygrccwBPIBB8doWH68S13m/4p99EmtjdrudbM1gelesdwewX5ouJXO9wyo1/zZhNZDrUpb21Ojw6O1hH
 
+install:
+  - docker version
+
 test: off
 
 artifacts: