/* ======================================================================== * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Collections.Generic; using System.Diagnostics; using System.Text; using System.Threading; using System.Threading.Tasks; using NUnit.Framework; using Opc.Ua; using Opc.Ua.AI; using Opc.Ua.AI.Inference; using Opc.Ua.AI.Server; using ObjectIds = Opc.Ua.ObjectIds; namespace Opc.Ua.AI.Tests { /// /// Verifies the asynchronous path and the model source. /// [TestFixture] [Category("AIModelManagement")] [SetCulture("en-us")] [SetUICulture("en-us")] public sealed class JobAndSourceTests { [Test] public async Task AnAsynchronousInferenceReturnsAJobThatLaterCarriesTheResultAsync() { var backend = new BlockingInferenceBackend(); using AINodeManager nm = await AIServerTestHarness .CreateAsync( new InferenceBackends(backend), new AIOptions { EnableFallback = false, AsyncInferenceDelay = TimeSpan.Zero }) .ConfigureAwait(false); var deployment = nm.FindPredefinedNode(nm.PrimaryDeploymentId); try { InvokeAsyncMethodStateResult started = await deployment.InvokeAsync!.OnCallAsync!( nm.SystemContext, deployment.InvokeAsync, nm.PrimaryDeploymentId, ByteString.From(Encoding.UTF8.GetBytes("{}")), string.Empty, "application/json", ArrayOf.Empty, CancellationToken.None).ConfigureAwait(false); Assert.That(started.Job, Is.Not.EqualTo(NodeId.Null)); var job = nm.FindPredefinedNode(started.Job); await backend.Entered.WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false); // Running before Halted: the caller has a NodeId to watch, and the // result belongs to the job rather than to the call that asked for it. Assert.That( job.CurrentState!.Id!.Value, Is.EqualTo(Opc.Ua.ObjectIds.ProgramStateMachineType_Running)); backend.Release(); await WaitForHaltedAsync(job).ConfigureAwait(false); Assert.Multiple(() => { Assert.That(job.ResponsePayload!.Value.Length, Is.GreaterThan(0)); Assert.That(job.ModelUsed!.Value, Is.Not.EqualTo(NodeId.Null)); Assert.That(job.Progress!.Value, Is.EqualTo(100)); Assert.That(job.FinishedAt, Is.Not.Null); }); } finally { backend.Release(); } } [Test] public async Task AFailedJobHaltsAndRecordsWhyAsync() { var primary = new FakeInferenceBackend("primary") { Healthy = false }; using AINodeManager nm = await AIServerTestHarness .CreateAsync( new InferenceBackends(primary), new AIOptions { EnableFallback = false, AsyncInferenceDelay = TimeSpan.Zero }) .ConfigureAwait(false); var deployment = nm.FindPredefinedNode(nm.PrimaryDeploymentId); InvokeAsyncMethodStateResult started = await deployment.InvokeAsync!.OnCallAsync!( nm.SystemContext, deployment.InvokeAsync, nm.PrimaryDeploymentId, ByteString.From(Encoding.UTF8.GetBytes("{}")), string.Empty, "application/json", ArrayOf.Empty, CancellationToken.None).ConfigureAwait(false); var job = nm.FindPredefinedNode(started.Job); await WaitForHaltedAsync(job).ConfigureAwait(false); Assert.Multiple(() => { // Halted either way. Whether the inference succeeded is answered by // whether ResponsePayload or LastError is set, not by the state the // program ended in. Assert.That(job.LastError!.Value.Text, Is.Not.Empty); // The member is published either way, so a client can browse to it // and subscribe before the job finishes. On failure it carries a // null ByteString - absent rather than empty, because an empty // answer and no answer are different things and a client deciding // whether to retry needs to tell them apart. Assert.That(job.ResponsePayload, Is.Not.Null); Assert.That(job.ResponsePayload!.Value.IsNull, Is.True); }); } [Test] public async Task AnAsynchronousInferenceForwardsParametersToTheBackendAsync() { var backend = new FakeInferenceBackend("primary"); using AINodeManager nm = await AIServerTestHarness .CreateAsync( new InferenceBackends(backend), new AIOptions { EnableFallback = false, AsyncInferenceDelay = TimeSpan.Zero }) .ConfigureAwait(false); DeploymentState deployment = nm.FindPredefinedNode(nm.PrimaryDeploymentId); ArrayOf parameters = [ new Opc.Ua.KeyValuePair { Key = new QualifiedName("top_p"), Value = Variant.From(0.8) } ]; InvokeAsyncMethodStateResult started = await deployment.InvokeAsync!.OnCallAsync!( nm.SystemContext, deployment.InvokeAsync, nm.PrimaryDeploymentId, ByteString.From(Encoding.UTF8.GetBytes("{}")), string.Empty, "application/json", parameters, CancellationToken.None).ConfigureAwait(false); InferenceJobState job = nm.FindPredefinedNode(started.Job); await WaitForHaltedAsync(job).ConfigureAwait(false); Assert.Multiple(() => { Assert.That(backend.Requests, Has.Count.EqualTo(1)); Assert.That(backend.Requests[0].Parameters["top_p"], Is.EqualTo("0.8")); }); } [Test] public async Task TheSourceReportsWhatItCanReachAsync() { var primary = new FakeInferenceBackend("primary"); primary.Models.Add(new BackendModel { Publisher = "contoso", Name = "weld-inspect", Version = "2.1.0" }); using AINodeManager nm = await AIServerTestHarness .CreateAsync( new InferenceBackends(primary), new AIOptions { EnableFallback = false }) .ConfigureAwait(false); ModelSourceState source = FindSource(nm); TestConnectionMethodStateResult probe = await source.TestConnection!.OnCallAsync!( nm.SystemContext, source.TestConnection, source.NodeId, CancellationToken.None).ConfigureAwait(false); ListModelsMethodStateResult listed = await source.ListModels!.OnCallAsync!( nm.SystemContext, source.ListModels, source.NodeId, string.Empty, 0, ByteString.Empty, CancellationToken.None).ConfigureAwait(false); Assert.Multiple(() => { Assert.That(probe.Reachable, Is.True); Assert.That(source.Reachability!.Value, Is.EqualTo(ReachabilityEnum.Reachable)); // Answered from the source, not from what this Server has already // imported: the question is what COULD be deployed. Assert.That(listed.Models.Count, Is.EqualTo(1)); Assert.That(listed.Models[0].Name, Is.EqualTo("weld-inspect")); }); } [Test] public async Task AnUnreachableSourceSaysSoAsync() { var primary = new FakeInferenceBackend("primary") { Reachable = false }; using AINodeManager nm = await AIServerTestHarness .CreateAsync( new InferenceBackends(primary), new AIOptions { EnableFallback = false }) .ConfigureAwait(false); ModelSourceState source = FindSource(nm); TestConnectionMethodStateResult probe = await source.TestConnection!.OnCallAsync!( nm.SystemContext, source.TestConnection, source.NodeId, CancellationToken.None).ConfigureAwait(false); Assert.Multiple(() => { Assert.That(probe.Reachable, Is.False); Assert.That( source.Reachability!.Value, Is.EqualTo(ReachabilityEnum.Unreachable)); Assert.That(source.ConsecutiveFailures!.Value, Is.EqualTo(1u)); }); } private static Task CreateAsync() { return AIServerTestHarness.CreateAsync( new InferenceBackends(new FakeInferenceBackend("primary")), new AIOptions { EnableFallback = false, AsyncInferenceDelay = TimeSpan.Zero }); } private static ModelSourceState FindSource(AINodeManager nm) { var root = nm.FindPredefinedNode(nm.RootId); var children = new System.Collections.Generic.List(); root.Sources!.GetChildren(nm.SystemContext, children); foreach (BaseInstanceState child in children) { if (child is ModelSourceState source) { return source; } } throw new InvalidOperationException("No model source was published."); } /// /// Waits for the job to leave Running. /// /// /// Polls rather than sleeping a fixed interval, so the test is neither /// flaky on a loaded machine nor slower than it needs to be on an idle one. /// private static async Task WaitForHaltedAsync(InferenceJobState job) { var stopwatch = Stopwatch.StartNew(); while (stopwatch.Elapsed < TimeSpan.FromSeconds(10)) { if (job.CurrentState!.Id!.Value == Opc.Ua.ObjectIds.ProgramStateMachineType_Halted) { return; } await Task.Delay(20).ConfigureAwait(false); } Assert.Fail("The job did not reach Halted."); } private sealed class BlockingInferenceBackend : IInferenceBackend { private readonly FakeInferenceBackend m_inner = new("primary"); private readonly TaskCompletionSource m_entered = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource m_release = new(TaskCreationOptions.RunContinuationsAsynchronously); public InferenceSite Site => m_inner.Site; public Task Entered => m_entered.Task; public void Release() { m_release.TrySetResult(); } public ValueTask> ListModelsAsync( string? filter, uint maxResults, CancellationToken ct) { return m_inner.ListModelsAsync(filter, maxResults, ct); } public async ValueTask InvokeAsync( InferenceRequest request, CancellationToken ct) { m_entered.TrySetResult(); await m_release.Task.WaitAsync(ct).ConfigureAwait(false); return await m_inner.InvokeAsync(request, ct).ConfigureAwait(false); } public ValueTask ProbeAsync(CancellationToken ct) { return m_inner.ProbeAsync(ct); } } } }