Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/EvidenceTests.cs
T

262 lines
14 KiB
C#

using System.Buffers.Binary;
using System.Globalization;
using System.Text;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;
/// <summary>
/// Hardware-evidence legs of Gitea #592 (hw4 lane evidence), READ-ONLY by construction and covered by the same
/// source/IL guard as the rest of this folder. Both legs are opt-in on top of <c>AB_DIFF_ENDPOINT</c> because each
/// is its own press session (the browse is ~3000 tags):
/// <list type="bullet">
/// <item><c>AB_EVID_MIX=1</c> — #582: reads a STRING, an array element, a whole array and scalars concurrently
/// through the native transport and proves, from the exchange tap on the connection, that they travelled
/// as ONE Multiple Service Packet and decode to what individual reads return. Tag set:
/// <c>AB_EVID_TAGS</c> = <c>name|Type|count</c> entries separated by <c>;</c> (count &gt; 1 or
/// <c>[]</c> suffix on Type ⇒ array).</item>
/// <item><c>AB_EVID_BROWSE=1</c> — #572: one native <c>@tags</c> browse with every listing page's instance-id
/// sequence recorded, reporting any page that is not ascending.</item>
/// </list>
/// </summary>
[Collection(DifferentialCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class EvidenceTests
{
private const string DefaultMix =
"Program:UN01_Press.Sts_CycleTime_STR|String|1;Program:EM01_PowerUnit._SEQ_EP[1]|DInt|1;" +
"Program:EM01_PowerUnit._SEQ_EP|DInt|32;test2|DInt|1;Program:UN01_Press.xxtest|Bool|1";
private sealed record ExchangeRecord(byte Service, int RequestBytes, int ReplyBytes, int Embedded, string Detail);
[Fact]
public async Task Mixed_string_array_and_scalar_reads_travel_as_one_multiple_service_packet()
{
var target = DifferentialTarget.FromEnvironment();
if (target is null) Assert.Skip(DifferentialTarget.SkipReason);
if (Environment.GetEnvironmentVariable("AB_EVID_MIX") != "1") Assert.Skip("AB_EVID_MIX=1 not set.");
var specs = ParseMix(Environment.GetEnvironmentVariable("AB_EVID_TAGS") is { Length: > 0 } t ? t : DefaultMix);
using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
var ct = cts.Token;
var diagnostics = DeviceSessionPool.Shared.Diagnostics;
var exchanges = new List<ExchangeRecord>();
var factory = AbCipTransportFactories.For(AbCipTransport.Native).Tags;
const int Attempts = 5;
var ((before, attempts, after), opens, closes) =
await DifferentialRun.WithNativeAnchorAsync("evidence mix", target, async checkOpens =>
{
var runtimes = new List<(MixSpec Spec, IAbCipTagRuntime Rt)>();
try
{
foreach (var s in specs)
{
await DifferentialRun.PaceAsync(target, ct);
var rt = factory.Create(target.Params(s.Name, s.Count, s.Count > 1, s.Type));
await rt.InitializeAsync(ct); // one read each (create-read)
checkOpens($"create of '{s.Name}'");
rt.GetStatus().ShouldBe((int)AbStatus.Ok, $"'{s.Name}' must create/read Ok");
runtimes.Add((s, rt));
}
async Task<List<string>> SequentialAsync()
{
var values = new List<string>();
foreach (var (spec, rt) in runtimes)
{
await DifferentialRun.PaceAsync(target, ct);
await rt.ReadAsync(ct);
checkOpens($"single read of '{spec.Name}'");
values.Add(Format(spec, rt));
await Task.Delay(1000, ct); // >= 1 s between repeated reads
}
return values;
}
var beforeValues = await SequentialAsync();
// The batch: every read issued at once so the scheduler packs them. Whether the pump wakes
// before or after the last request is queued is a scheduling race, so the batch is repeated
// (>= 1 s apart) and each attempt reports how it was packed.
var done = new List<(int Requests, List<ExchangeRecord> Records, List<string> Values)>();
for (var attempt = 0; attempt < Attempts; attempt++)
{
exchanges.Clear();
diagnostics.ExchangeObserver = (req, rep) => exchanges.Add(Describe(req, rep));
var requestsBefore = diagnostics.RequestsSent;
await Task.WhenAll(runtimes.Select(r => r.Rt.ReadAsync(ct)));
var requests = diagnostics.RequestsSent - requestsBefore;
diagnostics.ExchangeObserver = null;
checkOpens($"batch {attempt + 1}");
done.Add((requests, exchanges.ToList(), runtimes.Select(r => Format(r.Spec, r.Rt)).ToList()));
await Task.Delay(1000, ct);
}
var afterValues = await SequentialAsync();
return (beforeValues, done, afterValues);
}
finally
{
diagnostics.ExchangeObserver = null;
foreach (var (_, rt) in runtimes) rt.Dispose();
}
});
Log($"## Mixed batch — {target}");
Log($"Tags: {string.Join(" ; ", specs.Select(s => $"{s.Name} ({s.Type}{(s.Count > 1 ? $"[{s.Count}]" : "")})"))}");
Log($"Native Forward Opens: {opens}; Forward Closes sent/acked: {closes.Sent}/{closes.Acked}");
var oneMsp = 0;
for (var n = 0; n < attempts.Count; n++)
{
var (requests, records, values) = attempts[n];
var single = requests == 1 && records.Count == 1 && records[0].Service == 0x0A && records[0].Embedded == specs.Count;
if (single) oneMsp++;
Log($"Attempt {n + 1}: {specs.Count} concurrent reads -> {requests} channel send(s); {(single ? "ONE MSP carrying all" : "split")}:");
foreach (var e in records)
Log($" - request service 0x{e.Service:X2}, {e.RequestBytes} B -> reply {e.ReplyBytes} B, embedded={e.Embedded}: {e.Detail}");
records.ShouldAllBe(e => !e.Detail.Contains("status=0x06"), "a 0x06 inside the MSP would mean a truncated reply");
records.Sum(e => e.Embedded).ShouldBe(specs.Count);
records.ShouldAllBe(e => e.Service == 0x0A || e.Embedded == 1);
}
Log("| Tag | single (before) | " + string.Join(" | ", Enumerable.Range(1, attempts.Count).Select(i => $"batch {i}")) + " | single (after) | all batch values match a bracketing single |");
Log("|---|---|" + string.Concat(Enumerable.Repeat("---|", attempts.Count)) + "---|---|");
var failures = new List<string>();
for (var i = 0; i < specs.Count; i++)
{
var ok = attempts.All(a => a.Values[i] == before[i] || a.Values[i] == after[i]);
if (!ok) failures.Add(specs[i].Name);
Log($"| `{specs[i].Name}` | {before[i]} | {string.Join(" | ", attempts.Select(a => a.Values[i]))} | {after[i]} | {(ok ? "yes" : "**NO**")} |");
}
Log($"Attempts that carried every read in ONE MSP: {oneMsp}/{attempts.Count}");
opens.ShouldBe(1);
closes.Sent.ShouldBe(1);
oneMsp.ShouldBeGreaterThan(0, "at least one attempt must put the whole mixed batch in a single Multiple Service Packet");
failures.ShouldBeEmpty("a batch value matched neither bracketing single read");
}
[Fact]
public async Task Symbol_listing_pages_are_ascending_on_a_real_controller()
{
var target = DifferentialTarget.FromEnvironment();
if (target is null) Assert.Skip(DifferentialTarget.SkipReason);
if (Environment.GetEnvironmentVariable("AB_EVID_BROWSE") != "1") Assert.Skip("AB_EVID_BROWSE=1 not set.");
using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(10));
var ct = cts.Token;
var diagnostics = DeviceSessionPool.Shared.Diagnostics;
var pages = new List<(string Path, List<uint> Ids, bool More)>();
var gate = new object();
var factory = AbCipTransportFactories.For(AbCipTransport.Native).Enumerators;
var (count, opens, closes) = await DifferentialRun.WithNativeAnchorAsync("evidence browse", target, async checkOpens =>
{
diagnostics.ExchangeObserver = (req, rep) =>
{
if (req.Span.Length == 0 || req.Span[0] != 0x55) return;
var page = ParseListingPage(req.Span, rep.Span);
if (page is { } p) lock (gate) pages.Add(p);
};
try
{
var n = 0;
await DifferentialRun.PaceAsync(target, ct);
using var enumerator = factory.Create();
await foreach (var _ in enumerator.EnumerateAsync(target.Params("@tags"), ct))
{
n++;
if (n % 200 == 0) checkOpens($"browse entry {n}");
}
checkOpens("browse");
return n;
}
finally { diagnostics.ExchangeObserver = null; }
});
Log($"## Listing order — {target}");
Log($"Tags browsed: {count}; listing pages observed: {pages.Count}; Forward Opens {opens}; Forward Closes {closes.Sent}/{closes.Acked}");
var nonAscending = 0;
for (var i = 0; i < pages.Count; i++)
{
var (path, ids, more) = pages[i];
var strictly = true;
for (var k = 1; k < ids.Count; k++) if (ids[k] <= ids[k - 1]) { strictly = false; break; }
if (!strictly) nonAscending++;
var sample = ids.Count <= 12
? string.Join(",", ids)
: string.Join(",", ids.Take(6)) + ",...," + string.Join(",", ids.Skip(ids.Count - 4));
Log($"page {i + 1}: scope={path} entries={ids.Count} first={(ids.Count > 0 ? ids[0] : 0)} last={(ids.Count > 0 ? ids[^1] : 0)} " +
$"max={(ids.Count > 0 ? ids.Max() : 0)} more={more} ascending={strictly} ids=[{sample}]");
}
Log($"Non-ascending pages: {nonAscending}");
Log(nonAscending == 0 ? "RESULT: every listing page strictly ascending." : $"RESULT: {nonAscending} NON-ASCENDING page(s) observed.");
pages.ShouldNotBeEmpty();
opens.ShouldBe(1);
}
private sealed record MixSpec(string Name, AbCipDataType Type, int Count);
private static List<MixSpec> ParseMix(string raw) => raw.Split(';', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)
.Select(e =>
{
var p = e.Split('|');
return new MixSpec(p[0], Enum.Parse<AbCipDataType>(p[1], ignoreCase: true),
p.Length > 2 ? int.Parse(p[2], CultureInfo.InvariantCulture) : 1);
}).ToList();
private static string Format(MixSpec spec, IAbCipTagRuntime rt)
{
if (spec.Count > 1)
return rt.DecodeArray(spec.Type, spec.Count) is System.Collections.IEnumerable e
? "[" + string.Join(",", e.Cast<object?>().Select(v => Convert.ToString(v, CultureInfo.InvariantCulture))) + "]"
: "null";
var v = rt.DecodeValue(spec.Type, null);
return v is string s ? $"'{s}'" : Convert.ToString(v, CultureInfo.InvariantCulture) ?? "null";
}
private static ExchangeRecord Describe(ReadOnlyMemory<byte> request, ReadOnlyMemory<byte> reply)
{
var req = request.Span;
var service = req.Length > 0 ? req[0] : (byte)0;
if (service != 0x0A)
return new ExchangeRecord(service, req.Length, reply.Length, 1, $"single, reply head {Hex(reply.Span, 4)}");
var count = req.Length >= 8 ? BinaryPrimitives.ReadUInt16LittleEndian(req[6..]) : 0;
var outer = CipReply.Parse(reply);
var parts = new List<string> { $"outer status=0x{outer.GeneralStatus:X2}" };
if (outer.GeneralStatus is 0x00 or 0x1E)
{
var inner = MultipleServicePacket.Parse(outer, count);
for (var i = 0; i < inner.Count; i++)
parts.Add($"#{i}: svc=0x{inner[i].Service:X2} status=0x{inner[i].GeneralStatus:X2} data={inner[i].Data.Length}B");
}
return new ExchangeRecord(service, req.Length, reply.Length, count, string.Join("; ", parts));
}
private static (string Path, List<uint> Ids, bool More)? ParseListingPage(ReadOnlySpan<byte> request, ReadOnlySpan<byte> reply)
{
var parsed = CipReply.Parse(reply.ToArray());
if (parsed.GeneralStatus is not (0x00 or 0x06)) return null;
var pathWords = request.Length > 1 ? request[1] : 0;
var scope = request.Length >= 2 + pathWords * 2 && request.Length > 2 && request[2] == 0x91
? Encoding.ASCII.GetString(request.Slice(4, request[3])) : "controller";
var d = parsed.Data.Span;
var ids = new List<uint>();
var at = 0;
while (at + 4 + 2 + 2 + 12 + 2 <= d.Length)
{
ids.Add(BinaryPrimitives.ReadUInt32LittleEndian(d[at..]));
var nameLen = BinaryPrimitives.ReadUInt16LittleEndian(d[(at + 4 + 2 + 2 + 12)..]);
at += 4 + 2 + 2 + 12 + 2 + nameLen;
}
return (scope, ids, parsed.GeneralStatus == 0x06);
}
private static string Hex(ReadOnlySpan<byte> b, int max) => Convert.ToHexString(b[..Math.Min(max, b.Length)]);
private static void Log(string line) => TestContext.Current.TestOutputHelper?.WriteLine(line);
}