test(mesh): #485 empty/corrupt/mismatch coverage on the FetchAndCache paths
Phase 3 Task 7. Regression net pinning the #485 negative on every FetchAndCache seam — nothing rebuilds, nothing is torn down on a failure: (1) a null fetch mid-steady-state (all endpoints down / NotFound / SHA mismatch — indistinguishable to the actor) keeps the served address space and revision (no RebuildAddressSpace for the failed revision); (2) a zero-length blob handed to the actor fails via ReconcileDriversFromBlob's own #485 guard, independent of the fetcher's null contract; (3) a corrupt cache at boot (surfaced as GetCurrentUnkeyedAsync == null) lands Steady-no-revision and applies NOTHING — never a partial address space. No new mechanism; all green as a net. Removing the #485 empty guard reddens the zero-length test. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
+156
@@ -0,0 +1,156 @@
|
|||||||
|
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 = CreateTestProbe();
|
||||||
|
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>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||||
|
publishProbe.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5)); // revA rebuild
|
||||||
|
|
||||||
|
actor.Tell(new DispatchDeployment(DeploymentId.NewId(), RevB, CorrelationId.NewId()));
|
||||||
|
coordinator.ExpectMsg<ApplyAck>(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<string, byte[]?> { [RevA.Value] = Array.Empty<byte>() });
|
||||||
|
var publishProbe = CreateTestProbe();
|
||||||
|
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>(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<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: TimeSpan.FromSeconds(5));
|
||||||
|
|
||||||
|
actor.Tell(new GetDiagnostics(CorrelationId.NewId()), probe.Ref);
|
||||||
|
return probe.ExpectMsg<NodeDiagnosticsSnapshot>(TimeSpan.FromSeconds(5));
|
||||||
|
}
|
||||||
|
|
||||||
|
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");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user