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;
///
/// 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
/// FetchAndCache 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.
///
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() });
[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
{
[RevA.Value] = Artifact(),
[RevB.Value] = null,
});
var publishProbe = CreateTestProbe();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet { "driver" },
opcUaPublishActor: publishProbe.Ref,
deploymentArtifactCache: new NullReadCache(),
fetchAndCacheMode: true,
artifactFetcher: fetcher));
actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevA, CorrelationId.NewId()));
coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
publishProbe.ExpectMsg(TimeSpan.FromSeconds(5)); // revA rebuild
actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevB, CorrelationId.NewId()));
coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).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 { [RevA.Value] = Array.Empty() });
var publishProbe = CreateTestProbe();
var coordinator = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref,
localRoles: new HashSet { "driver" },
opcUaPublishActor: publishProbe.Ref,
deploymentArtifactCache: new NullReadCache(),
fetchAndCacheMode: true,
artifactFetcher: fetcher));
actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevA, CorrelationId.NewId()));
coordinator.ExpectMsg(TimeSpan.FromSeconds(5)).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 = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
new ThrowingDbFactory(), TestNode, ackRouter: null,
localRoles: new HashSet { "driver" },
opcUaPublishActor: publishProbe.Ref,
deploymentArtifactCache: new NullReadCache(), // corrupt → null, per the cache contract
fetchAndCacheMode: true,
artifactFetcher: new MapFetcher(new Dictionary())));
// 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(TimeSpan.FromSeconds(2));
},
duration: TimeSpan.FromSeconds(5));
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
return probe.ExpectMsg(TimeSpan.FromSeconds(5));
}
private sealed class MapFetcher(Dictionary map) : IDeploymentArtifactFetcher
{
public Task FetchAsync(string deploymentId, string expectedRevisionHash, CancellationToken ct)
{
map.TryGetValue(expectedRevisionHash, out var bytes);
return Task.FromResult(bytes);
}
}
/// A cache that reads empty (a miss, or the indistinguishable corrupt-artifact case).
private sealed class NullReadCache : IDeploymentArtifactCache
{
public Task StoreAsync(string clusterId, string deploymentId, string revisionHash,
byte[] artifact, CancellationToken ct = default)
=> Task.CompletedTask;
public Task GetCurrentAsync(string clusterId, CancellationToken ct = default)
=> Task.FromResult(null);
public Task GetCurrentUnkeyedAsync(CancellationToken ct = default)
=> Task.FromResult(null);
}
private sealed class ThrowingDbFactory : IDbContextFactory
{
public OtOpcUaConfigDbContext CreateDbContext()
=> throw new InvalidOperationException("ConfigDb unreachable");
}
}