lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorFetchAndCacheUnreadableTests.cs

using System.Text.Json;
using Akka.Actor;
using Microsoft.EntityFrameworkCore;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Fleet;
using ZB.MOM.WW.OtOpcUa.Commons.Types;
using ZB.MOM.WW.OtOpcUa.Configuration;
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;

/// <summary>
///     Per-cluster mesh Phase 3 (Task 7) — the #485 "empty/unreadable bytes = no answer, keep
///     last-known-good, never tear down to an empty address space" contract, proven on every
///     <c>FetchAndCache</c> seam. This adds no new mechanism; it is the regression net that pins the
///     NEGATIVE — nothing rebuilds, nothing is torn down — on the failure paths.
/// </summary>
public sealed class DriverHostActorFetchAndCacheUnreadableTests : RuntimeActorTestBase
{
    private static readonly NodeId TestNode = NodeId.Parse("driver-test");
    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
    private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));

    private static byte[] Artifact() =>
        JsonSerializer.SerializeToUtf8Bytes(new { DriverInstances = Array.Empty<object>() });

    [Fact]
    public void ANullFetchMidSteadyState_KeepsTheServedState_AndFiresNoRebuild()
    {
        // revA fetches good bytes and is served; a later revB fetch returns null (all endpoints down /
        // NotFound / SHA mismatch — all indistinguishable to the actor). The served address space must
        // NOT be rebuilt for revB and the revision must stay revA.
        var db = NewInMemoryDbFactory();
        var fetcher = new MapFetcher(new Dictionary<string, byte[]?>
        {
            [RevA.Value] = Artifact(),
            [RevB.Value] = null,
        });
        var publishProbe = this.CreateCleanOutcomePublishProbe();
        var coordinator = CreateTestProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            db, TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            opcUaPublishActor: publishProbe.Ref,
            deploymentArtifactCache: new NullReadCache(),
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevA, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Applied);
        publishProbe.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(PresenceBudget); // revA rebuild

        actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevB, CorrelationId.NewId()));
        coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Failed);

        // The #485 negative: no rebuild for the failed revB — the served address space is untouched.
        publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
        AskDiagnostics(actor).CurrentRevision.ShouldBe(RevA);
    }

    [Fact]
    public void AZeroLengthFetch_FailsTheApply_ViaTheActorsOwn485Guard()
    {
        // The fetcher contract already turns zero bytes into null, but prove the actor ALSO refuses an
        // empty blob handed to it directly: ReconcileDriversFromBlob's #485 guard fails the apply.
        var db = NewInMemoryDbFactory();
        var fetcher = new MapFetcher(new Dictionary<string, byte[]?> { [RevA.Value] = Array.Empty<byte>() });
        var publishProbe = this.CreateCleanOutcomePublishProbe();
        var coordinator = CreateTestProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            db, TestNode, coordinator.Ref,
            localRoles: new HashSet<string> { "driver" },
            opcUaPublishActor: publishProbe.Ref,
            deploymentArtifactCache: new NullReadCache(),
            fetchAndCacheMode: true,
            artifactFetcher: fetcher));

        actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevA, CorrelationId.NewId()));

        coordinator.ExpectMsg<ApplyAck>(PresenceBudget).Outcome.ShouldBe(ApplyAckOutcome.Failed);
        publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); // nothing applied, nothing rebuilt
    }

    [Fact]
    public void ACorruptCacheAtBoot_DoesNotApplyAPartialConfig()
    {
        // A truncated/corrupt cached artifact surfaces as GetCurrentUnkeyedAsync == null (the cache's
        // integrity check is deliberately indistinguishable from a miss). Boot must land
        // Steady-no-revision and apply NOTHING — never a partial address space.
        var publishProbe = this.CreateCleanOutcomePublishProbe();

        var actor = Sys.ActorOf(DriverHostActor.Props(
            new ThrowingDbFactory(), TestNode, ackRouter: null,
            localRoles: new HashSet<string> { "driver" },
            opcUaPublishActor: publishProbe.Ref,
            deploymentArtifactCache: new NullReadCache(), // corrupt → null, per the cache contract
            fetchAndCacheMode: true,
            artifactFetcher: new MapFetcher(new Dictionary<string, byte[]?>())));

        // No rebuild at boot — a null (corrupt/empty) cache is not a configuration to apply.
        publishProbe.ExpectNoMsg(TimeSpan.FromMilliseconds(750));
        AskDiagnostics(actor).CurrentRevision.ShouldBeNull();
    }

    private NodeDiagnosticsSnapshot AskDiagnostics(IActorRef actor)
    {
        var probe = CreateTestProbe();
        AwaitAssert(
            () =>
            {
                actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
                probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(2));
            },
            duration: PresenceBudget);

        actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
        return probe.ExpectMsg<NodeDiagnosticsSnapshot>(PresenceBudget);
    }

    private sealed class MapFetcher(Dictionary<string, byte[]?> map) : IDeploymentArtifactFetcher
    {
        public Task<byte[]?> FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
        {
            map.TryGetValue(expectedRevisionHash, out var bytes);
            return Task.FromResult(bytes);
        }
    }

    /// <summary>A cache that reads empty (a miss, or the indistinguishable corrupt-artifact case).</summary>
    private sealed class NullReadCache : IDeploymentArtifactCache
    {
        public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
                               byte[] artifact, CancellationToken ct = default)
            => Task.CompletedTask;

        public Task<CachedDeploymentArtifact?> GetCurrentAsync(string clusterId, CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);

        public Task<CachedDeploymentArtifact?> GetCurrentUnkeyedAsync(CancellationToken ct = default)
            => Task.FromResult<CachedDeploymentArtifact?>(null);
    }

    private sealed class ThrowingDbFactory : IDbContextFactory<OtOpcUaConfigDbContext>
    {
        public OtOpcUaConfigDbContext CreateDbContext()
            => throw new InvalidOperationException("ConfigDb unreachable");
    }
}