dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge feat/804-mtconnect-parity

Browse Source
This commit is contained in:
Joseph Doherty
2026-11-10 23:24:27 -05:00
parent 83e8cf767d f17ff1e643
commit 8a0f0df05f
12 changed files with 1332 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.IntegrationTests/MTConnectAgentIntegrationTests.cs
+110
View File
@@ -837,6 +837,116 @@ public sealed class MTConnectAgentIntegrationTests(MTConnectAgentFixture agent)
}
}
// ---- #804: the 2.x Agent's error documents under a non-2xx status ----
/// <summary>
/// #804 — a 2.x Agent answers a <c>from</c> outside its buffer with <b>HTTP 400</b> + <c>MTConnectError</c>
/// <c>OUT_OF_RANGE</c> (live-verified against this fixture). The client must surface it as the
/// <see cref="InvalidDataException"/> the pump re-baselines on; before #804 it was a bare
/// <see cref="HttpRequestException"/>, which the pump read as transport loss and reconnected at the same
/// evicted cursor — the documented OUT_OF_RANGE re-baseline never ran against a real 2.x Agent.
/// </summary>
[Fact(Timeout = TestTimeoutMs)]
public async Task A_sample_cursor_outside_the_buffer_is_the_agents_out_of_range_refusal()
{
SkipIfDown();
// A cursor the Agent cannot hold: far past its nextSequence. (An OLD cursor is not reliable here — a
// freshly restarted Agent may still hold sequence 1, and then the stream simply opens.)
var next = (long?)(await CurrentAsync()).Root?.Elements().FirstOrDefault(e => e.Name.LocalName == "Header")
?.Attribute("nextSequence") ?? 0L;
using var client = new MTConnectAgentClient(LiveOptions());
var ex = await Should.ThrowAsync<InvalidDataException>(async () =>
{
await foreach (var _ in client.SampleAsync(next + 1_000_000, Ct))
{
}
});
ex.Message.ShouldContain("OUT_OF_RANGE");
}
/// <summary>
/// #804 — a <c>SampleCount</c> at or above the Agent's <c>bufferSize</c> is refused <c>OUT_OF_RANGE</c> on
/// EVERY <c>/sample</c> open (live: <c>'count' must be less than N</c>). The driver keeps serving values off
/// <c>/current</c> re-baselines, and — after three in a row with no <c>/sample</c> chunk — reports the stream
/// Degraded with <c>LastSuccessfulRead</c> still advancing, rather than Healthy for ever (before #804 the
/// 400 was a transport fault: Degraded, but every value stuck at the priming <c>/current</c>).
/// </summary>
[Fact(Timeout = TestTimeoutMs)]
public async Task A_sample_count_the_agent_refuses_is_reported_degraded_while_values_keep_flowing()
{
SkipIfDown();
var bufferSize = (int?)agent.ProbeDocument?.Root?.Elements().FirstOrDefault(e => e.Name.LocalName == "Header")
?.Attribute("bufferSize");
if (bufferSize is not > 0)
{
Assert.Skip($"The Agent at {agent.AgentUri} declares no bufferSize in its /probe header.");
}
var ids = await NumericSampleIdsAsync();
if (ids.Count == 0)
{
Assert.Skip($"The Agent at {agent.AgentUri} is currently reporting no numeric SAMPLE value.");
}
var options = new MTConnectDriverOptions
{
AgentUri = agent.AgentUri,
RequestTimeoutMs = 15_000,
SampleIntervalMs = 200,
SampleCount = bufferSize.Value + 1,
HeartbeatMs = 5_000,
Probe = new MTConnectProbeOptions { Enabled = false },
Reconnect = new MTConnectReconnectOptions { MinBackoffMs = 0, MaxBackoffMs = 200 },
};
var driver = new MTConnectDriver(options, "mtconnect-live-oor");
await using var live = new LiveDriver(driver);
await driver.InitializeAsync(
JsonSerializer.Serialize(new
{
agentUri = options.AgentUri,
requestTimeoutMs = options.RequestTimeoutMs,
sampleIntervalMs = options.SampleIntervalMs,
sampleCount = options.SampleCount,
heartbeatMs = options.HeartbeatMs,
probe = new { enabled = false },
reconnect = new { minBackoffMs = 0, maxBackoffMs = 200, backoffMultiplier = 2.0 },
tags = ids.Select(id => new { fullName = id, driverDataType = "Float64" }),
}),
Ct);
var seen = new ConcurrentQueue<DataChangeEventArgs>();
driver.OnDataChange += (_, e) => seen.Enqueue(e);
await driver.SubscribeAsync([.. ids], TimeSpan.FromMilliseconds(200), Ct);
var deadline = DateTime.UtcNow + StreamWait;
while (driver.GetHealth().LastError != "repeated /sample re-baselines with no observations in between" &&
DateTime.UtcNow < deadline)
{
await Task.Delay(100, Ct);
}
var health = driver.GetHealth();
health.State.ShouldBe(DriverState.Degraded);
health.LastError.ShouldBe("repeated /sample re-baselines with no observations in between");
var stamp = health.LastSuccessfulRead.ShouldNotBeNull();
// Values keep flowing off the re-baselines: the stamp advances and Good values keep arriving.
var before = seen.Count;
deadline = DateTime.UtcNow + StreamWait;
while ((driver.GetHealth().LastSuccessfulRead <= stamp || seen.Count <= before) && DateTime.UtcNow < deadline)
{
await Task.Delay(100, Ct);
}
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
driver.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(stamp);
seen.Skip(before).ShouldContain(e => e.Snapshot.StatusCode == Good);
}
private static bool Is(string? value, string expected) =>
string.Equals(value?.Trim(), expected, StringComparison.OrdinalIgnoreCase);
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectJavaParityTests.cs
+482
View File
@@ -0,0 +1,482 @@
using System.Net;
using System.Text;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
/// <summary>
/// Gitea #804 — the gaps a parity review against the ignitionoee Java MTConnect driver verified in this
/// driver, one region per gap. Each test failed against the pre-#804 driver; the capability matrix that
/// says which Java behaviours were already covered (and by which existing tests) is
/// <c>docs/drivers/MTConnect.md</c> §"Java reference parity (#804)".
/// </summary>
public sealed class MTConnectJavaParityTests
{
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private const string OutOfRangeError =
"""
<?xml version="1.0" encoding="UTF-8"?>
<MTConnectError xmlns="urn:mtconnect.org:MTConnectError:2.0">
<Header creationTime="2026-11-10T12:00:00Z" sender="agent" instanceId="1" version="2.7.0.12" bufferSize="16384"/>
<Errors>
<Error errorCode="OUT_OF_RANGE">'from' must be greater than 59962705</Error>
</Errors>
</MTConnectError>
""";
private const string NoDeviceError =
"""
<MTConnectError xmlns="urn:mtconnect.org:MTConnectError:2.0">
<Errors><Error errorCode="NO_DEVICE">Could not find the device 'nosuch'</Error></Errors>
</MTConnectError>
""";
private static MTConnectAgentClient NewClient(HttpMessageHandler handler) =>
new(new MTConnectDriverOptions { AgentUri = "http://agent:5000", RequestTimeoutMs = 5000 }, handler);
private static async Task DrainAsync(IAsyncEnumerable<MTConnectStreamsResult> stream)
{
await foreach (var _ in stream)
{
}
}
// ───────────── gap 1: an Agent error document under a non-2xx status is the Agent's refusal ─────────────
// cppagent 2.x (the fixture's mtconnect/agent:2.7, verified live) answers an evicted `from` and a count at or
// above its buffer with HTTP 400 + MTConnectError OUT_OF_RANGE. Pre-#804 the client threw a bare
// HttpRequestException, which the pump treats as a transient transport loss: it reconnected at the SAME
// evicted cursor for ever and the OUT_OF_RANGE re-baseline never ran against a real 2.x Agent.
[Fact]
public async Task Sample_answered_400_with_an_out_of_range_document_surfaces_as_the_agents_refusal()
{
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.BadRequest, OutOfRangeError)));
var ex = await Should.ThrowAsync<InvalidDataException>(async () => await DrainAsync(client.SampleAsync(1, Ct)));
ex.Message.ShouldContain("OUT_OF_RANGE: 'from' must be greater than 59962705");
ex.Message.ShouldContain("HTTP 400");
MTConnectXml.IsAgentRefusal(ex).ShouldBeTrue();
}
[Fact]
public async Task Current_answered_404_with_a_no_device_document_is_an_agent_refusal_not_a_comms_fault()
{
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.NotFound, NoDeviceError)));
var ex = await Should.ThrowAsync<InvalidDataException>(() => client.CurrentAsync(Ct));
ex.Message.ShouldContain("NO_DEVICE");
MTConnectXml.IsAgentRefusal(ex).ShouldBeTrue();
}
[Fact]
public async Task Probe_answered_404_with_an_error_document_carries_the_agents_own_words()
{
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.NotFound, NoDeviceError)));
var ex = await Should.ThrowAsync<InvalidDataException>(() => client.ProbeAsync(Ct));
ex.Message.ShouldContain("Could not find the device 'nosuch'");
ex.Message.ShouldContain("expected <MTConnectDevices>");
}
/// <summary>Positive control: a non-2xx body that is NOT an MTConnect error document keeps the status-only failure.</summary>
[Fact]
public async Task A_non_2xx_proxy_page_is_still_a_plain_http_failure()
{
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.BadGateway, "<html><body>502</body></html>")));
await Should.ThrowAsync<HttpRequestException>(() => client.CurrentAsync(Ct));
}
/// <summary>The failure body is read only up to the cap: a huge error page never becomes a huge string.</summary>
[Fact]
public async Task An_error_document_beyond_the_body_cap_is_not_read_in_full()
{
var padding = new string(' ', 70 * 1024); // past the 64 KiB failure-body cap
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.BadRequest, padding + OutOfRangeError)));
// The document starts past the cap, so it is never seen: status-only failure, not a refusal.
await Should.ThrowAsync<HttpRequestException>(async () => await DrainAsync(client.SampleAsync(1, Ct)));
}
// ───────────── gap 2 (driver-level, MTConnectSubscribeTests): repeated re-baselines without progress ─────────────
// ───────────── gap 3: streams parser nesting bound ─────────────
// ReadElement/ReadChildren recurse per nesting level; an unbounded depth from an unauthenticated peer is a
// StackOverflowException, which .NET cannot catch — the whole server process dies.
private static string Streams(string body) =>
$"""
<MTConnectStreams xmlns="urn:mtconnect.org:MTConnectStreams:2.0">
<Header creationTime="2026-11-10T12:00:00Z" instanceId="1" bufferSize="16" firstSequence="1" lastSequence="5" nextSequence="6"/>
<Streams>{body}</Streams>
</MTConnectStreams>
""";
[Fact]
public void Streams_parse_refuses_absurd_nesting_instead_of_recursing_without_bound()
{
const int depth = 80; // beyond the 64-level bound
var xml = Streams(string.Concat(Enumerable.Repeat("<x>", depth)) + string.Concat(Enumerable.Repeat("</x>", depth)));
var ex = Should.Throw<InvalidDataException>(() => MTConnectStreamsParser.Parse(xml));
ex.Message.ShouldContain("levels deep");
}
/// <summary>Positive control: a real document's nesting (with a TABLE's Entry/Cell) parses under the bound.</summary>
[Fact]
public void Streams_parse_accepts_ordinary_device_nesting()
{
var xml = Streams(
"""
<DeviceStream name="d" uuid="u"><ComponentStream component="Path" componentId="p">
<Events>
<WorkOffsetTable dataItemId="wo" timestamp="2026-11-10T12:00:00Z" sequence="5" count="1">
<Entry key="G54"><Cell key="X">1.0</Cell></Entry>
</WorkOffsetTable>
<Execution dataItemId="exec" timestamp="2026-11-10T12:00:00Z" sequence="4">ACTIVE</Execution>
</Events>
</ComponentStream></DeviceStream>
""");
var result = MTConnectStreamsParser.Parse(xml);
result.Observations.Select(o => o.DataItemId).ShouldBe(["wo", "exec"]);
}
// ───────────── gap 4: DOCTYPE refused on every tree-building parse ─────────────
// The streams parser already prohibited DTDs; the /probe parse (and the write-answer parse) went through
// XDocument.Parse, which PARSES a DTD and expands its internal entities.
private const string ProbeWithDoctype =
"""
<?xml version="1.0"?>
<!DOCTYPE MTConnectDevices [ <!ENTITY injected "Injected"> ]>
<MTConnectDevices xmlns="urn:mtconnect.org:MTConnectDevices:2.0">
<Header instanceId="1"/>
<Devices><Device id="d1" name="&injected;"><DataItems><DataItem id="a" category="EVENT" type="AVAILABILITY"/></DataItems></Device></Devices>
</MTConnectDevices>
""";
[Fact]
public void Probe_parse_refuses_a_doctype_from_a_string()
{
var ex = Should.Throw<InvalidDataException>(() => MTConnectProbeParser.Parse(ProbeWithDoctype));
ex.Message.ShouldContain("not well-formed XML");
}
[Fact]
public void Probe_parse_refuses_a_doctype_from_a_stream()
{
using var stream = new MemoryStream(Encoding.UTF8.GetBytes(ProbeWithDoctype));
Should.Throw<InvalidDataException>(() => MTConnectProbeParser.Parse(stream));
}
/// <summary>A write answered with a DOCTYPE'd document cannot be read, so it is NOT reported as accepted.</summary>
[Fact]
public async Task A_write_answered_with_a_doctype_document_is_not_confirmed()
{
const string body =
"""
<!DOCTYPE MTConnectError [ <!ENTITY e "x"> ]>
<MTConnectError><Errors><Error errorCode="UNSUPPORTED">&e;</Error></Errors></MTConnectError>
""";
using var client = NewClient(new StubHandler(_ => Respond(HttpStatusCode.OK, body)));
var outcome = await client.WriteObservationAsync("d", "item", "v", Ct);
outcome.Accepted.ShouldBeFalse();
outcome.AgentError.ShouldNotBeNull().ShouldContain("DOCTYPE");
}
// ───────────── gap 5: no fabricated non-finite floats ─────────────
// .NET parses an overflowing literal to ±Infinity and accepts NaN/Infinity spellings; all were Good.
[Theory]
[InlineData(DriverDataType.Float64, "1e400", 0x803C0000u)] // BadOutOfRange — a number the type cannot hold
[InlineData(DriverDataType.Float64, "-1e400", 0x803C0000u)]
[InlineData(DriverDataType.Float32, "1e39", 0x803C0000u)] // fits a double, overflows a float
[InlineData(DriverDataType.Float64, "NaN", 0x80740000u)] // BadTypeMismatch — not a measurement
[InlineData(DriverDataType.Float64, "Infinity", 0x80740000u)]
[InlineData(DriverDataType.Float32, "-Infinity", 0x80740000u)]
public void A_non_finite_float_is_never_served_as_good(DriverDataType type, string text, uint expected)
{
var snap = Coerce(type, text);
snap.StatusCode.ShouldBe(expected);
snap.Value.ShouldBeNull();
}
// ───────────── gap 6: an integer-valued decimal on an integral tag ─────────────
// Agents format integer-valued items as "42.000000"; that was BadOutOfRange on every integral tag.
[Theory]
[InlineData(DriverDataType.Int32, "42.000000", 42)]
[InlineData(DriverDataType.Int64, "1.5e3", 1500L)]
[InlineData(DriverDataType.Int16, "-7.0", (short)-7)]
[InlineData(DriverDataType.UInt32, "4000000000.0", 4000000000u)]
public void An_integer_valued_decimal_converts_losslessly(DriverDataType type, string text, object expected)
{
var snap = Coerce(type, text);
snap.StatusCode.ShouldBe(0u);
snap.Value.ShouldBe(expected);
}
/// <summary>…but a REAL fraction is still refused, never truncated (deliberately stricter than the Java driver).</summary>
[Theory]
[InlineData(DriverDataType.Int32, "42.5")]
[InlineData(DriverDataType.Int16, "40000.0")]
[InlineData(DriverDataType.UInt16, "-1.0")]
[InlineData(DriverDataType.Int64, "1e30")]
public void A_fraction_or_an_out_of_range_decimal_stays_out_of_range(DriverDataType type, string text)
{
var snap = Coerce(type, text);
snap.StatusCode.ShouldBe(0x803C0000u);
snap.Value.ShouldBeNull();
}
private static DataValueSnapshot Coerce(DriverDataType type, string text)
{
var index = new MTConnectObservationIndex([new MTConnectTagDefinition("x", type)]);
index.Apply(new MTConnectStreamsResult(
1L, 2L, 1L, [new MTConnectObservation("x", text, new DateTime(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc))]));
return index.Get("x");
}
// ───────────── gap 7: sibling containers resolving to one folder name ─────────────
// Two same-named sibling folders share one path key in the browse picker; the second one's data items were
// listed but unreachable on expand. Merged into ONE folder (what the Java driver does).
private const string CollidingProbe =
"""
<MTConnectDevices xmlns="urn:mtconnect.org:MTConnectDevices:2.0">
<Header instanceId="1"/>
<Devices>
<Device id="d1" name="Mill">
<Components>
<Axes id="ax" name="Axes">
<Components>
<Rotary id="c_rot" name="C"><DataItems><DataItem id="c_angle" category="SAMPLE" type="ANGLE"/></DataItems></Rotary>
<Linear id="c_lin" name="C"><DataItems><DataItem id="c_pos" category="SAMPLE" type="POSITION"/></DataItems></Linear>
<Linear id="X"><DataItems><DataItem id="x_pos" category="SAMPLE" type="POSITION"/></DataItems></Linear>
<Linear id="x2" name="X"><DataItems><DataItem id="x2_pos" category="SAMPLE" type="POSITION"/></DataItems></Linear>
</Components>
</Axes>
</Components>
</Device>
</Devices>
</MTConnectDevices>
""";
[Fact]
public async Task Sibling_components_resolving_to_one_name_share_one_folder_and_keep_every_data_item()
{
var client = CannedAgentClient.FromFixtures();
client.Probe = MTConnectProbeParser.Parse(CollidingProbe);
var driver = new MTConnectDriver(new MTConnectDriverOptions { AgentUri = "http://agent:5000" }, "mt1", _ => client);
await driver.InitializeAsync("{}", Ct);
var cap = new CapturingBuilder();
await driver.DiscoverAsync(cap, Ct);
cap.Folders.Select(f => f.Path).ShouldBe(["Mill", "Mill/Axes", "Mill/Axes/C", "Mill/Axes/X"]);
cap.Variables.Single(v => v.Attr.FullName == "c_angle").ParentPath.ShouldBe("Mill/Axes/C");
cap.Variables.Single(v => v.Attr.FullName == "c_pos").ParentPath.ShouldBe("Mill/Axes/C");
cap.Variables.Single(v => v.Attr.FullName == "x_pos").ParentPath.ShouldBe("Mill/Axes/X"); // id fallback
cap.Variables.Single(v => v.Attr.FullName == "x2_pos").ParentPath.ShouldBe("Mill/Axes/X"); // name
}
// ───────────── gap 8: structural elements are not components ─────────────
[Fact]
public void A_structural_element_under_components_or_devices_is_skipped_not_fatal()
{
const string xml =
"""
<MTConnectDevices xmlns="urn:mtconnect.org:MTConnectDevices:2.0">
<Header instanceId="1"/>
<Devices>
<Description>agent-wide note</Description>
<Device id="d1" name="Mill">
<Components>
<Description>stray, schema-invalid</Description>
<Controller id="ctl"><DataItems><DataItem id="mode" category="EVENT" type="CONTROLLER_MODE"/></DataItems></Controller>
</Components>
</Device>
</Devices>
</MTConnectDevices>
""";
var model = MTConnectProbeParser.Parse(xml);
model.Devices.ShouldHaveSingleItem().Components.ShouldHaveSingleItem().Id.ShouldBe("ctl");
}
// ───────────── plumbing ─────────────
private static HttpResponseMessage Respond(HttpStatusCode status, string body) =>
new(status) { Content = new StringContent(body, Encoding.UTF8, "application/xml") };
private sealed class StubHandler(Func<HttpRequestMessage, HttpResponseMessage> respond) : HttpMessageHandler
{
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken ct) =>
Task.FromResult(respond(request));
}
// ───────────── Java regression scenarios this driver already met (coverage, not gaps) ─────────────
// Ported from the Java MTConnectMultipartReaderTest / MTConnectStreamsParserTest cases that had no
// .NET counterpart. Each passed before #804; they pin behaviour the matrix records as "covered".
private static string Doc(long first, long next, string observations = "") => Streams(
$"""<DeviceStream name="d"><ComponentStream component="Path" componentId="p"><Events>{observations}</Events></ComponentStream></DeviceStream>""")
.Replace("firstSequence=\"1\"", $"firstSequence=\"{first}\"").Replace("nextSequence=\"6\"", $"nextSequence=\"{next}\"");
private static HttpResponseMessage MultipartResponse(byte[] body) => MultipartResponse(new ByteArrayContent(body));
private static HttpResponseMessage MultipartResponse(HttpContent content)
{
content.Headers.ContentType =
System.Net.Http.Headers.MediaTypeHeaderValue.Parse("multipart/x-mixed-replace; boundary=b0");
return new HttpResponseMessage(HttpStatusCode.OK) { Content = content };
}
private static async Task<List<MTConnectStreamsResult>> CollectAsync(MTConnectAgentClient client, int count)
{
var results = new List<MTConnectStreamsResult>();
await foreach (var chunk in client.SampleAsync(1, Ct))
{
results.Add(chunk);
if (results.Count == count)
{
break;
}
}
return results;
}
/// <summary>Java <c>contentLengthIsHonoredOverBoundaryScanning</c>: a document whose text contains the boundary token is returned whole.</summary>
[Fact]
public async Task A_declared_content_length_is_authoritative_over_a_boundary_token_inside_the_document()
{
var doc = Doc(1, 2, "<Message dataItemId=\"m\" timestamp=\"2026-11-10T12:00:00Z\">see\r\n--b0 inside the text</Message>");
var payload = Encoding.UTF8.GetBytes(doc);
var body = Encoding.ASCII.GetBytes($"--b0\r\nContent-type: text/xml\r\nContent-length: {payload.Length}\r\n\r\n")
.Concat(payload).Concat("\r\n--b0--\r\n"u8.ToArray()).ToArray();
using var client = NewClient(new StubHandler(_ => MultipartResponse(body)));
var chunk = (await CollectAsync(client, 1)).ShouldHaveSingleItem();
chunk.Observations.ShouldHaveSingleItem().Value.ShouldContain("--b0 inside the text");
}
/// <summary>Java <c>preambleBeforeTheFirstBoundaryIsSkipped</c> + <c>lineFeedOnlyFramingIsAccepted</c>.</summary>
[Fact]
public async Task A_preamble_is_skipped_and_line_feed_only_framing_is_accepted()
{
var first = Encoding.UTF8.GetBytes(Doc(1, 2));
var second = Encoding.UTF8.GetBytes(Doc(2, 3));
var body = Encoding.ASCII.GetBytes("this is preamble text\n--b0\nContent-length: " + first.Length + "\n\n")
.Concat(first)
.Concat(Encoding.ASCII.GetBytes("\n--b0\nContent-length: " + second.Length + "\n\n"))
.Concat(second)
.Concat("\n--b0--\n"u8.ToArray()).ToArray();
using var client = NewClient(new StubHandler(_ => MultipartResponse(body)));
var chunks = await CollectAsync(client, 2);
chunks.Select(c => c.NextSequence).ShouldBe([2L, 3L]);
}
/// <summary>Java <c>aBodyShorterThanItsDeclaredContentLengthThrows</c>: truncation is a lost stream, never an orderly end.</summary>
[Fact]
public async Task A_part_shorter_than_its_declared_content_length_is_a_lost_stream()
{
var payload = Encoding.UTF8.GetBytes(Doc(1, 2));
var body = Encoding.ASCII.GetBytes($"--b0\r\nContent-length: {payload.Length + 500}\r\n\r\n").Concat(payload).ToArray();
using var client = NewClient(new StubHandler(_ => MultipartResponse(body)));
await Should.ThrowAsync<MTConnectStreamEndedException>(async () => await DrainAsync(client.SampleAsync(1, Ct)));
}
/// <summary>Java <c>aPartLargerThanTheBufferCapIsRefused</c>: a peer that never frames a part cannot grow the buffer unbounded.</summary>
[Fact]
public async Task A_part_larger_than_the_buffer_cap_is_refused()
{
var head = "--b0\r\nContent-type: text/xml\r\n\r\n"u8.ToArray();
using var client = NewClient(new StubHandler(_ => MultipartResponse(new StreamContent(new EndlessPart(head)))));
var ex = await Should.ThrowAsync<InvalidDataException>(async () => await DrainAsync(client.SampleAsync(1, Ct)));
ex.Message.ShouldContain("refusing to buffer further");
}
/// <summary>Java <c>timestampWithANonZuluOffsetIsNormalizedToTheSameInstant</c>: the SourceTimestamp is the Agent's instant, in UTC.</summary>
[Fact]
public void A_timestamp_with_a_non_zulu_offset_is_normalized_to_the_same_utc_instant()
{
var result = MTConnectStreamsParser.Parse(
Doc(1, 2, """<Execution dataItemId="e" timestamp="2026-11-10T14:00:00.5+02:00">ACTIVE</Execution>"""));
var at = result.Observations.ShouldHaveSingleItem().TimestampUtc;
at.ShouldBe(new DateTime(2026, 11, 10, 12, 0, 0, 500, DateTimeKind.Utc));
at.Kind.ShouldBe(DateTimeKind.Utc);
}
/// <summary>
/// Java <c>anAbsurdlyDeepErrorDocumentStillYieldsTheAgentsWords</c>: the new nesting bound applies to the
/// observation walk only, so a deeply nested error document still reports the Agent's own text.
/// </summary>
[Fact]
public void An_absurdly_deep_error_document_still_yields_the_agents_words()
{
var nest = string.Concat(Enumerable.Repeat("<x>", 100));
var unnest = string.Concat(Enumerable.Repeat("</x>", 100));
var xml = $"""<MTConnectError><Errors>{nest}<Error errorCode="OUT_OF_RANGE">deep</Error>{unnest}</Errors></MTConnectError>""";
var ex = Should.Throw<InvalidDataException>(() => MTConnectStreamsParser.Parse(xml));
ex.Message.ShouldContain("OUT_OF_RANGE: deep");
}
/// <summary>A read-only stream that serves <c>head</c> then an endless run of document bytes with no boundary.</summary>
private sealed class EndlessPart(byte[] head) : Stream
{
private long _position;
public override int Read(byte[] buffer, int offset, int count)
{
var written = 0;
while (written < count)
{
buffer[offset + written] = _position < head.Length ? head[_position] : (byte)'a';
_position++;
written++;
}
return written;
}
public override bool CanRead => true;
public override bool CanSeek => false;
public override bool CanWrite => false;
public override long Length => throw new NotSupportedException();
public override long Position { get => _position; set => throw new NotSupportedException(); }
public override void Flush() { }
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
public override void SetLength(long value) => throw new NotSupportedException();
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs
+129
View File
@@ -452,6 +452,56 @@ public sealed class MTConnectSubscribeTests
For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(9.5d);
}
/// <summary>
/// #804 (Java-driver parity: the restart test runs on the re-baseline snapshot too). On a real 2.x Agent a
/// restart is usually seen FIRST by the re-baseline, not by a chunk: the restarted process refuses the dead
/// one's cursor with HTTP 400 <c>OUT_OF_RANGE</c> (live-observed on the fixture), so no chunk ever reaches the
/// in-stream <c>instanceId</c> check. The re-baseline used to adopt the new id silently — no rediscovery
/// prompt, and the dead model's held values kept being served.
/// </summary>
[Fact]
public async Task An_agent_restart_seen_only_by_the_rebaseline_drops_held_values_and_raises_rediscovery()
{
var (driver, client) = await InitializedDriverAsync();
var raised = new ConcurrentQueue<RediscoveryEventArgs>();
driver.OnRediscoveryNeeded += (_, e) => raised.Enqueue(e);
var seen = Record(driver);
// The restarted Agent refuses the old cursor, and its /current knows nothing about dev1_execution.
client.SampleFailures.Enqueue(new InvalidDataException("OUT_OF_RANGE: 'from' must be less than 42 (served under HTTP 400)"));
client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5")));
await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
driver.AgentInstanceId.ShouldBe(RestartedInstanceId);
raised.ShouldHaveSingleItem().Reason.ShouldContain("instanceId");
client.LastSampleFrom.ShouldBe(42);
driver.ObservationIndex.Get("dev1_execution").StatusCode.ShouldBe(BadWaitingForInitialData);
driver.ObservationIndex.Get("dev1_pos").Value.ShouldBe(9.5d);
For(seen, "dev1_execution").Last().Snapshot.StatusCode.ShouldBe(BadWaitingForInitialData);
}
/// <summary>…control: an OUT_OF_RANGE re-baseline against the SAME Agent raises nothing and keeps its values.</summary>
[Fact]
public async Task An_out_of_range_rebaseline_against_the_same_agent_raises_no_rediscovery()
{
var (driver, client) = await InitializedDriverAsync();
var raised = new ConcurrentQueue<RediscoveryEventArgs>();
driver.OnRediscoveryNeeded += (_, e) => raised.Enqueue(e);
client.SampleFailures.Enqueue(new InvalidDataException("OUT_OF_RANGE (served under HTTP 400)"));
client.CurrentAnswers.Enqueue(Chunk(5000, 5005, ("dev1_pos", "2.5")));
await driver.SubscribeAsync(["dev1_pos", "dev1_execution"], TimeSpan.FromMilliseconds(50), Ct);
await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
raised.ShouldBeEmpty();
driver.AgentInstanceId.ShouldBe(FixtureInstanceId);
driver.ObservationIndex.Get("dev1_execution").StatusCode.ShouldBe(Good);
}
// ---- stream ends ----
/// <summary>
@@ -1116,6 +1166,85 @@ public sealed class MTConnectSubscribeTests
driver.ReconnectAttempts.ShouldBe(0);
}
/// <summary>
/// <b>#804 (Java-driver parity: <c>MAX_REBASELINES_WITHOUT_PROGRESS</c>).</b> An Agent that answers EVERY
/// <c>/sample</c> open with <c>OUT_OF_RANGE</c> (live: a <c>count</c> at or above its <c>bufferSize</c>) is
/// re-baselined from <c>/current</c> each time, and the backoff paces that — but every re-baseline also marked
/// the stream delivered, so the driver read <b>Healthy</b> for ever while not one <c>/sample</c> chunk arrived.
/// After <see cref="MTConnectDriver.MaxRebaselinesWithoutProgress"/> in a row it is Degraded and stays so
/// across further re-baselines, with <c>LastSuccessfulRead</c> still advancing (each <c>/current</c> is the
/// Agent answering — the comms-loss watchdog must not reconnect-loop a configuration problem).
/// </summary>
[Fact]
public async Task Repeated_rebaselines_without_a_delivered_chunk_report_the_stream_degraded()
{
var options = Opts(new MTConnectReconnectOptions { MinBackoffMs = 0, MaxBackoffMs = 1 });
var (driver, client) = await InitializedDriverAsync(options: options);
client.SampleFailure = new InvalidDataException("MTConnectError OUT_OF_RANGE: 'count' must be less than 65 (served under HTTP 400)");
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
try
{
// Well past the cap, so the hold across LATER re-baselines is what is asserted, not a lucky instant.
await client.WaitForSampleCallsAsync(MTConnectDriver.MaxRebaselinesWithoutProgress + 3).WaitAsync(Watchdog, Ct);
var stalledAt = driver.GetHealth();
stalledAt.State.ShouldBe(DriverState.Degraded);
stalledAt.LastError.ShouldBe(MTConnectDriver.StalledStreamReason);
stalledAt.LastSuccessfulRead.ShouldNotBeNull();
// LastSuccessfulRead keeps advancing while it is stalled: the /current answers are real.
await client.WaitForSampleCallsAsync(client.SampleCallCount + 2).WaitAsync(Watchdog, Ct);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
driver.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThanOrEqualTo(stalledAt.LastSuccessfulRead!.Value);
// Positive control: the Agent starts streaming again — one delivered chunk clears the verdict.
client.SampleFailure = null;
await client.PumpAsync(Chunk(PrimedNextSequence, PrimedNextSequence + 1, ("dev1_pos", "9.5")))
.WaitAsync(Watchdog, Ct);
await WaitForHealthAsync(driver, DriverState.Healthy, Ct);
}
finally
{
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
}
}
/// <summary>
/// …the falsification control: FEWER than the cap is an ordinary recovery and must stay Healthy (a genuine
/// ring-buffer overrun is resolved by one re-baseline).
/// </summary>
[Fact]
public async Task A_rebaseline_short_of_the_cap_stays_healthy()
{
var (driver, client) = await InitializedDriverAsync();
client.SampleFailures.Enqueue(new InvalidDataException("OUT_OF_RANGE (served under HTTP 400)"));
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
try
{
await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
await client.PumpAsync(Chunk(PrimedNextSequence, PrimedNextSequence + 1, ("dev1_pos", "1.5")))
.WaitAsync(Watchdog, Ct);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
}
finally
{
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
}
}
private static async Task WaitForHealthAsync(MTConnectDriver driver, DriverState state, CancellationToken ct)
{
using var watchdog = CancellationTokenSource.CreateLinkedTokenSource(ct);
watchdog.CancelAfter(Watchdog);
while (driver.GetHealth().State != state)
{
await Task.Delay(5, watchdog.Token).ConfigureAwait(false);
}
}
/// <summary>
/// <b>I3 — a non-positive cap is an operator-authorable hot loop.</b> Clamping to it (the
/// obvious <c>Math.Max(0, …)</c>) makes EVERY delay zero, so the pump reconnect-spins as fast