fix(mtconnect): make a /sample stream end loud, and correct the gap contract (Task 7 review)
Code review of c46540ae approved the framing algorithm but moved the risk onto the contract around it. C1 (critical). SampleAsync returned normally on three different events: EOF, the closing --boundary--, and a response that was not multipart at all (where it yielded one parsed snapshot and finished). The seam documents the stream as yielding until ct is cancelled, so a Task 11 pump written to that contract would silently drop its subscription or hot-loop reconnecting; the non-multipart leg reported a configuration error as a healthy finished stream, which is the #485 quiet-successful-termination shape aimed at the very next task. Every non-cancellation end now throws: MTConnectStreamEndedException (ConnectionClosed / ClosingBoundary - transient) or MTConnectStreamNotSupportedException (configuration - reconnecting reproduces it forever), sharing one catchable base. The non-multipart body is still parsed first so an Agent's own MTConnectError text wins, but the document is NOT yielded. I1. IsSequenceGap moved to IMTConnectAgentClient (static) and its first parameter is now expectedFrom. The old name and doc were wrong for every chunk after the first - the correct comparand is the PREVIOUS chunk's NextSequence, and comparing against the opening "from" argument reports a gap on every chunk once the ring buffer rolls, i.e. an endless /current re-baseline storm. Both doc blocks also now record that cppagent answers from < firstSequence with an OUT_OF_RANGE MTConnectError under HTTP 200, which surfaces as InvalidDataException and NOT as a gap - so Task 11 must treat that parse failure as a re-baseline trigger too. I2. Every multipart test served the whole body from one buffer, so every framing test completed in a single ReadAsync - the split-boundary path, the split-header path and the multi-fill loop were correct by inspection only. A ChunkedStream double (N bytes per read, N = 1/3/7) now drives the fixtures through a split transport, plus a >16 KB document that forces a buffer resize. Falsifiability: removing either scan overlap, or collapsing the fill loop, fails ONLY the new chunked tests. I3. Disposal mid-enumeration surfaces as OperationCanceledException via an internal dispose-linked token, not a raw ObjectDisposedException that Task 9's re-init would inflict on an enumerating pump. I5. HeartbeatMs, SampleIntervalMs and SampleCount join RequestTimeoutMs in construction-time positive-value validation. All three reach the query string and an Agent answers heartbeat=0 with an HTTP-200 MTConnectError that names no config key. I4/M2. The no-Content-length framing fallback logs a one-shot warning (the client takes an optional ILogger); a multipart response with no boundary parameter fails fast instead of degrading into a watchdog timeout. M5/M6. Shared element/attribute reading rules extracted to MTConnectXml; the probe parser documents why it alone does not trim BOM/whitespace. The literal U+FEFF in source became a '' escape. Also, from Task 8: MTConnectObservation gained IsStructured (default false), set from element.HasElements. A DATA_SET/TABLE observation's <Entry> children concatenate through element.Value to nonsense ("12" for two entries) that the index would publish as Good, and only the parser can tell that apart from a legitimate space-bearing Message. False for CONDITION observations - their value comes from the element name, so child content cannot corrupt it. 275/275 green in this suite's own files.
This commit is contained in:
Joseph Doherty
+522
-15
@@ -1,7 +1,9 @@
|
||||
using System.Diagnostics;
|
||||
using System.Globalization;
|
||||
using System.Net;
|
||||
using System.Net.Http.Headers;
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
@@ -10,7 +12,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
|
||||
/// <summary>
|
||||
/// Task 7 — the <c>/current</c> + <c>/sample</c> legs: <see cref="MTConnectStreamsParser"/>
|
||||
/// turning an <c>MTConnectStreams</c> document into <see cref="MTConnectStreamsResult"/>, the
|
||||
/// ring-buffer sequence-gap signal (<see cref="MTConnectAgentClient.IsSequenceGap"/>), and the
|
||||
/// ring-buffer sequence-gap signal (<see cref="IMTConnectAgentClient.IsSequenceGap"/>), and the
|
||||
/// <c>multipart/x-mixed-replace</c> chunk framing + heartbeat watchdog on the long-poll leg.
|
||||
/// Every test runs against canned fixtures or a stubbed <see cref="HttpMessageHandler"/> — no
|
||||
/// socket is opened anywhere in this file.
|
||||
@@ -179,7 +181,7 @@ public sealed class MTConnectStreamsParseTests
|
||||
{
|
||||
var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
|
||||
|
||||
MTConnectAgentClient.IsSequenceGap(requestedFrom: 1, chunk).ShouldBeTrue();
|
||||
IMTConnectAgentClient.IsSequenceGap(expectedFrom: 1, chunk).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -188,18 +190,18 @@ public sealed class MTConnectStreamsParseTests
|
||||
// The falsifiability leg: a helper hard-wired to true would pass the gap test alone.
|
||||
var contiguous = MTConnectStreamsParser.Parse(File.ReadAllText(SampleFixture));
|
||||
|
||||
MTConnectAgentClient.IsSequenceGap(requestedFrom: 108, contiguous).ShouldBeFalse();
|
||||
IMTConnectAgentClient.IsSequenceGap(expectedFrom: 108, contiguous).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(4999L, true)] // requested older than the buffer holds — data was lost
|
||||
[InlineData(4999L, true)] // expected older than the buffer holds — data was lost
|
||||
[InlineData(5000L, false)] // exactly the oldest retained sequence — nothing was lost
|
||||
[InlineData(5001L, false)] // the agent simply had nothing older to send
|
||||
public void Sequence_gap_is_strictly_firstSequence_greater_than_requested_from(long requestedFrom, bool expected)
|
||||
public void Sequence_gap_is_strictly_firstSequence_greater_than_expected_from(long expectedFrom, bool expected)
|
||||
{
|
||||
var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
|
||||
|
||||
MTConnectAgentClient.IsSequenceGap(requestedFrom, chunk).ShouldBe(expected);
|
||||
IMTConnectAgentClient.IsSequenceGap(expectedFrom, chunk).ShouldBe(expected);
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------- legal but degenerate documents
|
||||
@@ -499,11 +501,17 @@ public sealed class MTConnectStreamsParseTests
|
||||
|
||||
var from = 108L;
|
||||
var gaps = new List<bool>();
|
||||
await foreach (var chunk in client.SampleAsync(from, Ct))
|
||||
await Should.ThrowAsync<MTConnectStreamEndedException>(async () =>
|
||||
{
|
||||
gaps.Add(MTConnectAgentClient.IsSequenceGap(from, chunk));
|
||||
from = chunk.NextSequence;
|
||||
}
|
||||
await foreach (var chunk in client.SampleAsync(from, Ct))
|
||||
{
|
||||
// `from` is the RUNNING cursor, not the original argument — comparing every chunk
|
||||
// against the opening `from` would report a gap on every chunk once the Agent's
|
||||
// buffer rolls past it (I1).
|
||||
gaps.Add(IMTConnectAgentClient.IsSequenceGap(from, chunk));
|
||||
from = chunk.NextSequence;
|
||||
}
|
||||
});
|
||||
|
||||
gaps.ShouldBe([false, true]);
|
||||
from.ShouldBe(5005);
|
||||
@@ -541,7 +549,7 @@ public sealed class MTConnectStreamsParseTests
|
||||
var chunks = await DrainAsync(client.SampleAsync(113, Ct));
|
||||
|
||||
chunks.ShouldHaveSingleItem().Observations.ShouldBeEmpty();
|
||||
MTConnectAgentClient.IsSequenceGap(113, chunks[0]).ShouldBeFalse();
|
||||
IMTConnectAgentClient.IsSequenceGap(113, chunks[0]).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact(Timeout = 30_000)]
|
||||
@@ -623,14 +631,107 @@ public sealed class MTConnectStreamsParseTests
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_ends_cleanly_when_the_agent_closes_the_stream()
|
||||
public async Task SampleAsync_reports_the_closing_boundary_rather_than_ending_quietly()
|
||||
{
|
||||
// C1. The seam contracts SampleAsync to run until cancelled, so a pump has no reason to
|
||||
// handle normal completion: returning here would either drop the subscription silently or
|
||||
// hot-loop reconnecting on an enumerator that ends immediately (#485).
|
||||
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
|
||||
var chunks = new List<MTConnectStreamsResult>();
|
||||
var ex = await Should.ThrowAsync<MTConnectStreamEndedException>(async () =>
|
||||
{
|
||||
await foreach (var chunk in client.SampleAsync(108, Ct))
|
||||
{
|
||||
chunks.Add(chunk);
|
||||
}
|
||||
});
|
||||
|
||||
chunks.ShouldHaveSingleItem();
|
||||
ex.Reason.ShouldBe(MTConnectStreamEndReason.ClosingBoundary);
|
||||
ex.ChunksDelivered.ShouldBe(1);
|
||||
ex.AgentUri.ShouldContain("/sample?from=108");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_reports_a_connection_that_drops_mid_stream()
|
||||
{
|
||||
// No closing boundary — the body simply stops. Distinguished from a deliberate close so an
|
||||
// operator can tell a flaky network from an Agent honouring a bounded request.
|
||||
var truncated = StubHandler.MultipartBody(
|
||||
"bnd", withContentLength: true, closing: false, await File.ReadAllTextAsync(SampleFixture, Ct));
|
||||
var handler = StubHandler.RespondingWithBody("bnd", truncated);
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var ex = await Should.ThrowAsync<MTConnectStreamEndedException>(
|
||||
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
|
||||
|
||||
ex.Reason.ShouldBe(MTConnectStreamEndReason.ConnectionClosed);
|
||||
ex.ChunksDelivered.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_reports_a_non_multipart_answer_as_a_configuration_error()
|
||||
{
|
||||
// The worst C1 shape: the Agent ignored the streaming request entirely. Yielding the one
|
||||
// snapshot it did return and finishing would report a dead subscription as a healthy one.
|
||||
var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var ex = await Should.ThrowAsync<MTConnectStreamNotSupportedException>(
|
||||
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
|
||||
|
||||
ex.Message.ShouldContain("multipart/x-mixed-replace");
|
||||
ex.Message.ShouldContain("text/xml");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_yields_nothing_at_all_from_a_non_multipart_answer()
|
||||
{
|
||||
// Explicitly pinned: the parsed document must NOT reach the caller. One snapshot plus a
|
||||
// finished stream is precisely how a dead data path disguises itself as a working one.
|
||||
var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var chunks = new List<MTConnectStreamsResult>();
|
||||
await Should.ThrowAsync<MTConnectStreamNotSupportedException>(async () =>
|
||||
{
|
||||
await foreach (var chunk in client.SampleAsync(108, Ct))
|
||||
{
|
||||
chunks.Add(chunk);
|
||||
}
|
||||
});
|
||||
|
||||
chunks.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_fails_fast_when_a_multipart_response_carries_no_boundary_parameter()
|
||||
{
|
||||
// M2. Falling through to whole-body reading would surface only as a watchdog TimeoutException
|
||||
// much later, pointing an operator at the network instead of at the malformed header.
|
||||
var handler = StubHandler.RespondingWithBoundarylessMultipart();
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var ex = await Should.ThrowAsync<MTConnectStreamNotSupportedException>(
|
||||
async () => await EnumerateAsync(client.SampleAsync(1, Ct)));
|
||||
|
||||
ex.Message.ShouldContain("boundary");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_stream_end_and_a_stream_misconfiguration_share_one_catchable_base_type()
|
||||
{
|
||||
// A pump that only wants "the stream is over" should not have to name both.
|
||||
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var ended = await Should.ThrowAsync<MTConnectStreamEndedException>(
|
||||
async () => await EnumerateAsync(client.SampleAsync(108, Ct)));
|
||||
|
||||
ended.ShouldBeAssignableTo<MTConnectStreamException>();
|
||||
typeof(MTConnectStreamException).IsAssignableFrom(typeof(MTConnectStreamNotSupportedException)).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -644,6 +745,255 @@ public sealed class MTConnectStreamsParseTests
|
||||
handler.RequestCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
// ------------------------------------------- I2: framing when the transport splits every read
|
||||
|
||||
/// <summary>
|
||||
/// Serving the whole body from one buffer completes every frame in a single
|
||||
/// <c>ReadAsync</c>, so the split-boundary path, the split-header path and the multi-fill
|
||||
/// loop are never entered — on the one behaviour whose headline risk is framing. These drive
|
||||
/// the same fixtures through a transport that hands back only N bytes at a time; at N = 1
|
||||
/// every delimiter, every header line and every document is split maximally.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(1)]
|
||||
[InlineData(3)]
|
||||
[InlineData(7)]
|
||||
public async Task SampleAsync_frames_correctly_when_the_transport_splits_every_boundary_and_header(int bytesPerRead)
|
||||
{
|
||||
var handler = StubHandler.RespondingWithChunkedMultipart(
|
||||
"--------bnd",
|
||||
bytesPerRead,
|
||||
withContentLength: true,
|
||||
await File.ReadAllTextAsync(SampleFixture, Ct),
|
||||
await File.ReadAllTextAsync(GapFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
|
||||
|
||||
chunks.Select(c => c.FirstSequence).ShouldBe([108L, 5000L]);
|
||||
chunks.Select(c => c.NextSequence).ShouldBe([113L, 5005L]);
|
||||
chunks[0].Observations.Count.ShouldBe(5);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(1)]
|
||||
[InlineData(3)]
|
||||
[InlineData(7)]
|
||||
public async Task SampleAsync_boundary_scan_framing_survives_a_split_transport_too(int bytesPerRead)
|
||||
{
|
||||
var handler = StubHandler.RespondingWithChunkedMultipart(
|
||||
"bnd",
|
||||
bytesPerRead,
|
||||
withContentLength: false,
|
||||
await File.ReadAllTextAsync(SampleFixture, Ct),
|
||||
await File.ReadAllTextAsync(GapFixture, Ct));
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var chunks = await DrainAsync(client.SampleAsync(108, Ct));
|
||||
|
||||
chunks.Select(c => c.FirstSequence).ShouldBe([108L, 5000L]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SampleAsync_grows_its_buffer_for_a_part_larger_than_the_initial_read_buffer()
|
||||
{
|
||||
// Forces Array.Resize + a Consume across the grown buffer, which a 2 KB fixture never does.
|
||||
var large = LargeStreamsDocument(observationCount: 900);
|
||||
Encoding.UTF8.GetByteCount(large).ShouldBeGreaterThan(16 * 1024);
|
||||
|
||||
var handler = StubHandler.RespondingWithChunkedMultipart(
|
||||
"bnd", bytesPerRead: 1024, withContentLength: true, large, large);
|
||||
using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
var chunks = await DrainAsync(client.SampleAsync(1, Ct));
|
||||
|
||||
chunks.Count.ShouldBe(2);
|
||||
chunks[0].Observations.Count.ShouldBe(900);
|
||||
chunks[1].Observations.Count.ShouldBe(900);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------- I3: disposal during a live enumeration
|
||||
|
||||
[Fact(Timeout = 30_000)]
|
||||
public async Task Disposing_the_client_mid_enumeration_surfaces_as_cancellation()
|
||||
{
|
||||
// Task 9 re-initializes by disposing and rebuilding this client, which can happen while a
|
||||
// pump is still enumerating. An unclassified ObjectDisposedException/IOException from the
|
||||
// socket is something no caller is written to expect.
|
||||
var prefix = StubHandler.MultipartBody(
|
||||
"bnd", withContentLength: true, closing: false, await File.ReadAllTextAsync(SampleFixture, Ct));
|
||||
var handler = StubHandler.RespondingWithStallingStream("bnd", prefix);
|
||||
var client = NewClient(handler, new MTConnectDriverOptions
|
||||
{
|
||||
AgentUri = "http://agent:5000",
|
||||
RequestTimeoutMs = 5000,
|
||||
HeartbeatMs = 20_000
|
||||
});
|
||||
|
||||
await using var chunks = client.SampleAsync(108, Ct).GetAsyncEnumerator(Ct);
|
||||
(await chunks.MoveNextAsync()).ShouldBeTrue();
|
||||
|
||||
var pending = chunks.MoveNextAsync().AsTask();
|
||||
client.Dispose();
|
||||
|
||||
var ex = await Should.ThrowAsync<OperationCanceledException>(async () => await pending);
|
||||
ex.ShouldNotBeOfType<ObjectDisposedException>();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Using_a_disposed_client_throws_object_disposed_rather_than_dialling()
|
||||
{
|
||||
var handler = StubHandler.Hanging();
|
||||
var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
client.Dispose();
|
||||
|
||||
await Should.ThrowAsync<ObjectDisposedException>(async () => await client.CurrentAsync(Ct));
|
||||
await Should.ThrowAsync<ObjectDisposedException>(
|
||||
async () => await EnumerateAsync(client.SampleAsync(1, Ct)));
|
||||
handler.RequestCount.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Dispose_is_idempotent()
|
||||
{
|
||||
var client = NewClient(StubHandler.Hanging(), new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
|
||||
|
||||
client.Dispose();
|
||||
|
||||
Should.NotThrow(client.Dispose);
|
||||
}
|
||||
|
||||
// ------------------------------------------- I4: the degraded framing mode is operator-visible
|
||||
|
||||
[Fact]
|
||||
public async Task A_part_without_content_length_warns_once_that_framing_is_degraded()
|
||||
{
|
||||
var logger = new RecordingLogger();
|
||||
var handler = StubHandler.RespondingWithMultipart(
|
||||
"bnd",
|
||||
withContentLength: false,
|
||||
await File.ReadAllTextAsync(SampleFixture, Ct),
|
||||
await File.ReadAllTextAsync(GapFixture, Ct));
|
||||
using var client = new MTConnectAgentClient(
|
||||
new MTConnectDriverOptions { AgentUri = "http://agent:5000" }, handler, logger);
|
||||
|
||||
await DrainAsync(client.SampleAsync(108, Ct));
|
||||
|
||||
// Once, not once per part — a per-chunk warning on a long poll is its own outage.
|
||||
logger.Warnings.Count.ShouldBe(1);
|
||||
logger.Warnings[0].ShouldContain("Content-length");
|
||||
logger.Warnings[0].ShouldContain("http://agent:5000/sample");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_normal_content_length_path_warns_about_nothing()
|
||||
{
|
||||
var logger = new RecordingLogger();
|
||||
var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
|
||||
using var client = new MTConnectAgentClient(
|
||||
new MTConnectDriverOptions { AgentUri = "http://agent:5000" }, handler, logger);
|
||||
|
||||
await DrainAsync(client.SampleAsync(108, Ct));
|
||||
|
||||
logger.Warnings.ShouldBeEmpty();
|
||||
}
|
||||
|
||||
// ---------------------------------------- I5: operator-authorable options that reach the wire
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-1)]
|
||||
public void Ctor_rejects_a_non_positive_heartbeat(int heartbeatMs)
|
||||
{
|
||||
// heartbeat=0 goes on the query string, and an Agent answers it with an MTConnectError
|
||||
// under HTTP 200 — surfacing as a parse failure that never names the offending key.
|
||||
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
|
||||
new MTConnectDriverOptions { AgentUri = "http://agent:5000", HeartbeatMs = heartbeatMs }));
|
||||
|
||||
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.HeartbeatMs));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-1)]
|
||||
public void Ctor_rejects_a_non_positive_sample_interval(int sampleIntervalMs)
|
||||
{
|
||||
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
|
||||
new MTConnectDriverOptions { AgentUri = "http://agent:5000", SampleIntervalMs = sampleIntervalMs }));
|
||||
|
||||
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.SampleIntervalMs));
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-1)]
|
||||
public void Ctor_rejects_a_non_positive_sample_count(int sampleCount)
|
||||
{
|
||||
var ex = Should.Throw<ArgumentOutOfRangeException>(() => new MTConnectAgentClient(
|
||||
new MTConnectDriverOptions { AgentUri = "http://agent:5000", SampleCount = sampleCount }));
|
||||
|
||||
ex.Message.ShouldContain(nameof(MTConnectDriverOptions.SampleCount));
|
||||
}
|
||||
|
||||
// ------------------------------------------- structured (DATA_SET / TABLE) observation signal
|
||||
|
||||
[Fact]
|
||||
public void An_observation_whose_content_is_child_entries_is_flagged_structured()
|
||||
{
|
||||
// <Entry> children concatenate to "12" through element.Value — keys gone, values run
|
||||
// together. The index codes this BadNotSupported rather than publishing noise as Good.
|
||||
var result = MTConnectStreamsParser.Parse(StreamsDocument(
|
||||
"""
|
||||
<Events><PathFeedrateOverride dataItemId="ds" timestamp="2026-07-24T12:00:00Z" count="2">
|
||||
<Entry key="V1">1</Entry><Entry key="V2">2</Entry>
|
||||
</PathFeedrateOverride></Events>
|
||||
"""));
|
||||
|
||||
var observation = result.Observations.ShouldHaveSingleItem();
|
||||
observation.IsStructured.ShouldBeTrue();
|
||||
observation.DataItemId.ShouldBe("ds");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_plain_text_observation_containing_spaces_is_NOT_flagged_structured()
|
||||
{
|
||||
// The load-bearing case: this proves the signal is "the element had element children", not
|
||||
// a whitespace heuristic. A Message reading "Coolant level low" is perfectly valid data and
|
||||
// is textually indistinguishable from concatenated entries.
|
||||
var result = MTConnectStreamsParser.Parse(StreamsDocument(
|
||||
"""<Events><Message dataItemId="msg" timestamp="2026-07-24T12:00:00Z">Coolant level low</Message></Events>"""));
|
||||
|
||||
var observation = result.Observations.ShouldHaveSingleItem();
|
||||
observation.IsStructured.ShouldBeFalse();
|
||||
observation.Value.ShouldBe("Coolant level low");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Every_observation_in_the_committed_fixtures_is_unstructured()
|
||||
{
|
||||
// Includes the space-separated TIME_SERIES vector, which is text and must stay unflagged.
|
||||
var current = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
|
||||
|
||||
current.Observations.ShouldAllBe(o => !o.IsStructured);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void A_condition_is_never_flagged_structured_even_with_child_content()
|
||||
{
|
||||
// A condition's value comes from the element NAME, so child content cannot corrupt it.
|
||||
// Flagging one would throw away a perfectly well-determined state.
|
||||
var result = MTConnectStreamsParser.Parse(StreamsDocument(
|
||||
"""
|
||||
<Condition><Fault dataItemId="c" timestamp="2026-07-24T12:00:00Z" type="SYSTEM">
|
||||
<Detail>vendor extension</Detail>
|
||||
</Fault></Condition>
|
||||
"""));
|
||||
|
||||
var observation = result.Observations.ShouldHaveSingleItem();
|
||||
observation.IsStructured.ShouldBeFalse();
|
||||
observation.Value.ShouldBe("Fault");
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------------------------ helpers
|
||||
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
@@ -654,14 +1004,51 @@ public sealed class MTConnectStreamsParseTests
|
||||
private static async Task<List<MTConnectStreamsResult>> DrainAsync(IAsyncEnumerable<MTConnectStreamsResult> source)
|
||||
{
|
||||
var results = new List<MTConnectStreamsResult>();
|
||||
await foreach (var item in source)
|
||||
try
|
||||
{
|
||||
results.Add(item);
|
||||
await foreach (var item in source)
|
||||
{
|
||||
results.Add(item);
|
||||
}
|
||||
}
|
||||
catch (MTConnectStreamEndedException)
|
||||
{
|
||||
// Expected under the C1 contract: every non-cancellation end of a /sample stream is
|
||||
// loud. Tests that care about HOW it ended assert on the exception directly.
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
/// <summary>Enumerates to completion, discarding chunks and letting every exception escape.</summary>
|
||||
private static async Task EnumerateAsync(IAsyncEnumerable<MTConnectStreamsResult> source)
|
||||
{
|
||||
await foreach (var _ in source)
|
||||
{
|
||||
// Drained for effect only.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A streams document big enough to force the frame reader to grow its buffer.</summary>
|
||||
private static string LargeStreamsDocument(int observationCount)
|
||||
{
|
||||
var samples = new StringBuilder();
|
||||
for (var i = 0; i < observationCount; i++)
|
||||
{
|
||||
samples.Append(CultureInfo.InvariantCulture,
|
||||
$"""<Position dataItemId="d{i}" timestamp="2026-07-24T12:00:00Z" sequence="{i + 1}">{i}.5</Position>""");
|
||||
}
|
||||
|
||||
return $"""
|
||||
<MTConnectStreams xmlns="urn:mtconnect.org:MTConnectStreams:1.3">
|
||||
<Header instanceId="1655000000" firstSequence="1" lastSequence="{observationCount}" nextSequence="{observationCount + 1}"/>
|
||||
<Streams><DeviceStream name="D" uuid="d"><ComponentStream component="Path" componentId="p">
|
||||
<Samples>{samples}</Samples>
|
||||
</ComponentStream></DeviceStream></Streams>
|
||||
</MTConnectStreams>
|
||||
""";
|
||||
}
|
||||
|
||||
private static Dictionary<string, MTConnectObservation> ParseCurrent() => ParseFixture(CurrentFixture);
|
||||
|
||||
private static Dictionary<string, MTConnectObservation> ParseFixture(string path) =>
|
||||
@@ -717,6 +1104,35 @@ public sealed class MTConnectStreamsParseTests
|
||||
public static StubHandler RespondingWithStallingStream(string boundary, byte[] prefix) =>
|
||||
new(() => Multipart(boundary, new StallingContent(prefix)), hang: false);
|
||||
|
||||
/// <summary>Serves an already-built multipart body verbatim (e.g. one with no closing delimiter).</summary>
|
||||
public static StubHandler RespondingWithBody(string boundary, byte[] body) =>
|
||||
new(() => Multipart(boundary, new ByteArrayContent(body)), hang: false);
|
||||
|
||||
/// <summary>
|
||||
/// Serves a multipart body through a transport that hands back at most
|
||||
/// <paramref name="bytesPerRead"/> bytes per <c>ReadAsync</c>, so boundaries, part
|
||||
/// headers and documents are all split across reads.
|
||||
/// </summary>
|
||||
public static StubHandler RespondingWithChunkedMultipart(
|
||||
string boundary, int bytesPerRead, bool withContentLength, params string[] parts)
|
||||
{
|
||||
var body = MultipartBody(boundary, withContentLength, closing: true, parts);
|
||||
|
||||
return new StubHandler(() => Multipart(boundary, new ChunkedContent(body, bytesPerRead)), hang: false);
|
||||
}
|
||||
|
||||
/// <summary>A response that claims multipart but omits the <c>boundary</c> parameter.</summary>
|
||||
public static StubHandler RespondingWithBoundarylessMultipart() =>
|
||||
new(
|
||||
() =>
|
||||
{
|
||||
var content = new ByteArrayContent([]);
|
||||
content.Headers.ContentType = new MediaTypeHeaderValue("multipart/x-mixed-replace");
|
||||
|
||||
return new HttpResponseMessage(HttpStatusCode.OK) { Content = content };
|
||||
},
|
||||
hang: false);
|
||||
|
||||
/// <summary>Builds a <c>multipart/x-mixed-replace</c> body exactly as an Agent writes one.</summary>
|
||||
public static byte[] MultipartBody(string boundary, bool withContentLength, bool closing, params string[] parts)
|
||||
{
|
||||
@@ -769,6 +1185,97 @@ public sealed class MTConnectStreamsParseTests
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Hands back at most <c>bytesPerRead</c> bytes per read — the split-transport double.</summary>
|
||||
private sealed class ChunkedContent(byte[] body, int bytesPerRead) : HttpContent
|
||||
{
|
||||
protected override Task<Stream> CreateContentReadStreamAsync() =>
|
||||
Task.FromResult<Stream>(new ChunkedStream(body, bytesPerRead));
|
||||
|
||||
protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) =>
|
||||
stream.WriteAsync(body, 0, body.Length);
|
||||
|
||||
protected override bool TryComputeLength(out long length)
|
||||
{
|
||||
length = body.Length;
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class ChunkedStream(byte[] body, int bytesPerRead) : Stream
|
||||
{
|
||||
private int _position;
|
||||
|
||||
public override bool CanRead => true;
|
||||
|
||||
public override bool CanSeek => false;
|
||||
|
||||
public override bool CanWrite => false;
|
||||
|
||||
public override long Length => body.Length;
|
||||
|
||||
public override long Position
|
||||
{
|
||||
get => _position;
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
if (_position >= body.Length)
|
||||
{
|
||||
return ValueTask.FromResult(0);
|
||||
}
|
||||
|
||||
var count = Math.Min(Math.Min(bytesPerRead, buffer.Length), body.Length - _position);
|
||||
body.AsMemory(_position, count).CopyTo(buffer);
|
||||
_position += count;
|
||||
|
||||
return ValueTask.FromResult(count);
|
||||
}
|
||||
|
||||
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
|
||||
ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask();
|
||||
|
||||
public override void Flush()
|
||||
{
|
||||
}
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
/// <summary>Captures warning-level log lines so the degraded-framing notice can be asserted.</summary>
|
||||
private sealed class RecordingLogger : ILogger
|
||||
{
|
||||
public List<string> Warnings { get; } = [];
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state)
|
||||
where TState : notnull => null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(
|
||||
LogLevel logLevel,
|
||||
EventId eventId,
|
||||
TState state,
|
||||
Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
if (logLevel == LogLevel.Warning)
|
||||
{
|
||||
Warnings.Add(formatter(state, exception));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Serves <paramref name="prefix"/>, then never completes another read.</summary>
|
||||
private sealed class StallingContent(byte[] prefix) : HttpContent
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user