/* ========================================================================
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
* 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.
*
* The complete license agreement can be found here:
* http://opcfoundation.org/License/MIT/1.00/
* ======================================================================*/
// CA2000: system-under-test disposables are created per test and released at teardown;
// there is no cross-test resource leak. Suppressed file-level for the suite.
#pragma warning disable CA2000 // Dispose objects before losing scope
// CA2007: tests run without a SynchronizationContext; ConfigureAwait(false)
// adds noise without a behavioural benefit. Disabled file-level for the suite.
#pragma warning disable CA2007
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Crdt;
using Crdt.Transport;
using NUnit.Framework;
using Opc.Ua.Redundancy;
namespace Opc.Ua.Client.Redundancy.Tests
{
///
/// Coverage for the client-side CRDT shared key/value store.
///
[TestFixture]
[Category("ClientRedundancy")]
public sealed class ReplicatedSharedKeyValueStoreTests
{
[Test]
public async Task SetThenGetRoundTripsAsync()
{
await using var network = new InMemoryNetwork();
await using var store = new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
var value = new ByteString(new byte[] { 1, 2, 3, 4 });
await store.SetAsync("session/a", value).ConfigureAwait(false);
(bool found, ByteString got) = await store.TryGetAsync("session/a").ConfigureAwait(false);
Assert.That(found, Is.True);
Assert.That(got.ToArray(), Is.EqualTo(new byte[] { 1, 2, 3, 4 }));
}
[Test]
public async Task ReplicatesSetAndDeleteAcrossReplicasAsync()
{
await using var network = new InMemoryNetwork();
await using ReplicatedSharedKeyValueStore storeA = CreateStore(network);
await using ReplicatedSharedKeyValueStore storeB = CreateStore(network);
// Warm up both transports so neither misses the other's broadcasts.
await storeA.TryGetAsync("warmup").ConfigureAwait(false);
await storeB.TryGetAsync("warmup").ConfigureAwait(false);
var value = new ByteString(new byte[] { 1, 2, 3, 4 });
await storeA.SetAsync("session/abc", value).ConfigureAwait(false);
await AssertEventuallyAsync(
async () =>
{
(bool found, ByteString stored) = await storeB.TryGetAsync("session/abc").ConfigureAwait(false);
return found && stored.ToArray().AsSpan().SequenceEqual(value.ToArray());
},
"a value set on A replicates to B").ConfigureAwait(false);
bool existed = await storeA.DeleteAsync("session/abc").ConfigureAwait(false);
Assert.That(existed, Is.True, "delete reports the entry existed locally");
await AssertEventuallyAsync(
async () =>
{
(bool found, _) = await storeB.TryGetAsync("session/abc").ConfigureAwait(false);
return !found;
},
"a delete on A replicates to B").ConfigureAwait(false);
}
[Test]
public async Task DeleteReturnsFalseWhenKeyAbsentAsync()
{
await using var network = new InMemoryNetwork();
await using ReplicatedSharedKeyValueStore store = CreateStore(network);
bool existed = await store.DeleteAsync("missing").ConfigureAwait(false);
Assert.That(existed, Is.False);
}
[Test]
public async Task CompareAndSwapThrowsNotSupportedAsync()
{
await using var network = new InMemoryNetwork();
await using ReplicatedSharedKeyValueStore store = CreateStore(network);
Assert.That(
async () => await store.CompareAndSwapAsync("k", default, new ByteString(new byte[] { 1 }))
.ConfigureAwait(false),
Throws.TypeOf(),
"CRDT stores cannot provide a linearizable compare-and-swap");
}
[Test]
public async Task ScanReturnsOnlyMatchingPrefixAsync()
{
await using var network = new InMemoryNetwork();
await using ReplicatedSharedKeyValueStore store = CreateStore(network);
await store.SetAsync("session/a", new ByteString(new byte[] { 1 })).ConfigureAwait(false);
await store.SetAsync("session/b", new ByteString(new byte[] { 2 })).ConfigureAwait(false);
await store.SetAsync("other/c", new ByteString(new byte[] { 3 })).ConfigureAwait(false);
var keys = new List();
await foreach (KeyValuePair pair in store.ScanAsync("session/").ConfigureAwait(false))
{
keys.Add(pair.Key);
}
Assert.That(keys, Is.EquivalentTo(["session/a", "session/b"]));
}
[Test]
public async Task WatchThrowsNotSupportedAsync()
{
await using var network = new InMemoryNetwork();
await using ReplicatedSharedKeyValueStore store = CreateStore(network);
// WatchAsync throws synchronously (its body throws, so it is not a
// deferred iterator), hence the delegate is not awaited.
Assert.That(() => store.WatchAsync("session/"), Throws.TypeOf());
}
[Test]
public async Task ConstructorRejectsNullArgumentsAsync()
{
await using var network = new InMemoryNetwork();
Assert.That(
() => new ReplicatedSharedKeyValueStore(
ReplicaId.New(), null!, TimeProvider.System, CrdtReaderOptions.Default),
Throws.TypeOf(), "transport must not be null");
Assert.That(
() => new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, null!),
Throws.TypeOf(), "reader options must not be null");
Assert.That(
() => new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), null!, CrdtReaderOptions.Default),
Throws.TypeOf(), "time provider must not be null");
}
[Test]
public async Task DisposeIsIdempotentAsync()
{
await using var network = new InMemoryNetwork();
ReplicatedSharedKeyValueStore store = CreateStore(network);
await store.DisposeAsync().ConfigureAwait(false);
Assert.That(async () => await store.DisposeAsync().ConfigureAwait(false), Throws.Nothing);
}
private static ReplicatedSharedKeyValueStore CreateStore(InMemoryNetwork network)
{
return new ReplicatedSharedKeyValueStore(
ReplicaId.New(), network.CreateTransport(), TimeProvider.System, CrdtReaderOptions.Default);
}
private static async Task AssertEventuallyAsync(Func> condition, string message)
{
// Generous deadline: CRDT convergence over the in-memory gossip network is normally
// sub-second, but background loops can be CPU-starved on a loaded CI runner.
DateTime deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30);
while (DateTime.UtcNow < deadline)
{
if (await condition().ConfigureAwait(false))
{
return;
}
await Task.Delay(25).ConfigureAwait(false);
}
Assert.Fail(message);
}
}
}