Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8828e999c4 | |||
| 4f7a07fbd3 | |||
| da2e441bb9 | |||
| c8e3e279b4 | |||
| 468f13cb56 | |||
| f71fe13826 | |||
| 6f0674296b | |||
| 19bc13bf6a | |||
| eecf301200 |
@@ -41,6 +41,8 @@ jobs:
|
||||
# #606: regenerated review summary must remain in sync with module findings.
|
||||
- name: code-reviews README drift check
|
||||
run: python3 code-reviews/regen-readme.py --check
|
||||
- name: TRX skip guard fixtures
|
||||
run: python3 scripts/ci/test-assert-not-all-skipped.py
|
||||
# #707: bound project workers and avoid concurrent NuGet restore work.
|
||||
- name: dotnet restore
|
||||
run: dotnet restore ZB.MOM.WW.OtOpcUa.slnx --disable-parallel -m:2
|
||||
|
||||
@@ -28,3 +28,26 @@ AKKA_TEST_TIMEFACTOR=3 dotnet test ZB.MOM.WW.OtOpcUa.slnx \
|
||||
--logger trx --results-directory artifacts/trx
|
||||
MIN_EXECUTED=12000 bash scripts/ci/assert-not-all-skipped.sh artifacts/trx/*.trx
|
||||
```
|
||||
|
||||
The guard uses the existing TRX `Counters/@executed` value for the aggregate
|
||||
floor. Its `skipped` diagnostic counts individual `Results/UnitTestResult`
|
||||
records whose outcome is `NotExecuted`, using the document's XML namespace.
|
||||
Some VSTest reports set summary `notExecuted` to zero even when these records
|
||||
are present. Skip records are counted per file, including repeated runs; they
|
||||
are not a count of unique test methods and do not affect the executed floor.
|
||||
For malformed XML, the diagnostic retains the old summary-counter fallback;
|
||||
this guard does not add XML validation to its existing gate policy.
|
||||
|
||||
Run the deterministic script fixtures without building or running the solution:
|
||||
|
||||
```bash
|
||||
python3 scripts/ci/test-assert-not-all-skipped.py
|
||||
```
|
||||
|
||||
Verified hosted baseline: [Gitea run 975](https://gitea.dohertylan.com/dohertj2/lmxopcua/actions/runs/975)
|
||||
completed successfully on 2026-10-13 for commit
|
||||
`76bdca90e94e814e06ef10b78271e2b492b2ba32` (merged by PR #711).
|
||||
Restore, Release build, README drift check, the serialized whole-solution test
|
||||
step, the 12,000-test floor and TRX upload all passed. The floor counted
|
||||
14,630 executed tests across 62 TRX files. This baseline predates the individual
|
||||
skip-reporting correction; its printed zero skip diagnostic is not skip evidence.
|
||||
|
||||
@@ -33,12 +33,31 @@ for f in "$@"; do
|
||||
exit 2
|
||||
fi
|
||||
seen=$((seen + 1))
|
||||
# The trx <Counters .../> element in <ResultSummary> carries executed + notExecuted counts.
|
||||
# Keep the executed-counter floor unchanged; notExecuted is only a fallback
|
||||
# for malformed XML that this guard historically accepted.
|
||||
counters="$(grep -oE '<Counters[^>]*/?>' "$f" | head -1)"
|
||||
ex="$(printf '%s' "$counters" | grep -oE 'executed="[0-9]+"' | grep -oE '[0-9]+' || true)"
|
||||
ne="$(printf '%s' "$counters" | grep -oE 'notExecuted="[0-9]+"' | grep -oE '[0-9]+' || true)"
|
||||
ex="${ex:-0}"
|
||||
ne="${ne:-0}"
|
||||
# VSTest can emit notExecuted="0" despite individual NotExecuted results.
|
||||
# Count direct Results/UnitTestResult records in the document's namespace,
|
||||
# excluding foreign elements, nested records and escaped diagnostic text.
|
||||
ne="$(python3 - "$f" "$ne" <<'PY'
|
||||
import sys
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
try:
|
||||
root = ET.parse(sys.argv[1]).getroot()
|
||||
except ET.ParseError:
|
||||
# Skip reporting must not introduce a new malformed-XML gate failure.
|
||||
print(sys.argv[2])
|
||||
else:
|
||||
namespace = root.tag.split("}", 1)[0] + "}" if root.tag.startswith("{") else ""
|
||||
results = root.findall(f"{namespace}Results/{namespace}UnitTestResult")
|
||||
print(sum(result.get("outcome") == "NotExecuted" for result in results))
|
||||
PY
|
||||
)"
|
||||
total_executed=$((total_executed + ex))
|
||||
total_skipped=$((total_skipped + ne))
|
||||
echo " $(basename "$f"): executed=${ex} skipped=${ne}"
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Deterministic subprocess checks for the TRX executed-floor guard."""
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
SCRIPT = Path(__file__).with_name("assert-not-all-skipped.sh")
|
||||
NS = "http://microsoft.com/schemas/VisualStudio/TeamTest/2010"
|
||||
|
||||
|
||||
def trx(executed, outcomes, summary_skipped=0, namespace=NS, extra=""):
|
||||
declaration = f' xmlns="{namespace}"' if namespace else ""
|
||||
results = "".join(f'<UnitTestResult outcome="{outcome}"/>' for outcome in outcomes)
|
||||
return (f'<TestRun{declaration}><Results>{results}{extra}</Results>'
|
||||
f'<ResultSummary><Counters total="{len(outcomes)}" executed="{executed}" '
|
||||
f'notExecuted="{summary_skipped}"/></ResultSummary></TestRun>')
|
||||
|
||||
|
||||
class SkipGuardTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.directory = tempfile.TemporaryDirectory(prefix="trx-skip-fixtures-")
|
||||
self.addCleanup(self.directory.cleanup)
|
||||
self.root = Path(self.directory.name)
|
||||
|
||||
def write(self, name, content):
|
||||
path = self.root / name
|
||||
path.write_text(content, encoding="utf-8")
|
||||
return path
|
||||
|
||||
def run_guard(self, *paths, minimum=None):
|
||||
env = os.environ.copy()
|
||||
env.pop("MIN_EXECUTED", None)
|
||||
if minimum is not None:
|
||||
env["MIN_EXECUTED"] = str(minimum)
|
||||
return subprocess.run(["bash", str(SCRIPT), *map(str, paths)], env=env,
|
||||
text=True, capture_output=True, check=False)
|
||||
|
||||
def assert_report(self, result, status, executed, skipped, files=1):
|
||||
self.assertEqual(status, result.returncode, result.stdout + result.stderr)
|
||||
self.assertIn(f"files={files} executed={executed} skipped={skipped} ", result.stdout)
|
||||
|
||||
def test_mixed_results_ignore_zero_summary_skip_counter(self):
|
||||
# A summary-only skip count misses the three actual NotExecuted records.
|
||||
path = self.write("mixed.trx", trx(827, ["Passed"] * 827 + ["NotExecuted"] * 3))
|
||||
result = self.run_guard(path)
|
||||
self.assert_report(result, 0, 827, 3)
|
||||
self.assertIn("mixed.trx: executed=827 skipped=3", result.stdout)
|
||||
|
||||
def test_all_skipped_fails_default_floor_but_report_only_passes(self):
|
||||
path = self.write("all-skipped.trx", trx(0, ["NotExecuted"] * 188))
|
||||
self.assert_report(self.run_guard(path), 1, 0, 188)
|
||||
self.assert_report(self.run_guard(path, minimum=0), 0, 0, 188)
|
||||
|
||||
def test_zero_tests_fails_default_floor_but_report_only_passes(self):
|
||||
path = self.write("zero.trx", trx(0, []))
|
||||
self.assert_report(self.run_guard(path), 1, 0, 0)
|
||||
self.assert_report(self.run_guard(path, minimum=0), 0, 0, 0)
|
||||
|
||||
def test_agreeing_summary_and_outcomes_are_not_double_counted(self):
|
||||
path = self.write("agreeing.trx", trx(1, ["Passed", "NotExecuted"], 1))
|
||||
self.assert_report(self.run_guard(path), 0, 1, 1)
|
||||
|
||||
def test_outcomes_are_authoritative_when_summary_overstates_skips(self):
|
||||
path = self.write("overstated.trx", trx(1, ["Passed", "NotExecuted"], 9))
|
||||
self.assert_report(self.run_guard(path), 0, 1, 1)
|
||||
|
||||
def test_namespace_path_and_exact_outcome_exclude_decoys(self):
|
||||
extra = ('<Other><UnitTestResult outcome="NotExecuted"/></Other>'
|
||||
'<foreign:UnitTestResult xmlns:foreign="urn:other" outcome="NotExecuted"/>'
|
||||
'<Output><Message><UnitTestResult outcome="NotExecuted"/></Message></Output>')
|
||||
content = trx(2, ["Passed", "Failed", "NotExecuted", "notexecuted"], extra=extra)
|
||||
# Prefixes on Results/UnitTestResult must work just like the default namespace.
|
||||
content = content.replace('<TestRun ', f'<TestRun xmlns:t="{NS}" ')
|
||||
content = content.replace('<Results>', '<t:Results>').replace('</Results>', '</t:Results>')
|
||||
content = content.replace('<UnitTestResult ', '<t:UnitTestResult ')
|
||||
path = self.write("namespaced.trx", content)
|
||||
self.assert_report(self.run_guard(path), 0, 2, 1)
|
||||
|
||||
def test_unnamespaced_results_remain_supported(self):
|
||||
path = self.write("legacy.trx", trx(1, ["Passed", "NotExecuted"], namespace=""))
|
||||
self.assert_report(self.run_guard(path), 0, 1, 1)
|
||||
|
||||
def test_executed_counter_controls_aggregate_floor_at_boundary(self):
|
||||
# Counters retain authority even when result records disagree with executed.
|
||||
mixed = self.write("floor.trx", trx(12000, ["Passed", "NotExecuted"]))
|
||||
skipped = self.write("skipped.trx", trx(0, ["NotExecuted"]))
|
||||
self.assert_report(self.run_guard(mixed, skipped, minimum=12000), 0, 12000, 2, 2)
|
||||
self.assert_report(self.run_guard(mixed, skipped, minimum=12001), 1, 12000, 2, 2)
|
||||
|
||||
def test_missing_arguments_and_file_keep_exit_two(self):
|
||||
self.assertEqual(2, self.run_guard().returncode)
|
||||
self.assertEqual(2, self.run_guard(self.root / "missing.trx").returncode)
|
||||
|
||||
def test_missing_counters_keeps_existing_exit_one(self):
|
||||
path = self.write("no-counters.trx", '<TestRun><Results/></TestRun>')
|
||||
self.assertEqual(1, self.run_guard(path, minimum=0).returncode)
|
||||
|
||||
def test_malformed_xml_keeps_summary_fallback_and_gate_status(self):
|
||||
path = self.write("malformed.trx", '<TestRun><Counters executed="2" notExecuted="3"/>')
|
||||
self.assert_report(self.run_guard(path, minimum=2), 0, 2, 3)
|
||||
self.assert_report(self.run_guard(path, minimum=3), 1, 2, 3)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -1,21 +1,38 @@
|
||||
namespace ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
|
||||
/// <summary>
|
||||
/// Late-binding adapter that holds an inner <see cref="IServiceLevelPublisher"/> reference
|
||||
/// swappable at runtime. Mirrors <see cref="DeferredAddressSpaceSink"/>: Akka actors resolve
|
||||
/// the publisher at DI time, but the production <c>SdkServiceLevelPublisher</c> only exists
|
||||
/// after <c>StandardServer.Start</c>. The Host's hosted service swaps the inner once the SDK
|
||||
/// is up; until then writes route through <see cref="NullServiceLevelPublisher"/>.
|
||||
/// Late-binding adapter that accepts and retains the latest requested ServiceLevel until
|
||||
/// the SDK publisher is available. Binding synchronously replays that request, including zero.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Publishing and binding are serialized through the synchronous inner call. Inner publishers
|
||||
/// must not call back into this adapter. Acceptance while unbound is not an SDK write confirmation.
|
||||
/// A failed call retains both the requested level and the installed publisher for later recovery.
|
||||
/// </remarks>
|
||||
public sealed class DeferredServiceLevelPublisher : IServiceLevelPublisher
|
||||
{
|
||||
private volatile IServiceLevelPublisher _inner = NullServiceLevelPublisher.Instance;
|
||||
private readonly object _gate = new();
|
||||
private IServiceLevelPublisher? _inner;
|
||||
private byte? _latest;
|
||||
|
||||
/// <summary>Swap the underlying publisher. Pass null to revert to the Null no-op.</summary>
|
||||
/// <param name="inner">The publisher implementation to use, or null to use the null publisher.</param>
|
||||
public void SetInner(IServiceLevelPublisher? inner) =>
|
||||
_inner = inner ?? NullServiceLevelPublisher.Instance;
|
||||
/// <summary>Bind and replay the latest request, or unbind while retaining that request.</summary>
|
||||
/// <param name="inner">The publisher implementation to use, or null to unbind.</param>
|
||||
public void SetInner(IServiceLevelPublisher? inner)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_inner = inner;
|
||||
if (_latest is { } latest) _inner?.Publish(latest);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Publish(byte serviceLevel) => _inner.Publish(serviceLevel);
|
||||
public void Publish(byte serviceLevel)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_latest = serviceLevel;
|
||||
_inner?.Publish(serviceLevel);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
|
||||
/// <summary>Coordinates endpoint cleanup after ServiceLevel binding fails during startup.</summary>
|
||||
internal static class FailedServiceLevelBindingCleanup
|
||||
{
|
||||
/// <summary>Stops and disposes the endpoint, returning whether ownership can be released.</summary>
|
||||
/// <param name="stop">The SDK stop operation.</param>
|
||||
/// <param name="dispose">The application-host disposal operation.</param>
|
||||
/// <param name="logger">The cleanup logger.</param>
|
||||
/// <returns>Whether cleanup completed successfully.</returns>
|
||||
internal static async Task<bool> StopAndDisposeAsync(
|
||||
Func<Task> stop, Func<ValueTask> dispose, ILogger logger)
|
||||
{
|
||||
var completed = true;
|
||||
try
|
||||
{
|
||||
await stop().ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
completed = false;
|
||||
logger.LogError(ex, "OtOpcUaServerHostedService: SDK stop failed after binding failure; retaining endpoint ownership for shutdown retry");
|
||||
}
|
||||
// Attempt application cleanup even when stopping faults; keep ownership on either failure.
|
||||
try
|
||||
{
|
||||
await dispose().ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
completed = false;
|
||||
logger.LogError(ex, "OtOpcUaServerHostedService: endpoint disposal failed after binding failure; retaining endpoint ownership for shutdown retry");
|
||||
}
|
||||
return completed;
|
||||
}
|
||||
}
|
||||
@@ -339,11 +339,36 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_server.NodeManager.Clock = historyUpdate.Clock;
|
||||
|
||||
// ServiceLevel publisher needs IServerInternal — only available after Start.
|
||||
if (_server.CurrentInstance is { } serverInternal)
|
||||
try
|
||||
{
|
||||
_deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
|
||||
serverInternal,
|
||||
_loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
|
||||
if (_server.CurrentInstance is { } serverInternal)
|
||||
{
|
||||
_deferredServiceLevel.SetInner(new SdkServiceLevelPublisher(
|
||||
serverInternal,
|
||||
_loggerFactory.CreateLogger<SdkServiceLevelPublisher>()));
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex,
|
||||
"OtOpcUaServerHostedService: ServiceLevel binding failed; stopping the OPC UA endpoint while the admin host continues");
|
||||
try
|
||||
{
|
||||
await StopAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception cleanupError)
|
||||
{
|
||||
_logger.LogError(cleanupError, "OtOpcUaServerHostedService: failed to detach OPC UA bindings after startup failure");
|
||||
}
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
async () => await _server.StopAsync(CancellationToken.None).ConfigureAwait(false),
|
||||
() => _appHost.DisposeAsync(), _logger).ConfigureAwait(false);
|
||||
if (released)
|
||||
{
|
||||
_appHost = null;
|
||||
_server = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// arch-review 03/S19 — LAST, after the node manager is known non-null and after the ACL gate is
|
||||
|
||||
@@ -558,10 +558,15 @@ public sealed class OpcUaApplicationHost : IAsyncDisposable
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
|
||||
try
|
||||
{
|
||||
try { sessionManager.ImpersonateUser -= _impersonateHandler; }
|
||||
catch (Exception ex) { _logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw"); }
|
||||
// An explicit SDK stop may already have halted CurrentInstance; still continue disposal.
|
||||
if (_impersonateHandler is not null && _server?.CurrentInstance?.SessionManager is { } sessionManager)
|
||||
sessionManager.ImpersonateUser -= _impersonateHandler;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "OpcUaApplicationHost: detaching ImpersonateUser threw");
|
||||
}
|
||||
_impersonateHandler = null;
|
||||
|
||||
|
||||
@@ -38,8 +38,8 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
|
||||
var node = _serverInternal.ServerObject?.ServiceLevel;
|
||||
if (node is null)
|
||||
{
|
||||
_logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable; skipping write");
|
||||
return;
|
||||
_logger.LogWarning("SdkServiceLevelPublisher: ServerObject.ServiceLevel unavailable");
|
||||
throw new InvalidOperationException("Server.ServiceLevel is unavailable.");
|
||||
}
|
||||
|
||||
try
|
||||
@@ -55,6 +55,7 @@ public sealed class SdkServiceLevelPublisher : IServiceLevelPublisher
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "SdkServiceLevelPublisher: write to Server.ServiceLevel threw");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -970,10 +970,10 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
|
||||
// node starting Detached/role-less (first level = 0) would be dedup'd away, leaving the SDK's
|
||||
// built-in default (255 = full service) standing — a degraded node wrongly advertising 255.
|
||||
if (_publishedAtLeastOnce && msg.ServiceLevel == _lastServiceLevel) return;
|
||||
_lastServiceLevel = msg.ServiceLevel;
|
||||
try
|
||||
{
|
||||
_serviceLevel.Publish(msg.ServiceLevel);
|
||||
_lastServiceLevel = msg.ServiceLevel;
|
||||
_publishedAtLeastOnce = true;
|
||||
OtOpcUaTelemetry.ServiceLevelChange.Add(1,
|
||||
new KeyValuePair<string, object?>("level", msg.ServiceLevel));
|
||||
@@ -981,6 +981,8 @@ public sealed class OpcUaPublishActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// A failed write may be partially applied; even a previous successful value must retry.
|
||||
_publishedAtLeastOnce = false;
|
||||
_log.Warning(ex, "OpcUaPublish: ServiceLevel publisher threw at level {Level}", msg.ServiceLevel);
|
||||
}
|
||||
}
|
||||
|
||||
+136
-5
@@ -10,10 +10,10 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Tests.OpcUa;
|
||||
public class DeferredServiceLevelPublisherTests
|
||||
{
|
||||
[Fact]
|
||||
public void Before_SetInner_Publish_is_a_noop()
|
||||
public void Before_SetInner_Publish_accepts_a_cached_request()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
// Must not throw; the NullServiceLevelPublisher absorbs it.
|
||||
// Accept the request before an SDK publisher exists.
|
||||
publisher.Publish(200);
|
||||
}
|
||||
|
||||
@@ -30,7 +30,7 @@ public class DeferredServiceLevelPublisherTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SetInner_null_reverts_to_null_publisher()
|
||||
public void SetInner_null_detaches_the_previous_publisher()
|
||||
{
|
||||
var spy = new SpyPublisher();
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
@@ -39,8 +39,139 @@ public class DeferredServiceLevelPublisherTests
|
||||
|
||||
publisher.Publish(100);
|
||||
|
||||
// The spy should NOT have received the publish after revert.
|
||||
spy.LastPublished.ShouldBe((byte)0, "spy should not have been updated after revert to null");
|
||||
// The previous publisher must not receive an unbound request.
|
||||
spy.LastPublished.ShouldBe((byte)0, "the previous publisher should not be updated after unbinding");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Binding_without_a_request_does_not_invent_a_level()
|
||||
{
|
||||
var received = new List<byte>();
|
||||
new DeferredServiceLevelPublisher().SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData((byte)100)]
|
||||
[InlineData((byte)0)]
|
||||
public void Binding_replays_the_latest_request_including_zero(byte latest)
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(240);
|
||||
publisher.Publish(latest);
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new[] { latest });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Unbinding_retains_latest_for_rebinding()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
var old = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(old.Add));
|
||||
publisher.Publish(240);
|
||||
publisher.SetInner(null);
|
||||
publisher.Publish(0);
|
||||
var replacement = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(replacement.Add));
|
||||
old.ShouldBe(new byte[] { 240 });
|
||||
replacement.ShouldBe(new byte[] { 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_publish_retains_the_requested_value_for_replay()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.SetInner(new CallbackPublisher(_ => throw new InvalidOperationException()));
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new byte[] { 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_replay_keeps_candidate_bound_for_later_publish()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(100);
|
||||
var fail = true;
|
||||
var received = new List<byte>();
|
||||
var candidate = new CallbackPublisher(value =>
|
||||
{
|
||||
if (fail) throw new InvalidOperationException();
|
||||
received.Add(value);
|
||||
});
|
||||
Should.Throw<InvalidOperationException>(() => publisher.SetInner(candidate));
|
||||
fail = false;
|
||||
publisher.Publish(0);
|
||||
publisher.SetInner(null);
|
||||
publisher.SetInner(candidate);
|
||||
received.ShouldBe(new byte[] { 0, 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Failed_replay_retains_latest_for_a_replacement_publisher()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
publisher.Publish(100);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.SetInner(
|
||||
new CallbackPublisher(_ => throw new InvalidOperationException())));
|
||||
var received = new List<byte>();
|
||||
publisher.SetInner(new CallbackPublisher(received.Add));
|
||||
received.ShouldBe(new byte[] { 100 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Binding_waits_for_inflight_publish_then_replays_latest()
|
||||
{
|
||||
var publisher = new DeferredServiceLevelPublisher();
|
||||
using var entered = new ManualResetEventSlim();
|
||||
using var release = new ManualResetEventSlim();
|
||||
publisher.SetInner(new CallbackPublisher(_ =>
|
||||
{
|
||||
entered.Set();
|
||||
release.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
}));
|
||||
var publish = Task.Run(() => publisher.Publish(100), TestContext.Current.CancellationToken);
|
||||
var received = new List<byte>();
|
||||
using var bindingStarted = new ManualResetEventSlim();
|
||||
Exception? bindingError = null;
|
||||
var binding = new Thread(() =>
|
||||
{
|
||||
bindingStarted.Set();
|
||||
try { publisher.SetInner(new CallbackPublisher(received.Add)); }
|
||||
catch (Exception ex) { bindingError = ex; }
|
||||
});
|
||||
var bindingWasStarted = false;
|
||||
try
|
||||
{
|
||||
entered.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
binding.Start();
|
||||
bindingWasStarted = true;
|
||||
bindingStarted.Wait(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken).ShouldBeTrue();
|
||||
// Past the marker, SetInner is the thread's only potentially blocking operation.
|
||||
// Observe actual contention rather than assuming Task.Run has entered the adapter.
|
||||
SpinWait.SpinUntil(() => !binding.IsAlive ||
|
||||
(binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin) != 0,
|
||||
TimeSpan.FromSeconds(3)).ShouldBeTrue();
|
||||
(binding.ThreadState & System.Threading.ThreadState.WaitSleepJoin).ShouldNotBe(
|
||||
(System.Threading.ThreadState)0, "binding must wait for the in-flight delivery");
|
||||
}
|
||||
finally
|
||||
{
|
||||
release.Set();
|
||||
if (bindingWasStarted) binding.Join(TimeSpan.FromSeconds(5)).ShouldBeTrue();
|
||||
await publish.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken);
|
||||
}
|
||||
bindingError.ShouldBeNull();
|
||||
publisher.Publish(0);
|
||||
received.ShouldBe(new byte[] { 100, 0 });
|
||||
}
|
||||
|
||||
private sealed class CallbackPublisher(Action<byte> publish) : IServiceLevelPublisher
|
||||
{
|
||||
public void Publish(byte serviceLevel) => publish(serviceLevel);
|
||||
}
|
||||
|
||||
// ---- test double ----
|
||||
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
public sealed class FailedServiceLevelBindingCleanupTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Successful_cleanup_observes_stop_before_disposal_and_releases_ownership()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); return Task.CompletedTask; },
|
||||
() => { events.Add("dispose"); return ValueTask.CompletedTask; },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Failed_stop_still_disposes_and_retains_ownership_for_retry()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); throw new InvalidOperationException("controlled stop failure"); },
|
||||
() => { events.Add("dispose"); return ValueTask.CompletedTask; },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Failed_disposal_retains_ownership_without_escaping_the_binding_failure_handler()
|
||||
{
|
||||
var events = new List<string>();
|
||||
var released = await FailedServiceLevelBindingCleanup.StopAndDisposeAsync(
|
||||
() => { events.Add("stop"); return Task.CompletedTask; },
|
||||
() => { events.Add("dispose"); throw new InvalidOperationException("controlled disposal failure"); },
|
||||
NullLogger.Instance);
|
||||
events.ShouldBe(new[] { "stop", "dispose" });
|
||||
released.ShouldBeFalse();
|
||||
}
|
||||
}
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using Akka.Hosting;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Tests.OpcUa;
|
||||
|
||||
public sealed class OtOpcUaServerHostedServiceBindingFailureTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task Binding_failure_closes_started_endpoint_and_allows_other_host_services_to_start()
|
||||
{
|
||||
using var deadline = new CancellationTokenSource(TimeSpan.FromSeconds(15));
|
||||
using var probe = new TcpListener(IPAddress.Loopback, 0);
|
||||
probe.Start();
|
||||
var port = ((IPEndPoint)probe.LocalEndpoint).Port;
|
||||
probe.Stop();
|
||||
var pki = Path.Combine(Path.GetTempPath(), $"otopcua-binding-{Guid.NewGuid():N}");
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.Publish(0);
|
||||
var bindingReached = false;
|
||||
var listenerObserved = false;
|
||||
OtOpcUaServerHostedService? service = null;
|
||||
var factory = new FaultInjectingLoggerFactory(() =>
|
||||
{
|
||||
using var connected = new TcpClient();
|
||||
connected.Connect(IPAddress.Loopback, port);
|
||||
listenerObserved = connected.Connected;
|
||||
bindingReached = true;
|
||||
throw new InvalidOperationException("controlled binding factory failure");
|
||||
});
|
||||
service = new OtOpcUaServerHostedService(
|
||||
Options.Create(new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.BindingFailure",
|
||||
ApplicationUri = $"urn:OtOpcUa.BindingFailure:{Guid.NewGuid():N}",
|
||||
OpcUaPort = port, PublicHostname = "localhost", PkiStoreRoot = pki,
|
||||
}), new DeferredAddressSpaceSink(), deferred,
|
||||
NullOpcUaUserAuthenticator.Instance, NullHistorianDataSource.Instance,
|
||||
EmptyDataPlaneAuthoritySource.Instance, new ConfigurationBuilder().Build(),
|
||||
() => throw new NotSupportedException("no actor system required"), new ActorRegistry(),
|
||||
null, Options.Create(new AclOptions()), TimeProvider.System, factory);
|
||||
var admin = new AdminBootSentinel();
|
||||
using var host = new HostBuilder().ConfigureServices(services =>
|
||||
{
|
||||
services.AddSingleton<IHostedService>(service);
|
||||
services.AddSingleton<IHostedService>(admin);
|
||||
}).Build();
|
||||
try
|
||||
{
|
||||
await host.StartAsync(deadline.Token);
|
||||
listenerObserved.ShouldBeTrue("the SDK endpoint was actually listening before replay");
|
||||
bindingReached.ShouldBeTrue();
|
||||
admin.Started.ShouldBeTrue();
|
||||
using var client = new TcpClient();
|
||||
await Should.ThrowAsync<SocketException>(async () =>
|
||||
await client.ConnectAsync(IPAddress.Loopback, port, deadline.Token));
|
||||
// After construction failure, subsequent requests are accepted into the unbound cache.
|
||||
deferred.Publish(100);
|
||||
await host.StopAsync(deadline.Token);
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.DisposeAsync();
|
||||
if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FaultInjectingLoggerFactory(Action installFault) : ILoggerFactory
|
||||
{
|
||||
public ILogger CreateLogger(string categoryName)
|
||||
{
|
||||
if (categoryName == typeof(SdkServiceLevelPublisher).FullName) installFault();
|
||||
return NullLogger.Instance;
|
||||
}
|
||||
public void AddProvider(ILoggerProvider provider) { }
|
||||
public void Dispose() { }
|
||||
}
|
||||
|
||||
private sealed class AdminBootSentinel : IHostedService
|
||||
{
|
||||
public bool Started { get; private set; }
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
Started = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
+71
@@ -0,0 +1,71 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>Local loopback evidence that late binding replaces the SDK default on the wire.</summary>
|
||||
public sealed class ServiceLevelBindingReplayTests
|
||||
{
|
||||
[Theory]
|
||||
[InlineData((byte)100)]
|
||||
[InlineData((byte)0)]
|
||||
public async Task Late_binding_replays_latest_request_to_standard_ServiceLevel_read(byte latest)
|
||||
{
|
||||
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken);
|
||||
deadline.CancelAfter(TimeSpan.FromSeconds(15));
|
||||
var ct = deadline.Token;
|
||||
var pki = Path.Combine(Path.GetTempPath(), $"otopcua-replay-{Guid.NewGuid():N}");
|
||||
var clientPki = TestClientSecurity.AllocatePkiRoot();
|
||||
var port = 0;
|
||||
try
|
||||
{
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.Publish(240);
|
||||
deferred.Publish(latest);
|
||||
var started = await OpcUaHostStart.StartAsync(p =>
|
||||
{
|
||||
port = p;
|
||||
return new OpcUaApplicationHost(new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.Replay",
|
||||
ApplicationUri = $"urn:OtOpcUa.Replay:{Path.GetFileName(pki)}",
|
||||
OpcUaPort = p, PublicHostname = "localhost", PkiStoreRoot = pki,
|
||||
}, NullLogger<OpcUaApplicationHost>.Instance);
|
||||
}, () => new StandardServer(), ct);
|
||||
await using var host = started.Host;
|
||||
deferred.SetInner(new SdkServiceLevelPublisher(started.Server.CurrentInstance,
|
||||
NullLogger<SdkServiceLevelPublisher>.Instance));
|
||||
var config = new ApplicationConfiguration
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ReplayClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.ReplayClient:{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = TestClientSecurity.Build(clientPki),
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 10_000 },
|
||||
TransportQuotas = new TransportQuotas { OperationTimeout = 5_000 },
|
||||
};
|
||||
await config.ValidateAsync(ApplicationType.Client, ct);
|
||||
config.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(config,
|
||||
$"opc.tcp://localhost:{port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);
|
||||
using var session = await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
config, new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(config)),
|
||||
updateBeforeConnect: false, checkDomain: false, sessionName: "ServiceLevelReplay",
|
||||
sessionTimeout: 10_000, identity: new UserIdentity(new AnonymousIdentityToken()),
|
||||
preferredLocales: null, ct: ct);
|
||||
var read = await session.ReadValueAsync(new Opc.Ua.NodeId(2267u), ct);
|
||||
read.StatusCode.ShouldBe(StatusCodes.Good);
|
||||
read.Value.ShouldBe(latest);
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (Directory.Exists(pki)) Directory.Delete(pki, recursive: true);
|
||||
if (Directory.Exists(clientPki)) Directory.Delete(clientPki, recursive: true);
|
||||
}
|
||||
}
|
||||
}
|
||||
+4
-4
@@ -6,9 +6,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
public sealed class DeferredServiceLevelPublisherTests
|
||||
{
|
||||
/// <summary>Verifies that publish before SetInner is a safe no-op.</summary>
|
||||
/// <summary>Verifies that publish before SetInner safely accepts a cached request.</summary>
|
||||
[Fact]
|
||||
public void Publish_before_SetInner_is_a_safe_noop()
|
||||
public void Publish_before_SetInner_accepts_a_cached_request()
|
||||
{
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
|
||||
@@ -28,9 +28,9 @@ public sealed class DeferredServiceLevelPublisherTests
|
||||
recording.LastValue.ShouldBe((byte)200);
|
||||
}
|
||||
|
||||
/// <summary>Verifies that SetInner with null reverts to the null publisher.</summary>
|
||||
/// <summary>Verifies that SetInner with null detaches the previous publisher.</summary>
|
||||
[Fact]
|
||||
public void SetInner_null_reverts_to_Null_publisher()
|
||||
public void SetInner_null_detaches_the_previous_publisher()
|
||||
{
|
||||
var recording = new RecordingPublisher();
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
|
||||
@@ -19,6 +19,40 @@ public sealed class SdkServiceLevelPublisherTests : IDisposable
|
||||
Path.GetTempPath(),
|
||||
$"otopcua-pki-{Guid.NewGuid():N}");
|
||||
|
||||
[Fact]
|
||||
public void Missing_ServiceLevel_is_reported_as_failure()
|
||||
{
|
||||
var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
|
||||
var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Sdk_write_failure_propagates_the_original_exception()
|
||||
{
|
||||
var server = System.Reflection.DispatchProxy.Create<IServerInternal, FaultingServer>();
|
||||
var proxy = (FaultingServer)(object)server;
|
||||
proxy.Node = new Opc.Ua.ServerObjectState(null)
|
||||
{
|
||||
ServiceLevel = new Opc.Ua.PropertyState<byte>(null),
|
||||
};
|
||||
proxy.Fault = new InvalidOperationException("controlled diagnostics failure");
|
||||
var publisher = new SdkServiceLevelPublisher(server, NullLogger<SdkServiceLevelPublisher>.Instance);
|
||||
Should.Throw<InvalidOperationException>(() => publisher.Publish(0)).ShouldBeSameAs(proxy.Fault);
|
||||
}
|
||||
|
||||
public class FaultingServer : System.Reflection.DispatchProxy
|
||||
{
|
||||
public Opc.Ua.ServerObjectState? Node { get; set; }
|
||||
public Exception? Fault { get; set; }
|
||||
protected override object? Invoke(System.Reflection.MethodInfo? method, object?[]? args)
|
||||
{
|
||||
if (method?.Name == "get_ServerObject") return Node;
|
||||
if (method?.Name == "get_DiagnosticsLock") throw Fault!;
|
||||
throw new NotSupportedException(method?.Name);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that the publisher writes values to the standard Server.ServiceLevel variable.</summary>
|
||||
[Fact]
|
||||
public async Task Publish_writes_value_to_Server_ServiceLevel_variable()
|
||||
|
||||
@@ -15,6 +15,49 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.OpcUa;
|
||||
|
||||
public sealed class OpcUaPublishActorTests : RuntimeActorTestBase
|
||||
{
|
||||
[Fact]
|
||||
public void Failed_value_is_retried_and_last_level_only_tracks_success()
|
||||
{
|
||||
var sink = new RecoveringPublisher { Fail = false };
|
||||
var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: sink));
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Fail = true;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
actor.UnderlyingActor.LastServiceLevel.ShouldBe((byte)100);
|
||||
sink.Fail = false;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
sink.Delivered.ShouldBe(new byte[] { 100, 0 });
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Requesting_prior_success_after_failure_updates_delivery_and_replay()
|
||||
{
|
||||
var sink = new RecoveringPublisher();
|
||||
var deferred = new DeferredServiceLevelPublisher();
|
||||
deferred.SetInner(sink);
|
||||
var actor = ActorOfAsTestActorRef<OpcUaPublishActor>(OpcUaPublishActor.PropsForTests(serviceLevel: deferred));
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Fail = true;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(0));
|
||||
sink.Fail = false;
|
||||
actor.Tell(new OpcUaPublishActor.ServiceLevelChanged(100));
|
||||
sink.Delivered.ShouldBe(new byte[] { 100, 100 });
|
||||
var rebound = new RecoveringPublisher();
|
||||
deferred.SetInner(rebound);
|
||||
rebound.Delivered.ShouldBe(new byte[] { 100 });
|
||||
}
|
||||
|
||||
private sealed class RecoveringPublisher : IServiceLevelPublisher
|
||||
{
|
||||
public bool Fail { get; set; }
|
||||
public List<byte> Delivered { get; } = new();
|
||||
public void Publish(byte level)
|
||||
{
|
||||
if (Fail) throw new InvalidOperationException("controlled failure");
|
||||
Delivered.Add(level);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that message contracts are accepted without a pinned dispatcher in tests.</summary>
|
||||
[Fact]
|
||||
public void Accepts_message_contracts_without_pinned_dispatcher_in_tests()
|
||||
|
||||
Reference in New Issue
Block a user