using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
///
/// Review 2026-10-21 R06:T-4 and T-2 on the read path.
///
/// - T-4 — every ReadAsync batch fetched + parsed the whole device /current even while the
/// /sample pump held a live, healthy index. A read now answers from that index (READ only — the pump
/// stays its single writer) while the pump is live and not degraded, and keeps /current as the
/// fallback.
/// - T-2 — Agent-restart detection lived only on the pump, and the read path discarded the
/// /current answer's instanceId: a browse-only driver kept a cached /probe model across an
/// Agent restart and never raised rediscovery. The fallback read now checks it.
///
///
public sealed class MTConnectReadPathTests
{
private const long RestartedInstance = 1655000777L;
private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static MTConnectDriverOptions Opts() => new()
{
AgentUri = "http://fixture-agent:5000",
RequestTimeoutMs = 5000,
Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
};
/// T-4: with a live, healthy pump the read is answered from its index — no /current round trip.
[Fact]
public async Task A_read_with_a_live_pump_answers_from_the_index_without_a_current_call()
{
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-t4", _ => client);
await driver.InitializeAsync("{}", Ct);
try
{
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.PumpAsync(MTConnectPumpGenerationChunks.Pos(108, 113, "7.25")).WaitAsync(Watchdog, Ct);
var currentCalls = client.CurrentCallCount;
var result = (await driver.ReadAsync(["dev1_pos"], Ct)).Single();
result.Value.ShouldBe(7.25d, "the read must serve the pump's newest value");
client.CurrentCallCount.ShouldBe(currentCalls, "no /current while the pump's index is live");
}
finally
{
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// Positive control for T-4: with no pump the read still issues exactly one /current.
[Fact]
public async Task A_read_with_no_pump_falls_back_to_one_current_call()
{
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-t4-fallback", _ => client);
await driver.InitializeAsync("{}", Ct);
try
{
var currentCalls = client.CurrentCallCount;
await driver.ReadAsync(["dev1_pos"], Ct);
client.CurrentCallCount.ShouldBe(currentCalls + 1);
}
finally
{
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// T-4: a pump whose stream is failing does not serve reads — they fall back to /current.
[Fact]
public async Task A_read_with_a_degraded_pump_falls_back_to_current()
{
var client = CannedAgentClient.FromFixtures();
// Every /sample fails: the pump stays alive (reconnecting under backoff) but its stream is degraded.
client.SampleFailure = new IOException("connection refused");
var options = new MTConnectDriverOptions
{
AgentUri = "http://fixture-agent:5000",
RequestTimeoutMs = 5000,
Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
Reconnect = new MTConnectReconnectOptions { MinBackoffMs = 1000, MaxBackoffMs = 1000 },
};
var driver = new MTConnectDriver(options, "mt-t4-degraded", _ => client);
await driver.InitializeAsync("{}", Ct);
try
{
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
var deadline = DateTime.UtcNow + Watchdog;
while (driver.GetHealth().State != DriverState.Degraded && DateTime.UtcNow < deadline)
await Task.Delay(20, Ct);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
var currentCalls = client.CurrentCallCount;
await driver.ReadAsync(["dev1_pos"], Ct);
client.CurrentCallCount.ShouldBe(currentCalls + 1, "a degraded stream's index is not trusted for reads");
}
finally
{
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// T-2: a browse-only driver's read sees the Agent restart — once — and drops the stale model.
[Fact]
public async Task A_read_that_sees_a_new_instance_id_raises_rediscovery_once_and_drops_the_probe_model()
{
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-t2", _ => client);
await driver.InitializeAsync("{}", Ct);
var raised = 0;
driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised);
try
{
driver.CachedProbeModel.ShouldNotBeNull("positive control: initialize cached the model");
client.Current = new MTConnectStreamsResult(RestartedInstance, 5, 1, client.Current.Observations);
await driver.ReadAsync(["dev1_pos"], Ct);
await driver.ReadAsync(["dev1_pos"], Ct);
raised.ShouldBe(1, "once per distinct instanceId, not once per read");
driver.CachedProbeModel.ShouldBeNull("the model describes a process that no longer exists");
// Review m6: the session's cursor moves to the new process, so a later subscription does not open at the
// dead instance's sequence (and flicker through a clear + re-baseline on its first chunk).
driver.AgentInstanceId.ShouldBe(RestartedInstance);
driver.AgentNextSequence.ShouldBe(5);
}
finally
{
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// T-2 control: reads against the same Agent process never raise rediscovery.
[Fact]
public async Task Reads_against_the_same_instance_raise_nothing()
{
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-t2-same", _ => client);
await driver.InitializeAsync("{}", Ct);
var raised = 0;
driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised);
try
{
await driver.ReadAsync(["dev1_pos"], Ct);
await driver.ReadAsync(["dev1_pos"], Ct);
raised.ShouldBe(0);
driver.CachedProbeModel.ShouldNotBeNull();
}
finally
{
await driver.ShutdownAsync(CancellationToken.None);
}
}
private const string NoDeviceError =
"""
Could not find the device 'VMC-9'
""";
private static Exception ParseFailure(string xml)
{
try
{
MTConnectStreamsParser.Parse(xml);
}
catch (Exception ex)
{
return ex;
}
throw new InvalidOperationException("expected the parse to fail");
}
///
/// Health contract: an MTConnectError answer to /current is the Agent REFUSING (it answered) — the batch is Bad,
/// but health is restored / kept Healthy with the stamp advanced, and the refusal is logged once, not per read.
///
[Fact]
public async Task An_agent_refusal_of_current_is_a_reply_not_a_comms_fault()
{
var logger = new RecordingDriverLogger();
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-refusal", _ => client, logger);
await driver.InitializeAsync("{}", Ct);
try
{
// A comms fault first, so the refusal has a Degraded to restore.
client.CurrentFailure = new HttpRequestException("connection refused");
await driver.ReadAsync(["dev1_pos"], Ct);
driver.GetHealth().State.ShouldBe(DriverState.Degraded, "positive control: a comms fault does degrade");
var stamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
var refusal = ParseFailure(NoDeviceError);
MTConnectXml.IsAgentRefusal(refusal).ShouldBeTrue();
client.CurrentFailure = refusal;
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (DateTime.UtcNow <= stamp && DateTime.UtcNow < deadline) await Task.Yield();
var first = await driver.ReadAsync(["dev1_pos"], Ct);
await driver.ReadAsync(["dev1_pos"], Ct);
first.Single().StatusCode.ShouldBe(0x80050000u, "the batch is still Bad");
var h = driver.GetHealth();
h.State.ShouldBe(DriverState.Healthy);
h.LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(stamp);
logger.WarningsSnapshot().Count(w => w.Contains("NO_DEVICE")).ShouldBe(1, "a persistent refusal warns once");
}
finally
{
client.CurrentFailure = null;
await driver.ShutdownAsync(CancellationToken.None);
}
}
/// An answer that is not an MTConnect document at all is NOT marked a refusal (it stays a comms fault).
[Fact]
public void A_garbage_answer_is_not_an_agent_refusal() =>
MTConnectXml.IsAgentRefusal(ParseFailure("
proxy error")).ShouldBeFalse();
/// Review m7: a refusal that recurs AFTER a good read is news again and is logged again.
[Fact]
public async Task A_refusal_recurring_after_a_good_read_is_logged_again()
{
var logger = new RecordingDriverLogger();
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(Opts(), "mt-refusal-again", _ => client, logger);
await driver.InitializeAsync("{}", Ct);
try
{
var refusal = ParseFailure(NoDeviceError);
client.CurrentFailure = refusal;
await driver.ReadAsync(["dev1_pos"], Ct);
await driver.ReadAsync(["dev1_pos"], Ct);
logger.WarningsSnapshot().Count(w => w.Contains("NO_DEVICE")).ShouldBe(1, "positive control: deduplicated while it persists");
client.CurrentFailure = null;
await driver.ReadAsync(["dev1_pos"], Ct);
client.CurrentFailure = refusal;
await driver.ReadAsync(["dev1_pos"], Ct);
logger.WarningsSnapshot().Count(w => w.Contains("NO_DEVICE")).ShouldBe(2);
}
finally
{
client.CurrentFailure = null;
await driver.ShutdownAsync(CancellationToken.None);
}
}
}