feat(commons/transport): ScriptArtifactsChanged invalidation contract + post-commit publish from bundle import (consumer lands in plan 06)
New Commons seam: ScriptArtifactsChanged record + ScriptArtifactKinds + the IScriptArtifactChangeBus pub/sub interface. Host ships InProcessScriptArtifactChangeBus (lock-free copy-on-write, swallow-and-log per subscriber) registered as a central-role singleton. BundleImporter gains an optional IScriptArtifactChangeBus? and, AFTER tx.CommitAsync, publishes one notification per script-bearing kind (ApiMethod/ SharedScript/Template) whose resolution was Overwrite/Rename, using post-resolution names — fully guarded so a bad subscriber can't fail a committed import. Contract + plan-06 handoff table documented. No consumer yet (additive; publisher is correct). Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting;
|
||||
using ZB.MOM.WW.ScadaBridge.Host.Services;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// #05-T14: unit tests for the in-process script-artifact change bus — delivery,
|
||||
/// unsubscribe, and the swallow-and-log guard that keeps a faulting subscriber from
|
||||
/// faulting the producer.
|
||||
/// </summary>
|
||||
public sealed class InProcessScriptArtifactChangeBusTests
|
||||
{
|
||||
private static InProcessScriptArtifactChangeBus NewBus() =>
|
||||
new(NullLogger<InProcessScriptArtifactChangeBus>.Instance);
|
||||
|
||||
private static ScriptArtifactsChanged Sample(string kind = ScriptArtifactKinds.ApiMethod) =>
|
||||
new(kind, new[] { "M1" }, "Management", DateTimeOffset.UnixEpoch);
|
||||
|
||||
[Fact]
|
||||
public void Publish_delivers_to_all_current_subscribers()
|
||||
{
|
||||
var bus = NewBus();
|
||||
var a = new List<ScriptArtifactsChanged>();
|
||||
var b = new List<ScriptArtifactsChanged>();
|
||||
bus.Subscribe(a.Add);
|
||||
bus.Subscribe(b.Add);
|
||||
|
||||
var n = Sample();
|
||||
bus.Publish(n);
|
||||
|
||||
Assert.Same(n, Assert.Single(a));
|
||||
Assert.Same(n, Assert.Single(b));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Disposing_subscription_stops_delivery()
|
||||
{
|
||||
var bus = NewBus();
|
||||
var received = new List<ScriptArtifactsChanged>();
|
||||
var token = bus.Subscribe(received.Add);
|
||||
|
||||
bus.Publish(Sample());
|
||||
token.Dispose();
|
||||
bus.Publish(Sample());
|
||||
|
||||
Assert.Single(received); // only the pre-dispose publish landed.
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Dispose_is_idempotent()
|
||||
{
|
||||
var bus = NewBus();
|
||||
var received = new List<ScriptArtifactsChanged>();
|
||||
var token = bus.Subscribe(received.Add);
|
||||
|
||||
token.Dispose();
|
||||
token.Dispose(); // must not throw nor corrupt the handler list.
|
||||
|
||||
bus.Publish(Sample());
|
||||
Assert.Empty(received);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_faulting_subscriber_does_not_prevent_delivery_to_others_nor_throw()
|
||||
{
|
||||
var bus = NewBus();
|
||||
var good = new List<ScriptArtifactsChanged>();
|
||||
bus.Subscribe(_ => throw new InvalidOperationException("bad consumer"));
|
||||
bus.Subscribe(good.Add);
|
||||
|
||||
// Publish must not propagate the subscriber fault...
|
||||
var ex = Record.Exception(() => bus.Publish(Sample()));
|
||||
Assert.Null(ex);
|
||||
// ...and the well-behaved subscriber still receives it.
|
||||
Assert.Single(good);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Publish_with_no_subscribers_is_a_noop()
|
||||
{
|
||||
var bus = NewBus();
|
||||
Assert.Null(Record.Exception(() => bus.Publish(Sample())));
|
||||
}
|
||||
}
|
||||
+132
@@ -5,12 +5,15 @@ using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.ExternalSystems;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.InboundApi;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Scripts;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Scripting;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Scripting;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;
|
||||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
|
||||
@@ -42,6 +45,7 @@ namespace ZB.MOM.WW.ScadaBridge.Transport.IntegrationTests.Import;
|
||||
public sealed class BundleImporterApplyTests : IDisposable
|
||||
{
|
||||
private readonly ServiceProvider _provider;
|
||||
private readonly RecordingScriptArtifactChangeBus _artifactBus = new();
|
||||
|
||||
public BundleImporterApplyTests()
|
||||
{
|
||||
@@ -89,6 +93,10 @@ public sealed class BundleImporterApplyTests : IDisposable
|
||||
services.AddScoped<IAuditCorrelationContext, AuditCorrelationContext>();
|
||||
services.AddScoped<IAuditService, AuditService>();
|
||||
services.AddTransport();
|
||||
// #05-T14: a recording bus so the post-commit ScriptArtifactsChanged
|
||||
// publish is observable. BundleImporter takes IScriptArtifactChangeBus?
|
||||
// optionally, so this registration flows in via DI activation.
|
||||
services.AddSingleton<IScriptArtifactChangeBus>(_artifactBus);
|
||||
// #16 (M8 D2): the stale-instance probe is implemented in DeploymentManager
|
||||
// and the flatten/hash primitives in TemplateEngine — register both so
|
||||
// BundleImporter resolves a real IStaleInstanceProbe and ApplyAsync can
|
||||
@@ -1324,4 +1332,128 @@ public sealed class BundleImporterApplyTests : IDisposable
|
||||
Assert.Contains(deployedInstanceId, result.StaleInstanceIds);
|
||||
Assert.DoesNotContain(notDeployedInstanceId, result.StaleInstanceIds);
|
||||
}
|
||||
|
||||
// ============ #05-T14: post-commit ScriptArtifactsChanged publish ============
|
||||
|
||||
[Fact]
|
||||
public async Task ApplyAsync_publishes_ScriptArtifactsChanged_per_kind_after_commit()
|
||||
{
|
||||
// Seed a Template, a SharedScript and an ApiMethod; Overwrite all three.
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
||||
var t = new Template("Pump") { Description = "tpl" };
|
||||
t.Scripts.Add(new TemplateScript("init", "return 1;"));
|
||||
ctx.Templates.Add(t);
|
||||
ctx.SharedScripts.Add(new SharedScript("HelperFn", "return 1;"));
|
||||
ctx.ApiMethods.Add(new ApiMethod("DoThing", "return 1;") { TimeoutSeconds = 30 });
|
||||
await ctx.SaveChangesAsync();
|
||||
}
|
||||
|
||||
Guid sessionId;
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var exporter = scope.ServiceProvider.GetRequiredService<IBundleExporter>();
|
||||
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
||||
var selection = new ExportSelection(
|
||||
TemplateIds: await ctx.Templates.Select(t => t.Id).ToListAsync(),
|
||||
SharedScriptIds: await ctx.SharedScripts.Select(s => s.Id).ToListAsync(),
|
||||
ExternalSystemIds: Array.Empty<int>(),
|
||||
DatabaseConnectionIds: Array.Empty<int>(),
|
||||
NotificationListIds: Array.Empty<int>(),
|
||||
SmtpConfigurationIds: Array.Empty<int>(),
|
||||
ApiMethodIds: await ctx.ApiMethods.Select(m => m.Id).ToListAsync(),
|
||||
IncludeDependencies: false);
|
||||
var stream = await exporter.ExportAsync(selection, user: "alice", sourceEnvironment: "dev",
|
||||
passphrase: null, cancellationToken: CancellationToken.None);
|
||||
using var ms = new MemoryStream();
|
||||
await stream.CopyToAsync(ms);
|
||||
ms.Position = 0;
|
||||
var importer = scope.ServiceProvider.GetRequiredService<IBundleImporter>();
|
||||
var session = await importer.LoadAsync(ms, passphrase: null);
|
||||
sessionId = session.SessionId;
|
||||
}
|
||||
|
||||
// Act — Overwrite all three (they still exist in the target).
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var importer = scope.ServiceProvider.GetRequiredService<IBundleImporter>();
|
||||
await importer.ApplyAsync(sessionId,
|
||||
new List<ImportResolution>
|
||||
{
|
||||
new("Template", "Pump", ResolutionAction.Overwrite, null),
|
||||
new("SharedScript", "HelperFn", ResolutionAction.Overwrite, null),
|
||||
new("ApiMethod", "DoThing", ResolutionAction.Overwrite, null),
|
||||
},
|
||||
user: "bob");
|
||||
}
|
||||
|
||||
// Assert — exactly one notification per kind, post-commit, with the names.
|
||||
var byKind = _artifactBus.Received.ToDictionary(n => n.ArtifactKind);
|
||||
Assert.Equal(3, _artifactBus.Received.Count);
|
||||
Assert.Equal(new[] { "DoThing" }, byKind[ScriptArtifactKinds.ApiMethod].Names);
|
||||
Assert.Equal(new[] { "HelperFn" }, byKind[ScriptArtifactKinds.SharedScript].Names);
|
||||
Assert.Equal(new[] { "Pump" }, byKind[ScriptArtifactKinds.Template].Names);
|
||||
Assert.All(_artifactBus.Received, n => Assert.Equal("BundleImport", n.Source));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ApplyAsync_failed_apply_publishes_nothing()
|
||||
{
|
||||
// A template whose script calls an unresolved helper fails semantic
|
||||
// validation BEFORE commit, so nothing is published.
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
||||
var t = new Template("Bad");
|
||||
t.Scripts.Add(new TemplateScript("init", "var x = UnknownHelper();"));
|
||||
ctx.Templates.Add(t);
|
||||
await ctx.SaveChangesAsync();
|
||||
}
|
||||
|
||||
Guid sessionId;
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var exporter = scope.ServiceProvider.GetRequiredService<IBundleExporter>();
|
||||
var ctx = scope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>();
|
||||
var selection = new ExportSelection(
|
||||
TemplateIds: await ctx.Templates.Select(t => t.Id).ToListAsync(),
|
||||
SharedScriptIds: Array.Empty<int>(),
|
||||
ExternalSystemIds: Array.Empty<int>(),
|
||||
DatabaseConnectionIds: Array.Empty<int>(),
|
||||
NotificationListIds: Array.Empty<int>(),
|
||||
SmtpConfigurationIds: Array.Empty<int>(),
|
||||
ApiMethodIds: Array.Empty<int>(),
|
||||
IncludeDependencies: false);
|
||||
var stream = await exporter.ExportAsync(selection, user: "alice", sourceEnvironment: "dev",
|
||||
passphrase: null, cancellationToken: CancellationToken.None);
|
||||
using var ms = new MemoryStream();
|
||||
await stream.CopyToAsync(ms);
|
||||
ms.Position = 0;
|
||||
var importer = scope.ServiceProvider.GetRequiredService<IBundleImporter>();
|
||||
sessionId = (await importer.LoadAsync(ms, passphrase: null)).SessionId;
|
||||
}
|
||||
|
||||
await using (var scope = _provider.CreateAsyncScope())
|
||||
{
|
||||
var importer = scope.ServiceProvider.GetRequiredService<IBundleImporter>();
|
||||
await Assert.ThrowsAsync<SemanticValidationException>(() =>
|
||||
importer.ApplyAsync(sessionId,
|
||||
new List<ImportResolution> { new("Template", "Bad", ResolutionAction.Add, null) },
|
||||
user: "bob"));
|
||||
}
|
||||
|
||||
Assert.Empty(_artifactBus.Received);
|
||||
}
|
||||
|
||||
/// <summary>Records every published notification for assertion.</summary>
|
||||
private sealed class RecordingScriptArtifactChangeBus : IScriptArtifactChangeBus
|
||||
{
|
||||
public List<ScriptArtifactsChanged> Received { get; } = new();
|
||||
|
||||
public void Publish(ScriptArtifactsChanged notification) => Received.Add(notification);
|
||||
|
||||
public IDisposable Subscribe(Action<ScriptArtifactsChanged> handler) =>
|
||||
throw new NotSupportedException("The recording bus does not support subscriptions.");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user