ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealOpcUaClientBatchRollbackTests.cs

using System.Collections;
using System.Collections.Concurrent;
using System.Reflection;
using NSubstitute;
using Opc.Ua;
using Opc.Ua.Client;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;

/// <summary>
/// Arch-review 2026-08-19 <b>P2-4</b> (rollback half) — a batch subscribe whose
/// <c>ApplyChanges</c> throws must leave NO residue behind.
///
/// <para>
/// <b>The gap this pins.</b> <c>CreateSubscriptionsCoreAsync</c> registers each monitored item as
/// it places it (<c>_callbacks</c>, <c>_itemShard</c>, the shard's item count and the SDK
/// subscription's own item list) and only issues <c>ApplyChanges</c> once the whole chunk is
/// placed. A throw there — a dead session being the common cause — propagated with every one of
/// those registrations still in place and nothing left holding a handle to release them. The actor
/// sees the throw as a whole-chunk fault and re-issues the SAME batch on its backoff timer, so
/// each probe grew the bookkeeping again: unbounded shard growth and phantom shard occupancy.
/// </para>
///
/// <para>
/// The seam is reached with a data shard whose SDK <see cref="Subscription"/> was never created,
/// which is precisely how the SDK reports "you have no usable subscription": placement and
/// <c>AddItem</c> succeed, <c>ApplyChangesAsync</c> throws. Private state is read by reflection
/// rather than widened for the test.
/// </para>
/// </summary>
[Trait("Category", "Unit")]
public class RealOpcUaClientBatchRollbackTests
{
    private static object Field(object target, string name) =>
        target.GetType().GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(target)!;

    /// <summary>
    /// Gives the client a mocked session plus ONE pre-placed (never created) data shard, so a
    /// batch subscribe places its items and then throws out of ApplyChanges.
    /// </summary>
    private static Subscription ArrangeUncreatedShard(RealOpcUaClient client)
    {
        var session = Substitute.For<ISession>();
        session.NamespaceUris.Returns(new NamespaceTable());
        typeof(RealOpcUaClient)
            .GetField("_session", BindingFlags.Instance | BindingFlags.NonPublic)!
            .SetValue(client, session);

        // The parameterless ctor is obsolete in the current SDK in favour of a telemetry-context
        // overload; nothing here publishes telemetry, and the alternative would be to hand-build a
        // context purely to satisfy the signature.
#pragma warning disable CS0618
        var subscription = new Subscription { DisplayName = "uncreated-shard" };
#pragma warning restore CS0618

        var shardType = typeof(RealOpcUaClient).GetNestedType("DataShard", BindingFlags.NonPublic)!;
        var shard = Activator.CreateInstance(
            shardType,
            BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic,
            null, [subscription], null)!;
        ((IList)Field(client, "_dataShards")).Add(shard);

        return subscription;
    }

    private static int ShardItemCount(RealOpcUaClient client)
    {
        var shard = ((IList)Field(client, "_dataShards"))[0]!;
        return (int)shard.GetType().GetProperty("ItemCount")!.GetValue(shard)!;
    }

    [Fact]
    public async Task ApplyChangesThrow_RollsBackEveryPartiallyCreatedItem()
    {
        var client = new RealOpcUaClient();
        var subscription = ArrangeUncreatedShard(client);

        var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync(
            ["i=2258", "i=2259", "i=2260"],
            (_, _, _, _) => { }));

        // Positive control: the batch really did fail at ApplyChanges, not before placement.
        Assert.NotNull(fault);

        // Nothing registered survives the throw — pre-call state, exactly.
        Assert.Empty((ConcurrentDictionary<string, Action<string, object?, DateTime, uint>>)
            Field(client, "_callbacks"));
        Assert.Empty((ConcurrentDictionary<string, Subscription>)Field(client, "_itemShard"));
        Assert.Empty((ConcurrentDictionary<string, MonitoredItem>)Field(client, "_monitoredItems"));
        Assert.Equal(0, ShardItemCount(client));
        Assert.Empty(subscription.MonitoredItems);
    }

    [Fact]
    public async Task RepeatedFailedBatches_DoNotGrowTheAdapterBookkeeping()
    {
        // The actual operational damage: the actor re-issues the same chunk on its backoff timer,
        // so an un-rolled-back batch compounded on every probe.
        var client = new RealOpcUaClient();
        ArrangeUncreatedShard(client);

        for (var attempt = 0; attempt < 3; attempt++)
        {
            var fault = await Record.ExceptionAsync(() => client.CreateSubscriptionsAsync(
                ["i=2258", "i=2259"],
                (_, _, _, _) => { }));
            Assert.NotNull(fault);
        }

        Assert.Empty((ConcurrentDictionary<string, Action<string, object?, DateTime, uint>>)
            Field(client, "_callbacks"));
        Assert.Empty((ConcurrentDictionary<string, Subscription>)Field(client, "_itemShard"));
        Assert.Equal(0, ShardItemCount(client));
    }
}