ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs

using Akka.Actor;
using Akka.TestKit.Xunit2;
using Grpc.Net.Client;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Communication.Actors;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;

namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;

/// <summary>
/// A malformed <c>GrpcNodeAAddress</c>/<c>GrpcNodeBAddress</c> (review 2026-10-21 03/R-1) must not
/// park-then-dispose the site's still-referenced channel, must not stop the other sites in the same
/// refresh tick from being reconciled, and must not restart <see cref="CentralCommunicationActor"/>.
/// </summary>
public class SitePairChannelProviderBadAddressTests : TestKit
{
    /// <summary>Initializes the test kit with a quieter Akka log level.</summary>
    public SitePairChannelProviderBadAddressTests() : base(@"akka.loglevel = WARNING") { }

    private static SitePairChannelProvider Build(Func<string, HttpMessageHandler>? handlerFactory = null) => new(
        new NoKeyProvider(),
        Options.Create(new CommunicationOptions()),
        NullLogger<SitePairChannelProvider>.Instance,
        handlerFactory);

    private static Task<GrpcChannel> CurrentChannel(SitePairChannelProvider provider, string siteId) =>
        provider.ExecuteAsync(siteId, (channel, _) => Task.FromResult(channel), CancellationToken.None);

    /// <summary>Verifies that an endpoint with no scheme keeps the existing channel and endpoint and parks nothing.</summary>
    /// <param name="bad">A malformed gRPC endpoint.</param>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Theory]
    [InlineData("scadabridge-site-a-node-a")]
    [InlineData("site-a-node-a:8083")]
    [InlineData("ftp://site-a-node-a:8083")]
    public async Task UpdateSite_MalformedEndpoint_KeepsTheOldChannelAlive(string bad)
    {
        using var provider = Build();
        provider.UpdateSite("site-1", "http://localhost:15201", "http://localhost:15202");
        var before = await CurrentChannel(provider, "site-1");

        provider.UpdateSite("site-1", bad, "http://localhost:15202");

        Assert.Equal(0, provider.ParkedChannelCount);
        provider.SweepGraveyard(force: true);
        var after = await CurrentChannel(provider, "site-1");
        Assert.Same(before, after);
        _ = after.CreateCallInvoker(); // not disposed

        // The next tick re-presents the same bad value: still nothing parked (no double-park).
        provider.UpdateSite("site-1", bad, "http://localhost:15202");
        Assert.Equal(0, provider.ParkedChannelCount);
    }

    /// <summary>Verifies that a channel build that throws keeps the old channel and still applies the other node's valid change.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task UpdateSite_BuildThrows_KeepsOldSideAndAppliesTheOtherSide()
    {
        using var provider = Build(endpoint => endpoint.Contains("15299", StringComparison.Ordinal)
            ? throw new InvalidOperationException("No address resolver configured")
            : new SocketsHttpHandler());
        provider.UpdateSite("site-1", "http://localhost:15211", "http://localhost:15212");
        var beforeA = await CurrentChannel(provider, "site-1");

        provider.UpdateSite("site-1", "http://localhost:15299", "http://localhost:15213");

        // Only the B side really changed, so only the old B channel is parked.
        Assert.Equal(1, provider.ParkedChannelCount);
        Assert.Same(beforeA, await CurrentChannel(provider, "site-1"));
    }

    /// <summary>Verifies that a brand-new site whose only endpoint is malformed simply has no channel.</summary>
    [Fact]
    public async Task UpdateSite_NewSiteWithOnlyMalformedEndpoint_HasNoChannel()
    {
        using var provider = Build();
        provider.UpdateSite("site-1", "not a url", null);
        await Assert.ThrowsAsync<SiteChannelUnavailableException>(() => CurrentChannel(provider, "site-1"));
    }

    /// <summary>Verifies that one site that faults the reconcile does not stop later sites being reconciled.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReconcileSites_OneFaultingSite_OtherSitesStillReconciled()
    {
        using var provider = Build();
        var transport = new GrpcSiteTransport(
            provider, new CommunicationOptions(), NullLogger<GrpcSiteTransport>.Instance);

        // A whitespace site id makes UpdateSite throw; it is inserted FIRST so a foreach abort
        // would skip the good site.
        var contacts = new Dictionary<string, SiteGrpcEndpoints>
        {
            [" "] = new("http://localhost:15221", null),
            ["site-good"] = new("http://localhost:15222", null),
        };

        transport.ReconcileSites(new SiteAddressCacheLoaded(new[] { " ", "site-good" }, contacts, 1));

        _ = await CurrentChannel(provider, "site-good");
    }

    /// <summary>Verifies that a transport reconcile fault does not restart the actor (which would re-read the Sites table in a loop).</summary>
    [Fact]
    public void TransportReconcileFault_DoesNotRestartTheActor()
    {
        var repo = Substitute.For<ISiteRepository>();
        repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>
        {
            new("Bad", "site-bad") { GrpcNodeAAddress = "scadabridge-site-a-node-a" }
        });
        var services = new ServiceCollection();
        services.AddScoped(_ => repo);
        var sp = services.BuildServiceProvider();

        var transport = Substitute.For<ISiteCommandTransport>();
        transport.When(t => t.ReconcileSites(Arg.Any<SiteAddressCacheLoaded>()))
            .Do(_ => throw new UriFormatException("Invalid URI"));

        Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport)));

        AwaitAssert(() => transport.Received(1).ReconcileSites(Arg.Any<SiteAddressCacheLoaded>()),
            TimeSpan.FromSeconds(3));

        // A restart re-runs PreStart, whose refresh fires at TimeSpan.Zero — so a restart loop would
        // show up as repeated repository reads within a second.
        Thread.Sleep(TimeSpan.FromSeconds(1));
        repo.Received(1).GetAllSitesAsync(Arg.Any<CancellationToken>());
    }

    private sealed class NoKeyProvider : ISitePskProvider
    {
        /// <inheritdoc />
        public ValueTask<string> GetAsync(string siteId, CancellationToken ct) => new("k");
        /// <inheritdoc />
        public void Invalidate(string siteId) { }
    }
}