lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/AdvertisedAggregateFunctionsWireTests.cs

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     Gitea #682 over a real SDK client session: <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201) lists
///     exactly the aggregates the wired historian backend serves — the data sources are built through
///     <see cref="HistorianBackendRegistry"/> exactly as the Host builds them — while
///     <c>ServerCapabilities/AggregateFunctions</c> (i=2997) keeps the SDK's full registered set in every configuration,
///     because a live-data <c>AggregateFilter</c> subscription keeps working for every one of them.
/// </summary>
public sealed class AdvertisedAggregateFunctionsWireTests
{
    private static readonly string[] GatewayNames = ["Maximum", "Minimum", "TimeAverage", "Total"];

    private static readonly string[] SqlNames =
    [
        "Average", "Count", "Delta", "End", "Interpolative", "Maximum", "MaximumActualTime", "Minimum", "MinimumActualTime",
        "PercentBad", "PercentGood", "Range", "Start", "TimeAverage", "TimeAverage2", "Total", "Total2",
    ];

    /// <summary>The SDK's 37 registered aggregates (<c>StandardServer.CreateAggregateManager</c>) — i=2997, always.</summary>
    private static readonly string[] AllSdkNames =
    [
        "AnnotationCount", "Average", "Count", "Delta", "DeltaBounds", "DurationBad", "DurationGood", "DurationInStateNonZero",
        "DurationInStateZero", "End", "EndBound", "Interpolative", "Maximum", "Maximum2", "MaximumActualTime",
        "MaximumActualTime2", "Minimum", "Minimum2", "MinimumActualTime", "MinimumActualTime2", "NumberOfTransitions",
        "PercentBad", "PercentGood", "Range", "Range2", "StandardDeviationPopulation", "StandardDeviationSample", "Start",
        "StartBound", "TimeAverage", "TimeAverage2", "Total", "Total2", "VariancePopulation", "VarianceSample",
        "WorstQuality", "WorstQuality2",
    ];

    private static IHistorianDataSource Backend(string backend)
    {
        var config = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
        {
            ["ServerHistorian:SqlServer:ConnectionString"] = "Server=localhost;User Id=sa;Password=x;TrustServerCertificate=True",
            ["ServerHistorian:QuestDb:ConnectionString"] = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb",
        }).Build();
        var options = new ServerHistorianOptions
        {
            Backend = backend, Enabled = true, Endpoint = "http://localhost:5222", ApiKey = "histgw_id_secret", UseTls = false,
        };
        // Constructing a data source opens no connection; nothing here reaches a database or a gateway.
        return HistorianBackendRegistry.Resolve(options, config).CreateDataSource(new ServiceCollection().BuildServiceProvider());
    }

    [Theory]
    [InlineData("Gateway")]
    [InlineData("SqlServer")]
    [InlineData("QuestDb")]
    public async Task History_folder_advertises_the_backend_set_and_the_server_folder_keeps_every_aggregate(string backend)
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await StartAsync($"AggAdv{backend}", encrypted: false, ct);
        try
        {
            rig.Nm.HistorianDataSource = Backend(backend);
            rig.Enforce(false, [Key()]);
            using var session = await rig.ConnectAsync(anonymous: true, ct);

            var expected = backend == "Gateway" ? GatewayNames : SqlNames;
            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(expected);
            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames);
        }
        finally { await rig.DisposeAsync(); }
    }

    [Fact]
    public async Task With_no_historian_wired_the_history_folder_is_empty_from_boot()
    {
        // The Null default is applied by the CreateAggregateManager override, before any source is assigned.
        var ct = TestContext.Current.CancellationToken;
        var rig = await StartAsync("AggAdvNull", encrypted: false, ct);
        try
        {
            rig.Enforce(false, [Key()]);
            using var session = await rig.ConnectAsync(anonymous: true, ct);

            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBeEmpty();
            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames);
        }
        finally { await rig.DisposeAsync(); }
    }

    [Fact]
    public async Task Rewiring_the_source_resyncs_both_directions_of_the_references()
    {
        var ct = TestContext.Current.CancellationToken;
        var rig = await StartAsync("AggAdvResync", encrypted: false, ct);
        try
        {
            rig.Enforce(false, [Key()]);
            using var session = await rig.ConnectAsync(anonymous: true, ct);

            rig.Nm.HistorianDataSource = Backend("SqlServer");
            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(SqlNames);
            (await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe(
                [ObjectIds.Server_ServerCapabilities_AggregateFunctions, ObjectIds.HistoryServerCapabilities_AggregateFunctions],
                ignoreOrder: true);

            // Gateway has no arithmetic Average: dropped from the history folder (inverse reference too), kept in the
            // server folder.
            rig.Nm.HistorianDataSource = Backend("Gateway");
            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(GatewayNames);
            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct)).ShouldBe(AllSdkNames);
            (await ParentsAsync(session, ObjectIds.AggregateFunction_Average, ct)).ShouldBe(
                [ObjectIds.Server_ServerCapabilities_AggregateFunctions]);

            // And back: removed references are restored.
            rig.Nm.HistorianDataSource = Backend("QuestDb");
            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct)).ShouldBe(SqlNames);
        }
        finally { await rig.DisposeAsync(); }
    }

    [Theory]
    [InlineData("StandardDeviationSample")] // served by no historian backend, never in the history folder
    [InlineData("Average")]                 // not served by the Gateway backend wired below
    public async Task A_live_AggregateFilter_subscription_for_an_unadvertised_aggregate_still_works(string aggregateName)
    {
        // The trim touches only the advertised folders, not the AggregateManager's calculator factories, which is what
        // validates and computes a live-data AggregateFilter. Removing a factory turns this create into
        // BadAggregateNotSupported.
        var ct = TestContext.Current.CancellationToken;
        var rig = await StartAsync($"AggAdvLive{aggregateName}", encrypted: false, ct);
        try
        {
            rig.Nm.HistorianDataSource = Backend("Gateway");
            rig.Enforce(false, [Key()]);
            var id = rig.EnsureRawVariable("agg", "Double", writable: false);
            using var session = await rig.ConnectAsync(anonymous: true, ct);
            var aggregateId = (NodeId)typeof(ObjectIds).GetField("AggregateFunction_" + aggregateName)!.GetValue(null)!;

            (await AdvertisedAsync(session, ObjectIds.HistoryServerCapabilities_AggregateFunctions, ct))
                .ShouldNotContain(aggregateName);
            (await AdvertisedAsync(session, ObjectIds.Server_ServerCapabilities_AggregateFunctions, ct))
                .ShouldContain(aggregateName); // i=2997 tells the truth: the live aggregate IS supported

            var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
            session.AddSubscription(subscription);
            await subscription.CreateAsync(ct);
            var item = new MonitoredItem(subscription.DefaultItem)
            {
                StartNodeId = ExpandedNodeId.ToNodeId(ExpandedNodeId.Parse(RawNsu(id)), session.NamespaceUris),
                AttributeId = Attributes.Value,
                SamplingInterval = 100,
                QueueSize = 10,
                Filter = new AggregateFilter
                {
                    AggregateType = aggregateId,
                    ProcessingInterval = 1000,
                    StartTime = DateTime.UtcNow,
                },
            };
            var notifications = 0;
            item.Notification += (mi, _) => Interlocked.Add(ref notifications, mi.DequeueValues().Count);
            subscription.AddItem(item);
            await subscription.ApplyChangesAsync(ct);

            item.Status.Error.ShouldBeNull($"AggregateFilter({aggregateName}) must be accepted on live data");
            // And it is computed: the SDK calculator emits one value per processing interval.
            var deadline = DateTime.UtcNow.AddSeconds(15);
            while (Volatile.Read(ref notifications) == 0 && DateTime.UtcNow < deadline) await Task.Delay(100, ct);
            Volatile.Read(ref notifications).ShouldBeGreaterThan(0, "the live aggregate must produce interval values");
        }
        finally { await rig.DisposeAsync(); }
    }

    /// <summary>Forward Organizes children of an AggregateFunctions folder, by BrowseName, sorted.</summary>
    private static async Task<List<string>> AdvertisedAsync(ISession session, NodeId folder, CancellationToken ct)
    {
        var refs = await BrowseAsync(session, folder, BrowseDirection.Forward, ct);
        return refs.Select(r => r.BrowseName.Name).OrderBy(n => n, StringComparer.Ordinal).ToList();
    }

    /// <summary>The folders an aggregate-function node is organized under (inverse Organizes).</summary>
    private static async Task<List<NodeId>> ParentsAsync(ISession session, NodeId aggregate, CancellationToken ct)
    {
        var refs = await BrowseAsync(session, aggregate, BrowseDirection.Inverse, ct);
        return refs.Select(r => ExpandedNodeId.ToNodeId(r.NodeId, session.NamespaceUris)).ToList();
    }

    private static async Task<ReferenceDescriptionCollection> BrowseAsync(
        ISession session, NodeId node, BrowseDirection direction, CancellationToken ct)
    {
        var response = await session.BrowseAsync(
            requestHeader: null,
            view: null,
            requestedMaxReferencesPerNode: 0,
            nodesToBrowse:
            [
                new BrowseDescription
                {
                    NodeId = node,
                    BrowseDirection = direction,
                    ReferenceTypeId = ReferenceTypeIds.Organizes,
                    IncludeSubtypes = false,
                    NodeClassMask = 0,
                    ResultMask = (uint)BrowseResultMask.All,
                },
            ],
            ct: ct);
        StatusCode.IsGood(response.Results[0].StatusCode).ShouldBeTrue($"browse {node}: {response.Results[0].StatusCode}");
        return response.Results[0].References;
    }
}