[Bug] Fix mongo conventions (#10)

* Fix mongo conventions

* Move convention initialization to ScheduleRepository

* Configure GitHub actions for mongo tests (#2)

* Test with real mongo

* Remove msbuild

* test on linux

* Test with old document for deserialization

Author: bannikovilea

.github/workflows/dotnet-desktop.yaml

@@ -51,8 +51,9 @@ jobs:
         strategy:
             matrix:
                 configuration: [Release]
+                mongodb-version: ['4.2', '4.4', '5.0', '6.0']
 
-        runs-on: windows-latest  # For a list of available runner types, refer to
+        runs-on: ubuntu-latest  # For a list of available runner types, refer to
         # https://help.github.com/en/actions/reference/workflow-syntax-for-github-actions#jobsjob_idruns-on
 
         env:
@@ -70,9 +71,10 @@ jobs:
               with:
                   dotnet-version: 6.0.x
 
-            # Add  MSBuild to the PATH: https://github.com/microsoft/setup-msbuild
-            - name: Setup MSBuild.exe
-              uses: microsoft/[email protected]
+            - name: Start MongoDB
+              uses: supercharge/[email protected]
+              with:
+                mongodb-version: ${{ matrix.mongodb-version }}
 
             # Execute all unit tests in the solution
             - name: Execute unit tests

Source/EasyNetQ.Scheduler.Mongo.Tests/BSONs/Pending.json

@@ -0,0 +1,42 @@
+{
+  "_id": BinData(3, 'LH9BsL6fuEWaP4utMDFX1w=='),
+  "WakeTime" : ISODate("2022-09-28T07:17:22.451Z"),
+  "BindingKey" : "Message",
+  "Exchange" : "Message",
+  "ExchangeType" : "topic",
+  "RoutingKey" : "#",
+  "InnerMessage" : BinData(0,"eyJldmVudFR5cGUiOiJTZW5kTGVhcm5Qcm9tb0FjdGlvbkVtYWlsMSIsInVzZXJJZCI6Ijk1Njc0YzBhMmY5YjQ5ZjRhZjQ1MWQ4YTlmODQ0YzkzIiwidHJ5Q291bnQiOjB9"),
+  "BasicProperties" : {
+    "ContentType" : null,
+    "ContentEncoding" : null,
+    "Headers" : {
+
+    },
+    "DeliveryMode" : 2,
+    "Priority" : 0,
+    "CorrelationId" : "3952ffcc-a6f9-4706-87f9-9edbfc5f74b7",
+    "ReplyTo" : null,
+    "Expiration" : null,
+    "MessageId" : null,
+    "Timestamp" : 0,
+    "Type" : "Message",
+    "UserId" : null,
+    "AppId" : null,
+    "ClusterId" : null,
+    "ContentTypePresent" : false,
+    "ContentEncodingPresent" : false,
+    "HeadersPresent" : true,
+    "DeliveryModePresent" : true,
+    "PriorityPresent" : false,
+    "CorrelationIdPresent" : true,
+    "ReplyToPresent" : false,
+    "ExpirationPresent" : false,
+    "MessageIdPresent" : false,
+    "TimestampPresent" : false,
+    "TypePresent" : true,
+    "UserIdPresent" : false,
+    "AppIdPresent" : false,
+    "ClusterIdPresent" : false
+  },
+  "State" : 1
+}

Source/EasyNetQ.Scheduler.Mongo.Tests/EasyNetQ.Scheduler.Mongo.Tests.csproj

@@ -25,6 +25,11 @@
     <Reference Include="System" />
     <Reference Include="Microsoft.CSharp" />
   </ItemGroup>
+  <ItemGroup>
+  <None Include="BSONs\Pending.json">
+    <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+  </None>
+  </ItemGroup>
   <ItemGroup>
     <Service Include="{82a7f48d-3b50-4b1e-b82e-3ada8210c358}" />
   </ItemGroup>

Source/EasyNetQ.Scheduler.Mongo.Tests/ScheduleRepositoryTests.cs

@@ -1,72 +1,109 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Text;
+using MongoDB.Driver;
 using Xunit;
 
 namespace EasyNetQ.Scheduler.Mongo.Tests
 {
-    public class ScheduleRepositoryTests
+    public class ScheduleRepositoryTests : IClassFixture<SchedulesDatabaseFixture>
     {
         private readonly ScheduleRepository scheduleRepository;
+        private readonly SchedulesDatabaseFixture databaseFixture;
 
-        public ScheduleRepositoryTests()
+        public ScheduleRepositoryTests(SchedulesDatabaseFixture databaseFixture)
         {
-            var configuration = new ScheduleRepositoryConfiguration
-                {
-                    ConnectionString = "mongodb://localhost:27017/?w=1&readPreference=primary&uuidRepresentation=csharpLegacy",
-                    CollectionName = "Schedules",
-                    DatabaseName = "EasyNetQ",
-                    DeleteTimeout = TimeSpan.FromMinutes(5),
-                    PublishTimeout = TimeSpan.FromSeconds(60),
-                };
-            scheduleRepository = new ScheduleRepository(configuration, () => DateTime.UtcNow);
+            this.databaseFixture = databaseFixture;
+            scheduleRepository = new ScheduleRepository(databaseFixture.Configuration, () => DateTime.UtcNow);
         }
 
-        [Fact(Skip = "Required a database")]
+        [Fact]
         public void Should_be_able_to_store_a_schedule()
         {
-            scheduleRepository.Store(new Schedule
-                {
-                    BindingKey = "abc",
-                    CancellationKey = "bcd",
-                    WakeTime = new DateTime(2011, 5, 18),
-                    InnerMessage = Encoding.UTF8.GetBytes("Hello World!"),
-                    Id = Guid.Empty,
-                    State = ScheduleState.Pending,
-                });
+            var schedule = new Schedule
+            {
+                BindingKey = "abc",
+                CancellationKey = "bcd",
+                WakeTime = new DateTime(2011, 5, 18),
+                InnerMessage = Encoding.UTF8.GetBytes("Hello World!"),
+                Id = Guid.NewGuid(),
+                State = ScheduleState.Pending,
+            };
+            scheduleRepository.Store(schedule);
+
+            var insertedDoc = databaseFixture.GetDocument(schedule.Id);
+
+            Assert.Equal(schedule.Id, insertedDoc["_id"].AsGuid);
+            Assert.Equal(schedule.CancellationKey, insertedDoc["CancellationKey"].AsString);
         }
 
-        [Fact(Skip = "Required a database")]
+        [Fact]
         public void Should_be_able_to_cancel_a_schedule()
         {
-            scheduleRepository.Cancel("bcd");
+            var cancellation = Guid.NewGuid().ToString();
+            databaseFixture.InitScheduleWithCancellation(cancellation);
+
+            scheduleRepository.Cancel(cancellation);
+
+            var count = databaseFixture.Collection.CountDocuments(d => d["CancellationKey"] == cancellation);
+            Assert.Equal(0, count);
         }
 
-        [Fact(Skip = "Required a database")]
+        [Fact]
         public void Should_be_able_to_get_messages()
         {
-            while (true)
-            {
-                var schedule = scheduleRepository.GetPending();
-                if(schedule == null)
-                    break;
-                Console.WriteLine("{0}, {1}, {2}",
-                                  schedule.BindingKey,
-                                  schedule.WakeTime,
-                                  Encoding.UTF8.GetString(schedule.InnerMessage));
-            }
+            var id = Guid.NewGuid();
+            var wakeTime = DateTime.MinValue.AddDays(5);
+            databaseFixture.InitializeOne(id, ScheduleState.Pending, wakeTime);
+
+            var schedule = scheduleRepository.GetPending();
+
+            Assert.Equal(id, schedule.Id);
         }
 
-        [Fact(Skip = "Required a database")]
+        [Fact]
         public void Should_be_able_to_handle_timeout()
         {
+            var id = Guid.NewGuid();
+            var publishingTime = DateTime.UtcNow.Add(-databaseFixture.Configuration.PublishTimeout);
+            databaseFixture.InitializeOne(id, ScheduleState.Publishing, publishingTime: publishingTime);
+
             scheduleRepository.HandleTimeout();
+
+            var actual = databaseFixture.GetDocument(id);
+            Assert.Equal(ScheduleState.Pending, actual["State"]);
         }
 
 
-        [Fact(Skip = "Required a database")]
+        [Fact]
         public void Should_be_able_to_mark_as_published()
         {
-            scheduleRepository.MarkAsPublished(Guid.Empty);
+            var id = Guid.NewGuid();
+            databaseFixture.InitializeOne(id);
+            scheduleRepository.MarkAsPublished(id);
+
+            var actual = databaseFixture.GetDocument(id);
+
+            Assert.Equal(ScheduleState.Published, actual["State"]);
+        }
+
+        [Fact]
+        public void Should_be_able_to_get_original_message()
+        {
+            databaseFixture.InitDocumentsWithOriginalBSON();
+            var actualIds = new HashSet<Guid>();
+            while (true)
+            {
+                var schedule = scheduleRepository.GetPending();
+                if (schedule == null)
+                    break;
+
+                actualIds.Add(schedule.Id);
+            }
+
+            Assert.True(actualIds.Count > 0);
+            Assert.Contains(databaseFixture.PendingOldDocId, actualIds);
         }
     }
 }

Source/EasyNetQ.Scheduler.Mongo.Tests/SchedulesDatabaseFixture.cs

@@ -0,0 +1,99 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using MongoDB.Bson;
+using MongoDB.Bson.Serialization;
+using MongoDB.Bson.Serialization.Conventions;
+using MongoDB.Driver;
+
+namespace EasyNetQ.Scheduler.Mongo.Tests
+{
+
+    public class SchedulesDatabaseFixture : IDisposable
+    {
+        public readonly IMongoCollection<BsonDocument> Collection;
+        public readonly Guid PendingOldDocId = Guid.Parse("b0417f2c-9fbe-45b8-9a3f-8bad303157d7");
+        private readonly IMongoDatabase database;
+
+        public readonly ScheduleRepositoryConfiguration Configuration = new ScheduleRepositoryConfiguration
+        {
+            ConnectionString = "mongodb://localhost:27017/?w=1&readPreference=primary&uuidRepresentation=csharpLegacy",
+            CollectionName = "Schedules",
+            DatabaseName = "EasyNetQ",
+            DeleteTimeout = TimeSpan.FromMinutes(5),
+            PublishTimeout = TimeSpan.FromSeconds(60),
+        };
+
+        public SchedulesDatabaseFixture()
+        {
+            InitializeMongoConventions();
+            database = CreateDatabase(new MongoUrl(Configuration.ConnectionString), Configuration.DatabaseName);
+            Collection = database.GetCollection<BsonDocument>(Configuration.CollectionName);
+        }
+
+        public void Dispose()
+        {
+
+        }
+
+        public void InitScheduleWithCancellation(string cancellation)
+        {
+            var schedule = new Schedule
+            {
+                CancellationKey = cancellation,
+                Id = Guid.NewGuid()
+            };
+            Collection.InsertOne(schedule.ToBsonDocument());
+        }
+
+        public void InitializeOne(Guid id, ScheduleState state = ScheduleState.Pending, DateTime? wakeTime = null, DateTime? publishingTime = null)
+        {
+            var schedule = new Schedule
+            {
+                Id = id,
+                State = state,
+                WakeTime = wakeTime ?? DateTime.UtcNow.AddDays(-1),
+                PublishingTime = publishingTime
+            };
+            Collection.InsertOne(schedule.ToBsonDocument());
+        }
+
+        public BsonDocument GetDocument(Guid id)
+            => database.GetCollection<BsonDocument>(Configuration.CollectionName)
+                .Find(d => d["_id"] == id)
+                .FirstOrDefault();
+
+        public void InitDocumentsWithOriginalBSON()
+        {
+            var bsons = LoadOldBSONs().ToList();
+            Collection.InsertMany(bsons);
+        }
+
+        private static IEnumerable<BsonDocument> LoadOldBSONs()
+        {
+            var path = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "BSONs");
+            foreach (var fileName in Directory.GetFiles(path))
+            {
+                var content = File.ReadAllText(fileName);
+                var bson = BsonSerializer.Deserialize<BsonDocument>(content);
+                yield return  bson;
+            }
+        }
+
+        private static IMongoDatabase CreateDatabase(MongoUrl connectionString, string databaseName)
+        {
+            var settings = MongoClientSettings.FromUrl(connectionString);
+            settings.ReadEncoding = new UTF8Encoding(false, false);
+            var client = new MongoClient(settings);
+            return client.GetDatabase(databaseName);
+        }
+
+        private static void InitializeMongoConventions()
+        {
+            ConventionRegistry.Register("ignoreIfNull", new ConventionPack {new IgnoreIfNullConvention(true)}, t => true);
+            ConventionRegistry.Register("ignoreExtraElements", new ConventionPack {new IgnoreExtraElementsConvention(true)}, t => true);
+        }
+    }
+}

Source/EasyNetQ.Scheduler.Mongo/ScheduleRepository.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Text;
+using MongoDB.Bson.Serialization.Conventions;
 using MongoDB.Driver;
-using MongoDB.Driver.Builders;
 
 namespace EasyNetQ.Scheduler.Mongo
 {
@@ -79,6 +79,7 @@ public void HandleTimeout()
 
         private IMongoCollection<Schedule> CreateAndIndex()
         {
+            InitializeMongoConventions();
             var collection = GetICollection<Schedule>(configuration.ConnectionString, configuration.DatabaseName, configuration.CollectionName);
             collection.Indexes.CreateOne(new CreateIndexModel<Schedule>(
                 Builders<Schedule>.IndexKeys.Ascending(x => x.CancellationKey),
@@ -112,5 +113,11 @@ private static IMongoCollection<TDocument> GetICollection<TDocument>(string conn
             var database = CreateDatabase(new MongoUrl(connectionString), databaseName);
             return database.GetCollection<TDocument>(collectionName);
         }
+
+        private static void InitializeMongoConventions()
+        {
+            ConventionRegistry.Register("ignoreIfNull", new ConventionPack {new IgnoreIfNullConvention(true)}, t => true);
+            ConventionRegistry.Register("ignoreExtraElements", new ConventionPack {new IgnoreExtraElementsConvention(true)}, t => true);
+        }
     }
 }