lmxopcua/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineScriptRunnerTests.cs

using System.Collections.Concurrent;
using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;

namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests;

/// <summary>
///     lmxopcua#780 — the scripted-alarm seam on an <see cref="IScriptRunner"/> (design §4.5, §9). With a runner the engine
///     submits a pass's whole dirty set as ONE batch and applies the results in the serial order, so the Part 9 sequence
///     is the in-process one; and a predicate that did not complete freezes its condition with an annotation that says
///     whose fault it was: <c>BadInternalError</c> for the script (timeout, kill, quarantine), <c>BadResourceUnavailable</c>
///     for the runner (worker restarted, none ready) — re-evaluated by itself when the runner is back. A dead worker never
///     clears an alarm.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ScriptedAlarmEngineScriptRunnerTests
{
    private const uint StatusBadInternalError = 0x80020000u;
    private const uint StatusBadResourceUnavailable = 0x80040000u;

    /// <summary>Answers each predicate through <see cref="Answer"/> and records every batch.</summary>
    private sealed class FakeRunner : IScriptRunner
    {
        public Func<ScriptInvocation, ScriptRunOutcome> Answer { get; set; } = _ => ScriptRunOutcome.Ok(false);

        public ConcurrentQueue<IReadOnlyList<string>> Batches { get; } = new();

        public List<ScriptContextKind> Resets { get; } = [];

        public int SingleRuns;

        public Task<ScriptRunOutcome> RunAsync(ScriptInvocation invocation, CancellationToken ct = default)
        {
            Interlocked.Increment(ref SingleRuns);
            return Task.FromResult(Answer(invocation));
        }

        public IReadOnlyList<Task<ScriptRunOutcome>> RunBatch(IReadOnlyList<ScriptInvocation> invocations, CancellationToken ct = default)
        {
            invocations.ShouldAllBe(i => i.Kind == ScriptContextKind.AlarmPredicate);
            Batches.Enqueue(invocations.Select(i => i.Key).ToArray());
            return invocations.Select(i => Task.FromResult(Answer(i))).ToArray();
        }

        public void Reset(ScriptContextKind kind) => Resets.Add(kind);

        public IReadOnlyCollection<string> QuarantinedKeys(ScriptContextKind kind) => [];
    }

    private static ScriptedAlarmDefinition Alarm(string id, string predicate) =>
        new(AlarmId: id, EquipmentPath: "Plant/Line1/Reactor", AlarmName: id, Kind: AlarmKind.AlarmCondition,
            Severity: AlarmSeverity.High, MessageTemplate: "condition", PredicateScriptSource: predicate);

    private static ScriptedAlarmEngine Build(FakeUpstream up, IScriptRunner? runner)
    {
        var logger = new LoggerConfiguration().CreateLogger();
        return new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger,
            scriptTimeout: TimeSpan.FromMilliseconds(250), scriptRunner: runner);
    }

    /// <summary>The fake's stand-in for the five predicates: <c>Temp &gt; threshold</c>, threshold from the alarm id.</summary>
    private static ScriptRunOutcome Threshold(ScriptInvocation invocation)
        => ScriptRunOutcome.Ok((int)invocation.Inputs["Temp"].Value! > int.Parse(invocation.Key[1..]));

    private static readonly ScriptedAlarmDefinition[] Five =
        [.. new[] { 10, 20, 30, 40, 50 }.Select(t => Alarm($"a{t}", $"return (int)ctx.GetTag(\"Temp\").Value > {t};"))];

    private static async Task WaitFor(Func<bool> probe, string what)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
        while (DateTime.UtcNow < deadline)
        {
            if (probe()) return;
            await Task.Delay(10);
        }
        throw new TimeoutException(what);
    }

    [Fact]
    public async Task A_dirty_set_is_one_batch_and_the_transitions_come_out_exactly_as_in_process()
    {
        var ct = TestContext.Current.CancellationToken;
        var runner = new FakeRunner { Answer = Threshold };
        var outOfProcess = new FakeUpstream();
        var inProcess = new FakeUpstream();
        outOfProcess.Set("Temp", 0);
        inProcess.Set("Temp", 0);
        using var oop = Build(outOfProcess, runner);
        using var ip = Build(inProcess, null);
        var oopEvents = new ConcurrentQueue<(string, EmissionKind)>();
        var ipEvents = new ConcurrentQueue<(string, EmissionKind)>();
        oop.OnEvent += (_, e) => oopEvents.Enqueue((e.AlarmId, e.Emission));
        ip.OnEvent += (_, e) => ipEvents.Enqueue((e.AlarmId, e.Emission));

        await oop.LoadAsync(Five, ct);
        await ip.LoadAsync(Five, ct);
        runner.Batches.Count.ShouldBe(1, "startup recovery evaluates every alarm in one batch");
        runner.Resets.ShouldBe([ScriptContextKind.AlarmPredicate], "a load is the seam's deploy boundary");

        foreach (var temp in new[] { 35, 60, 15, 0 })
        {
            var before = runner.Batches.Count;
            outOfProcess.Push("Temp", temp);
            inProcess.Push("Temp", temp);
            await WaitFor(() => runner.Batches.Count > before && oop.GetAllStates().Count == 5, "the out-of-process pass");
            await WaitFor(() => ipEvents.Count == oopEvents.Count, "the in-process pass to catch up");
        }

        runner.Batches.Skip(1).ShouldAllBe(b => b.Count == 5, "each push re-evaluates the whole dirty set in one batch");
        runner.SingleRuns.ShouldBe(0);
        oopEvents.ToArray().ShouldBe(ipEvents.ToArray(), "the same Part 9 sequence, in the same order");
        oopEvents.Count(e => e.Item2 == EmissionKind.Activated).ShouldBe(5);
        oop.GetAllStates().Select(s => (s.AlarmId, s.Active)).OrderBy(x => x.AlarmId)
            .ShouldBe(ip.GetAllStates().Select(s => (s.AlarmId, s.Active)).OrderBy(x => x.AlarmId));
    }

    [Fact]
    public async Task A_predicate_that_did_not_complete_is_frozen_and_annotated_by_whose_fault_it_was()
    {
        var ct = TestContext.Current.CancellationToken;
        var runner = new FakeRunner { Answer = _ => ScriptRunOutcome.Ok(true) };
        var up = new FakeUpstream();
        up.Set("Temp", 1);
        using var eng = Build(up, runner);
        var events = new ConcurrentQueue<ScriptedAlarmEvent>();
        eng.OnEvent += (_, e) => events.Enqueue(e);
        await eng.LoadAsync([Alarm("a", "return (int)ctx.GetTag(\"Temp\").Value > 0;")], ct);
        eng.GetState("a")!.Active.ShouldBe(AlarmActiveState.Active);
        events.Clear();

        // The runner, not the script: frozen ACTIVE (a dead worker never clears an alarm), BadResourceUnavailable.
        var recovered = new TaskCompletionSource();
        runner.Answer = _ => ScriptRunOutcome.Unavailable(ScriptRunOutcome.WorkerRestartedReason) with { Recovery = recovered.Task };
        up.Push("Temp", 2);
        await WaitFor(() => !events.IsEmpty, "the runner-failure annotation");
        var annotation = events.Single();
        annotation.Emission.ShouldBe(EmissionKind.QualityChanged);
        annotation.WorstInputStatusCode.ShouldBe(StatusBadResourceUnavailable);
        eng.GetState("a")!.Active.ShouldBe(AlarmActiveState.Active, "not cleared because the worker died");

        // The worker is back: re-evaluated by itself, against the current inputs, with no new push.
        events.Clear();
        runner.Answer = _ => ScriptRunOutcome.Ok(false);
        var batches = runner.Batches.Count;
        recovered.SetResult();
        await WaitFor(() => runner.Batches.Count > batches && eng.GetState("a")!.Active == AlarmActiveState.Inactive, "the recovery re-evaluation");
        await WaitFor(() => events.Any(e => e.Emission == EmissionKind.Cleared), "the clear");

        // The completed run restored the quality on the transition itself (no separate annotation).
        events.Single().WorstInputStatusCode.ShouldBe(0u);

        // The script: a timeout (or kill, or quarantine) freezes it with BadInternalError, exactly as in process.
        events.Clear();
        runner.Answer = _ => ScriptRunOutcome.Failed("script timed out after 0.3s");
        up.Push("Temp", 3);
        await WaitFor(() => !events.IsEmpty, "the script-failure annotation");
        events.Single().WorstInputStatusCode.ShouldBe(StatusBadInternalError);
        eng.GetState("a")!.Active.ShouldBe(AlarmActiveState.Inactive);

        // A predicate that threw keeps its state with no annotation (as in process: its run completed).
        events.Clear();
        runner.Answer = _ => ScriptRunOutcome.Ok(true);
        up.Push("Temp", 4);
        await WaitFor(() => eng.GetState("a")!.Active == AlarmActiveState.Active, "a completed run");
        events.Clear();
        runner.Answer = _ => ScriptRunOutcome.Failed("script threw: boom");
        var before = runner.Batches.Count;
        up.Push("Temp", 5);
        await WaitFor(() => runner.Batches.Count > before, "the throwing pass");
        await Task.Delay(100, ct);
        events.ShouldBeEmpty();
        eng.GetState("a")!.Active.ShouldBe(AlarmActiveState.Active);
    }

    [Fact]
    public async Task Alarms_whose_inputs_are_not_ready_never_reach_the_runner()
    {
        var ct = TestContext.Current.CancellationToken;
        var runner = new FakeRunner { Answer = _ => ScriptRunOutcome.Ok(true) };
        var up = new FakeUpstream();
        up.Set("A", 1);
        up.Set("B", 0, 0x80000000u); // Bad
        using var eng = Build(up, runner);
        await eng.LoadAsync([Alarm("good", "return (int)ctx.GetTag(\"A\").Value > 0;"), Alarm("bad", "return (int)ctx.GetTag(\"B\").Value > 0;")], ct);
        runner.Batches.Single().ShouldBe(["good"], "a Bad input holds the predicate, as in process");
        eng.GetState("bad")!.Active.ShouldBe(AlarmActiveState.Inactive);
        eng.GetState("good")!.Active.ShouldBe(AlarmActiveState.Active);
    }
}