diff --git a/Directory.Packages.props b/Directory.Packages.props
index 3b622ce4..c9ad0acf 100644
--- a/Directory.Packages.props
+++ b/Directory.Packages.props
@@ -84,11 +84,11 @@
ships net6.0/net8.0 only.
-->
-
-
-
+
-
-
-
+
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectProbeParseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectProbeParseTests.cs
index 019758f0..1d285d5c 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectProbeParseTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectProbeParseTests.cs
@@ -196,6 +196,49 @@ public sealed class MTConnectProbeParseTests
device.Components[0].Components[0].Components[0].Id.ShouldBe("m");
}
+ [Fact]
+ public void Probe_parse_keeps_a_vendor_extension_component_in_its_own_namespace()
+ {
+ // The shape the parser's ignore-the-namespace rule exists for, and the reason attributes are
+ // read unqualified: the vendor component carries its OWN namespace prefix, while its
+ // standard / children inherit the DEFAULT MTConnect namespace. Matching
+ // components on a fully-qualified XName would silently drop and every data item
+ // beneath it, leaving the browse tree short with no error anywhere.
+ const string Xml = """
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ """;
+
+ var device = MTConnectProbeParser.Parse(Xml).Devices[0];
+
+ var pump = device.Components.ShouldHaveSingleItem();
+ pump.Id.ShouldBe("pump1");
+ pump.Name.ShouldBe("pump");
+ pump.Components.ShouldHaveSingleItem().Id.ShouldBe("imp1");
+ device.AllDataItems().Select(d => d.Id).ShouldBe(["d_avail", "pump_pressure", "imp_rpm"]);
+ device.AllDataItems().Single(d => d.Id == "pump_pressure").Units.ShouldBe("PASCAL");
+ }
+
[Fact]
public void Probe_parse_reads_every_device_of_a_multi_device_agent()
{
@@ -408,27 +451,6 @@ public sealed class MTConnectProbeParseTests
handler.RequestCount.ShouldBe(0);
}
- // ------------------------------------------------------------------- scope boundary: Task 7 legs
-
- [Fact]
- public async Task CurrentAsync_is_not_implemented_until_task_7()
- {
- using var client = new MTConnectAgentClient(new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
-
- var ex = await Should.ThrowAsync(
- async () => await client.CurrentAsync(TestContext.Current.CancellationToken));
- ex.Message.ShouldContain("Task 7");
- }
-
- [Fact]
- public void SampleAsync_is_not_implemented_until_task_7()
- {
- using var client = new MTConnectAgentClient(new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
-
- var ex = Should.Throw(() => client.SampleAsync(1, CancellationToken.None));
- ex.Message.ShouldContain("Task 7");
- }
-
private static Task ReadFixtureAsync() =>
File.ReadAllTextAsync(ProbeFixturePath, TestContext.Current.CancellationToken);
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs
new file mode 100644
index 00000000..089aa83d
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs
@@ -0,0 +1,838 @@
+using System.Diagnostics;
+using System.Net;
+using System.Net.Http.Headers;
+using System.Text;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
+
+///
+/// Task 7 — the /current + /sample legs:
+/// turning an MTConnectStreams document into , the
+/// ring-buffer sequence-gap signal (), and the
+/// multipart/x-mixed-replace chunk framing + heartbeat watchdog on the long-poll leg.
+/// Every test runs against canned fixtures or a stubbed — no
+/// socket is opened anywhere in this file.
+///
+public sealed class MTConnectStreamsParseTests
+{
+ private const string CurrentFixture = "Fixtures/current.xml";
+ private const string SampleFixture = "Fixtures/sample.xml";
+ private const string GapFixture = "Fixtures/sample-gap.xml";
+ private const string UnavailableFixture = "Fixtures/current-unavailable.xml";
+
+ /// The fixtures deliberately share one instanceId, so a gap — not a restart — is under test.
+ private const long FixtureInstanceId = 1655000000L;
+
+ private static readonly string[] AllFixtureDataItemIds =
+ [
+ "dev1_avail",
+ "dev1_pos",
+ "dev1_vibration_ts",
+ "dev1_partcount",
+ "dev1_execution",
+ "dev1_program",
+ "dev1_system_cond"
+ ];
+
+ // ------------------------------------------------------------------ header sequence parsing
+
+ [Fact]
+ public void Current_parse_reads_header_sequences_and_observations()
+ {
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
+
+ result.NextSequence.ShouldBeGreaterThan(0);
+ result.Observations.ShouldNotBeEmpty();
+ }
+
+ [Theory]
+ [InlineData(CurrentFixture, 1L, 108L)]
+ [InlineData(SampleFixture, 108L, 113L)]
+ [InlineData(GapFixture, 5000L, 5005L)]
+ [InlineData(UnavailableFixture, 200L, 207L)]
+ public void Parse_reads_first_and_next_sequence_and_the_instance_id_from_every_fixture(
+ string fixture, long firstSequence, long nextSequence)
+ {
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(fixture));
+
+ result.FirstSequence.ShouldBe(firstSequence);
+ result.NextSequence.ShouldBe(nextSequence);
+ result.InstanceId.ShouldBe(FixtureInstanceId);
+ }
+
+ // ------------------------------------------------------------------------ observation values
+
+ [Fact]
+ public void Current_parse_keys_every_observation_by_its_dataitem_id()
+ {
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
+
+ result.Observations.Select(o => o.DataItemId).ShouldBe(AllFixtureDataItemIds, ignoreOrder: true);
+ }
+
+ [Fact]
+ public void Current_parse_reads_the_good_value_the_agent_reported()
+ {
+ var observations = ParseCurrent();
+
+ // The observation ELEMENT is , not ; the value is its text content.
+ observations["dev1_pos"].Value.ShouldBe("123.4567");
+ observations["dev1_execution"].Value.ShouldBe("ACTIVE");
+ observations["dev1_program"].Value.ShouldBe("O1234");
+ observations["dev1_avail"].Value.ShouldBe("AVAILABLE");
+ }
+
+ [Fact]
+ public void Current_parse_carries_the_unavailable_text_sentinel_through_verbatim()
+ {
+ // Task 8 maps UNAVAILABLE => BadNoCommunication; this layer must not pre-interpret it.
+ ParseCurrent()["dev1_partcount"].Value.ShouldBe("UNAVAILABLE");
+ }
+
+ [Fact]
+ public void Current_parse_keeps_a_time_series_vector_as_its_raw_string()
+ {
+ ParseCurrent()["dev1_vibration_ts"].Value.ShouldBe("1.1 1.2 1.3 1.4 1.5 1.6 1.7 1.8 1.9 2.0");
+ }
+
+ // ---------------------------------------------------------------------- CONDITION observations
+
+ [Fact]
+ public void A_condition_observations_value_is_its_element_name_not_its_empty_text()
+ {
+ // — the element name IS the state, and the
+ // element is empty, so a text-content read yields "" and loses the observation entirely.
+ ParseCurrent()["dev1_system_cond"].Value.ShouldBe("Normal");
+ }
+
+ [Theory]
+ [InlineData("Normal", "Normal")]
+ [InlineData("Warning", "Warning")]
+ [InlineData("Fault", "Fault")]
+ public void A_condition_state_element_name_becomes_the_value(string element, string expected)
+ {
+ var result = MTConnectStreamsParser.Parse(StreamsDocument(
+ $"""<{element} dataItemId="c" timestamp="2026-07-24T12:00:00Z" sequence="1" type="SYSTEM"/>"""));
+
+ result.Observations.ShouldHaveSingleItem().Value.ShouldBe(expected);
+ }
+
+ [Fact]
+ public void A_condition_with_text_content_still_takes_its_value_from_the_element_name()
+ {
+ // Agents put the operator-facing condition message in the element text; the STATE is the name.
+ var result = MTConnectStreamsParser.Parse(StreamsDocument(
+ """Spindle overtemperature"""));
+
+ result.Observations.ShouldHaveSingleItem().Value.ShouldBe("Fault");
+ }
+
+ [Fact]
+ public void An_unavailable_condition_element_is_the_unavailable_sentinel()
+ {
+ // is genuinely no-comms and must land on the exact same sentinel Task 8
+ // matches for a Sample/Event's literal UNAVAILABLE text.
+ var observations = ParseFixture(UnavailableFixture);
+
+ observations["dev1_system_cond"].Value.ShouldBe("UNAVAILABLE");
+ }
+
+ [Fact]
+ public void The_all_unavailable_fixture_is_all_unavailable()
+ {
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(UnavailableFixture));
+
+ result.Observations.Select(o => o.DataItemId).ShouldBe(AllFixtureDataItemIds, ignoreOrder: true);
+ result.Observations.ShouldAllBe(o => o.Value == "UNAVAILABLE");
+ }
+
+ // ------------------------------------------------------------------------------- timestamps
+
+ [Fact]
+ public void Parse_normalizes_every_timestamp_to_utc()
+ {
+ // A wrong Kind silently shifts the OPC UA SourceTimestamp by the machine's UTC offset.
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
+
+ result.Observations.ShouldAllBe(o => o.TimestampUtc.Kind == DateTimeKind.Utc);
+ ParseCurrent()["dev1_pos"].TimestampUtc
+ .ShouldBe(new DateTime(2026, 7, 24, 12, 0, 0, 200, DateTimeKind.Utc));
+ }
+
+ [Fact]
+ public void An_offset_timestamp_is_converted_to_utc_rather_than_taken_at_face_value()
+ {
+ var result = MTConnectStreamsParser.Parse(StreamsDocument(
+ """O1"""));
+
+ var observation = result.Observations.ShouldHaveSingleItem();
+ observation.TimestampUtc.Kind.ShouldBe(DateTimeKind.Utc);
+ observation.TimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 12, 0, 0, DateTimeKind.Utc));
+ }
+
+ // -------------------------------------------------------------- sequence-gap detection (both ways)
+
+ [Fact]
+ public void Sequence_gap_is_detected_when_firstSequence_exceeds_requested_from()
+ {
+ var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
+
+ MTConnectAgentClient.IsSequenceGap(requestedFrom: 1, chunk).ShouldBeTrue();
+ }
+
+ [Fact]
+ public void No_sequence_gap_is_reported_for_a_contiguous_chunk()
+ {
+ // 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();
+ }
+
+ [Theory]
+ [InlineData(4999L, true)] // requested 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)
+ {
+ var chunk = MTConnectStreamsParser.Parse(File.ReadAllText(GapFixture));
+
+ MTConnectAgentClient.IsSequenceGap(requestedFrom, chunk).ShouldBe(expected);
+ }
+
+ // --------------------------------------------------------------- legal but degenerate documents
+
+ [Fact]
+ public void A_document_with_a_valid_header_and_no_observations_parses_rather_than_throwing()
+ {
+ // /sample chunks are deltas; an idle agent legitimately sends a document with nothing in it.
+ const string Xml = """
+
+
+
+
+ """;
+
+ var result = MTConnectStreamsParser.Parse(Xml);
+
+ result.Observations.ShouldBeEmpty();
+ result.InstanceId.ShouldBe(7);
+ result.NextSequence.ShouldBe(10);
+ result.FirstSequence.ShouldBe(10);
+ }
+
+ [Fact]
+ public void A_component_stream_may_omit_any_of_samples_events_and_condition()
+ {
+ // sample.xml carries no at all; absence is normal, not an error.
+ var result = MTConnectStreamsParser.Parse(File.ReadAllText(SampleFixture));
+
+ result.Observations.Select(o => o.DataItemId).ShouldBe(
+ ["dev1_pos", "dev1_vibration_ts", "dev1_partcount", "dev1_execution", "dev1_program"], ignoreOrder: true);
+ }
+
+ [Fact]
+ public void Parse_collects_observations_across_every_device_and_component_stream()
+ {
+ const string Xml = """
+
+
+
+
+
+ AVAILABLE
+
+
+
+
+ 1.0
+
+
+
+
+
+ """;
+
+ var result = MTConnectStreamsParser.Parse(Xml);
+
+ result.Observations.Select(o => o.DataItemId).ShouldBe(["a_avail", "b_pos", "b_cond"]);
+ result.Observations[2].Value.ShouldBe("Warning");
+ }
+
+ [Theory]
+ [InlineData("urn:mtconnect.org:MTConnectStreams:1.3")]
+ [InlineData("urn:mtconnect.org:MTConnectStreams:2.0")]
+ [InlineData("")]
+ public void Parse_is_agnostic_to_the_document_namespace_version(string namespaceUri)
+ {
+ var xmlns = namespaceUri.Length == 0 ? string.Empty : $" xmlns=\"{namespaceUri}\"";
+ var xml = $"""
+
+
+
+ O1
+
+
+ """;
+
+ MTConnectStreamsParser.Parse(xml).Observations.ShouldHaveSingleItem().DataItemId.ShouldBe("p");
+ }
+
+ [Fact]
+ public void Parse_from_a_stream_matches_the_string_overload()
+ {
+ using var stream = File.OpenRead(CurrentFixture);
+
+ var fromStream = MTConnectStreamsParser.Parse(stream);
+
+ var fromString = MTConnectStreamsParser.Parse(File.ReadAllText(CurrentFixture));
+ fromStream.NextSequence.ShouldBe(fromString.NextSequence);
+ fromStream.Observations.ShouldBe(fromString.Observations);
+ }
+
+ // ------------------------------------------------------------- fails loudly, never silently empty
+
+ [Theory]
+ [InlineData("")]
+ [InlineData(" ")]
+ [InlineData("not xml at all")]
+ [InlineData("")]
+ public void Parse_throws_on_malformed_xml(string xml)
+ {
+ Should.Throw(() => MTConnectStreamsParser.Parse(xml));
+ }
+
+ [Fact]
+ public void Parse_throws_on_a_devices_document_served_where_streams_were_expected()
+ {
+ var ex = Should.Throw(
+ () => MTConnectStreamsParser.Parse(""));
+
+ ex.Message.ShouldContain("MTConnectStreams");
+ ex.Message.ShouldContain("MTConnectDevices");
+ }
+
+ [Fact]
+ public void Parse_throws_on_an_mtconnect_error_document_and_surfaces_the_agent_error()
+ {
+ // An agent answers an out-of-range `from` with an MTConnectError document under HTTP 200.
+ const string Xml = """
+
+
+
+ 'from' must be greater than 100
+
+
+ """;
+
+ var ex = Should.Throw(() => MTConnectStreamsParser.Parse(Xml));
+
+ ex.Message.ShouldContain("OUT_OF_RANGE");
+ ex.Message.ShouldContain("'from' must be greater than 100");
+ }
+
+ [Theory]
+ [InlineData("")] // no at all
+ [InlineData("""""")] // no nextSequence
+ [InlineData("""""")] // no firstSequence
+ [InlineData("""""")] // no instanceId
+ [InlineData("""""")] // non-numeric
+ [InlineData("""""")] // non-numeric
+ public void Parse_throws_on_an_unusable_header(string headerXml)
+ {
+ var xml = $"{headerXml}";
+
+ Should.Throw(() => MTConnectStreamsParser.Parse(xml));
+ }
+
+ [Theory]
+ [InlineData("""1""")] // no dataItemId
+ [InlineData("""1""")] // no timestamp
+ [InlineData("""1""")] // unparseable timestamp
+ public void Parse_throws_on_a_malformed_observation(string observationXml)
+ {
+ Should.Throw(
+ () => MTConnectStreamsParser.Parse(StreamsDocument($"{observationXml}")));
+ }
+
+ [Fact]
+ public void Parse_reads_only_unqualified_attributes()
+ {
+ // A prefixed x:dataItemId is a vendor extension, not the observation's identity — reading it
+ // would invent a data item that the probe never declared.
+ var xml = StreamsDocument(
+ """1""");
+
+ Should.Throw(() => MTConnectStreamsParser.Parse(xml));
+ }
+
+ // ------------------------------------------------------------- MTConnectAgentClient.CurrentAsync
+
+ [Fact]
+ public async Task CurrentAsync_requests_the_agent_current_path_and_returns_the_parsed_snapshot()
+ {
+ var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ var result = await client.CurrentAsync(Ct);
+
+ handler.LastRequestUri!.AbsoluteUri.ShouldBe("http://agent:5000/current");
+ result.NextSequence.ShouldBe(108);
+ result.Observations.Count.ShouldBe(7);
+ }
+
+ [Fact]
+ public async Task CurrentAsync_scopes_the_request_to_the_configured_device_name()
+ {
+ var handler = StubHandler.RespondingWith(await File.ReadAllTextAsync(CurrentFixture, Ct));
+ using var client = NewClient(
+ handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000/", DeviceName = "VMC 3Axis" });
+
+ await client.CurrentAsync(Ct);
+
+ handler.LastRequestUri!.AbsoluteUri.ShouldBe("http://agent:5000/VMC%203Axis/current");
+ }
+
+ [Fact]
+ public async Task CurrentAsync_bounds_a_hung_agent_by_the_configured_request_timeout()
+ {
+ using var client = NewClient(
+ StubHandler.Hanging(),
+ new MTConnectDriverOptions { AgentUri = "http://agent:5000", RequestTimeoutMs = 50 });
+
+ var started = Stopwatch.StartNew();
+ var ex = await Should.ThrowAsync(async () => await client.CurrentAsync(Ct));
+
+ ex.ShouldNotBeAssignableTo();
+ ex.Message.ShouldContain("http://agent:5000/current");
+ started.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10));
+ }
+
+ [Fact]
+ public async Task CurrentAsync_throws_on_an_error_document_served_under_http_200()
+ {
+ var handler = StubHandler.RespondingWith(
+ """
+ nope
+ """);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ var ex = await Should.ThrowAsync(async () => await client.CurrentAsync(Ct));
+
+ ex.Message.ShouldContain("NO_DEVICE");
+ }
+
+ [Fact]
+ public async Task CurrentAsync_throws_on_a_non_success_status()
+ {
+ var handler = StubHandler.RespondingWith("nope", HttpStatusCode.ServiceUnavailable);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ await Should.ThrowAsync(async () => await client.CurrentAsync(Ct));
+ }
+
+ // -------------------------------------------------------------- MTConnectAgentClient.SampleAsync
+
+ [Fact]
+ public async Task SampleAsync_composes_the_from_interval_count_and_heartbeat_query()
+ {
+ var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct));
+ using var client = NewClient(handler, new MTConnectDriverOptions
+ {
+ AgentUri = "http://agent:5000",
+ DeviceName = "VMC-3Axis",
+ SampleIntervalMs = 250,
+ SampleCount = 64,
+ HeartbeatMs = 7000
+ });
+
+ await DrainAsync(client.SampleAsync(108, Ct));
+
+ handler.LastRequestUri!.AbsoluteUri.ShouldBe(
+ "http://agent:5000/VMC-3Axis/sample?from=108&interval=250&count=64&heartbeat=7000");
+ }
+
+ [Fact]
+ public async Task SampleAsync_yields_one_result_per_multipart_chunk()
+ {
+ var handler = StubHandler.RespondingWithMultipart(
+ "--------bnd",
+ 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.Count.ShouldBe(2);
+ chunks[0].FirstSequence.ShouldBe(108);
+ chunks[0].NextSequence.ShouldBe(113);
+ chunks[1].FirstSequence.ShouldBe(5000);
+ chunks[1].NextSequence.ShouldBe(5005);
+ }
+
+ [Fact]
+ public async Task SampleAsync_frames_chunks_from_the_boundary_when_the_agent_omits_content_length()
+ {
+ var handler = StubHandler.RespondingWithMultipart(
+ "bnd",
+ 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_exposes_the_gap_signal_to_the_caller_on_the_chunk_that_carries_it()
+ {
+ // Task 11's pump: advance `from` to NextSequence while contiguous, re-baseline on a gap.
+ var handler = StubHandler.RespondingWithMultipart(
+ "bnd",
+ await File.ReadAllTextAsync(SampleFixture, Ct),
+ await File.ReadAllTextAsync(GapFixture, Ct));
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ var from = 108L;
+ var gaps = new List();
+ await foreach (var chunk in client.SampleAsync(from, Ct))
+ {
+ gaps.Add(MTConnectAgentClient.IsSequenceGap(from, chunk));
+ from = chunk.NextSequence;
+ }
+
+ gaps.ShouldBe([false, true]);
+ from.ShouldBe(5005);
+ }
+
+ [Fact]
+ public async Task SampleAsync_ignores_an_empty_keepalive_part()
+ {
+ var handler = StubHandler.RespondingWithMultipart(
+ "bnd",
+ 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.Count.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task SampleAsync_yields_a_heartbeat_document_that_carries_no_observations()
+ {
+ // The agent's real keep-alive is a well-formed but observation-free MTConnectStreams
+ // document — it advances the sequence, so it must reach the caller, not be swallowed.
+ const string Heartbeat = """
+
+
+
+
+ """;
+ var handler = StubHandler.RespondingWithMultipart("bnd", Heartbeat);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ var chunks = await DrainAsync(client.SampleAsync(113, Ct));
+
+ chunks.ShouldHaveSingleItem().Observations.ShouldBeEmpty();
+ MTConnectAgentClient.IsSequenceGap(113, chunks[0]).ShouldBeFalse();
+ }
+
+ [Fact(Timeout = 30_000)]
+ public async Task SampleAsync_faults_on_the_heartbeat_watchdog_rather_than_hanging_on_a_silent_agent()
+ {
+ // The R2-01 frozen-peer shape: the agent accepts the connection, sends one chunk, then goes
+ // silent forever. Without a watchdog the pump wedges with no timeout ever firing.
+ var prefix = StubHandler.MultipartBody("bnd", withContentLength: true, closing: false,
+ await File.ReadAllTextAsync(SampleFixture, Ct));
+ var handler = StubHandler.RespondingWithStallingStream("bnd", prefix);
+ using var client = NewClient(handler, new MTConnectDriverOptions
+ {
+ AgentUri = "http://agent:5000",
+ RequestTimeoutMs = 100,
+ HeartbeatMs = 100
+ });
+
+ var started = Stopwatch.StartNew();
+ await using var chunks = client.SampleAsync(108, Ct).GetAsyncEnumerator(Ct);
+
+ (await chunks.MoveNextAsync()).ShouldBeTrue();
+ chunks.Current.FirstSequence.ShouldBe(108);
+
+ var ex = await Should.ThrowAsync(async () => await chunks.MoveNextAsync());
+ ex.ShouldNotBeAssignableTo();
+ ex.Message.ShouldContain("sample");
+ started.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(20));
+ }
+
+ [Fact(Timeout = 30_000)]
+ public async Task SampleAsync_bounds_the_response_header_phase_by_the_request_timeout()
+ {
+ // A peer that accepts the TCP connection but never answers must not wedge the pump either.
+ using var client = NewClient(
+ StubHandler.Hanging(),
+ new MTConnectDriverOptions { AgentUri = "http://agent:5000", RequestTimeoutMs = 50 });
+
+ var ex = await Should.ThrowAsync(async () => await DrainAsync(client.SampleAsync(1, Ct)));
+
+ ex.ShouldNotBeAssignableTo();
+ ex.Message.ShouldContain("/sample");
+ }
+
+ [Fact]
+ public async Task SampleAsync_throws_on_an_error_document_served_under_http_200()
+ {
+ // Not a multipart body at all — the agent answers a bad `from` with an error document.
+ var handler = StubHandler.RespondingWith(
+ """
+ bad from
+ """);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ var ex = await Should.ThrowAsync(
+ async () => await DrainAsync(client.SampleAsync(1, Ct)));
+
+ ex.Message.ShouldContain("OUT_OF_RANGE");
+ }
+
+ [Fact]
+ public async Task SampleAsync_throws_on_a_non_success_status()
+ {
+ var handler = StubHandler.RespondingWith("nope", HttpStatusCode.BadRequest);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ await Should.ThrowAsync(async () => await DrainAsync(client.SampleAsync(1, Ct)));
+ }
+
+ [Fact(Timeout = 30_000)]
+ public async Task SampleAsync_honours_the_callers_cancellation_token()
+ {
+ var handler = StubHandler.RespondingWithStallingStream("bnd", []);
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+ using var cts = new CancellationTokenSource();
+ await cts.CancelAsync();
+
+ await Should.ThrowAsync(
+ async () => await DrainAsync(client.SampleAsync(1, cts.Token)));
+ }
+
+ [Fact]
+ public async Task SampleAsync_ends_cleanly_when_the_agent_closes_the_stream()
+ {
+ 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));
+
+ chunks.ShouldHaveSingleItem();
+ }
+
+ [Fact]
+ public void SampleAsync_opens_no_connection_until_it_is_enumerated()
+ {
+ var handler = StubHandler.Hanging();
+ using var client = NewClient(handler, new MTConnectDriverOptions { AgentUri = "http://agent:5000" });
+
+ _ = client.SampleAsync(1, CancellationToken.None);
+
+ handler.RequestCount.ShouldBe(0);
+ }
+
+ // ------------------------------------------------------------------------------------ helpers
+
+ private static CancellationToken Ct => TestContext.Current.CancellationToken;
+
+ private static MTConnectAgentClient NewClient(HttpMessageHandler handler, MTConnectDriverOptions options) =>
+ new(options, handler);
+
+ private static async Task> DrainAsync(IAsyncEnumerable source)
+ {
+ var results = new List();
+ await foreach (var item in source)
+ {
+ results.Add(item);
+ }
+
+ return results;
+ }
+
+ private static Dictionary ParseCurrent() => ParseFixture(CurrentFixture);
+
+ private static Dictionary ParseFixture(string path) =>
+ MTConnectStreamsParser.Parse(File.ReadAllText(path)).Observations.ToDictionary(o => o.DataItemId);
+
+ /// Wraps observation-container XML in a minimal but valid streams document.
+ private static string StreamsDocument(string componentStreamBody) =>
+ $"""
+
+
+
+
+ {componentStreamBody}
+
+
+
+ """;
+
+ /// A canned — no socket is ever opened.
+ private sealed class StubHandler : HttpMessageHandler
+ {
+ private readonly Func? _responder;
+ private readonly bool _hang;
+
+ private StubHandler(Func? responder, bool hang)
+ {
+ _responder = responder;
+ _hang = hang;
+ }
+
+ public Uri? LastRequestUri { get; private set; }
+
+ public int RequestCount { get; private set; }
+
+ public static StubHandler RespondingWith(string body, HttpStatusCode status = HttpStatusCode.OK) =>
+ new(() => new HttpResponseMessage(status)
+ {
+ Content = new StringContent(body, Encoding.UTF8, "text/xml")
+ }, hang: false);
+
+ public static StubHandler Hanging() => new(responder: null, hang: true);
+
+ public static StubHandler RespondingWithMultipart(string boundary, params string[] parts) =>
+ RespondingWithMultipart(boundary, withContentLength: true, parts);
+
+ public static StubHandler RespondingWithMultipart(string boundary, bool withContentLength, params string[] parts)
+ {
+ var body = MultipartBody(boundary, withContentLength, closing: true, parts);
+
+ return new StubHandler(() => Multipart(boundary, new ByteArrayContent(body)), hang: false);
+ }
+
+ public static StubHandler RespondingWithStallingStream(string boundary, byte[] prefix) =>
+ new(() => Multipart(boundary, new StallingContent(prefix)), hang: false);
+
+ /// Builds a multipart/x-mixed-replace body exactly as an Agent writes one.
+ public static byte[] MultipartBody(string boundary, bool withContentLength, bool closing, params string[] parts)
+ {
+ var body = new List();
+ foreach (var part in parts)
+ {
+ var payload = Encoding.UTF8.GetBytes(part);
+ var header = new StringBuilder()
+ .Append("--").Append(boundary).Append("\r\n")
+ .Append("Content-type: text/xml\r\n");
+ if (withContentLength)
+ {
+ header.Append("Content-length: ").Append(payload.Length).Append("\r\n");
+ }
+
+ header.Append("\r\n");
+ body.AddRange(Encoding.ASCII.GetBytes(header.ToString()));
+ body.AddRange(payload);
+ body.AddRange("\r\n"u8.ToArray());
+ }
+
+ if (closing)
+ {
+ body.AddRange(Encoding.ASCII.GetBytes($"--{boundary}--\r\n"));
+ }
+
+ return [.. body];
+ }
+
+ private static HttpResponseMessage Multipart(string boundary, HttpContent content)
+ {
+ content.Headers.ContentType =
+ MediaTypeHeaderValue.Parse($"multipart/x-mixed-replace; boundary={boundary}");
+
+ return new HttpResponseMessage(HttpStatusCode.OK) { Content = content };
+ }
+
+ protected override async Task SendAsync(
+ HttpRequestMessage request, CancellationToken cancellationToken)
+ {
+ LastRequestUri = request.RequestUri;
+ RequestCount++;
+
+ if (_hang)
+ {
+ await Task.Delay(Timeout.Infinite, cancellationToken);
+ }
+
+ return _responder!();
+ }
+ }
+
+ /// Serves , then never completes another read.
+ private sealed class StallingContent(byte[] prefix) : HttpContent
+ {
+ protected override Task CreateContentReadStreamAsync() =>
+ Task.FromResult(new StallingStream(prefix));
+
+ protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) =>
+ Task.CompletedTask;
+
+ protected override bool TryComputeLength(out long length)
+ {
+ length = -1;
+
+ return false;
+ }
+ }
+
+ private sealed class StallingStream(byte[] prefix) : Stream
+ {
+ private int _position;
+
+ 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 => throw new NotSupportedException();
+ set => throw new NotSupportedException();
+ }
+
+ public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default)
+ {
+ if (_position < prefix.Length)
+ {
+ var count = Math.Min(buffer.Length, prefix.Length - _position);
+ prefix.AsMemory(_position, count).CopyTo(buffer);
+ _position += count;
+
+ return count;
+ }
+
+ await Task.Delay(Timeout.Infinite, cancellationToken);
+
+ return 0;
+ }
+
+ public override Task 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();
+ }
+}