using Opc.Ua;
using Opc.Ua.Client;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
/// Exercises the production session continuation store through actual HistoryRead requests.
public sealed class SessionHistoryContinuationLifecycleTests
{
private static readonly DateTime Epoch = new(2026, 10, 1, 0, 0, 0, DateTimeKind.Utc);
[Fact]
public async Task Tokens_are_single_use_and_cannot_cross_sessions()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartHistoryAsync(ct);
using var owner = await rig.ConnectAsync(anonymous: true, ct);
using var other = await rig.ConnectAsync(anonymous: true, ct);
var point = (await PageAsync(owner, default, false, ct)).ContinuationPoint;
point.IsEmpty.ShouldBeFalse();
(await PageAsync(other, point, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
var resumed = await PageAsync(owner, point, false, ct);
resumed.StatusCode.ShouldBe(StatusCodes.Good);
resumed.ContinuationPoint.IsEmpty.ShouldBeFalse();
(await PageAsync(owner, point, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
}
[Fact]
public async Task Release_consumes_a_token_and_malformed_tokens_are_rejected()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartHistoryAsync(ct);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var point = (await PageAsync(session, default, false, ct)).ContinuationPoint;
point.IsEmpty.ShouldBeFalse();
var released = await PageAsync(session, point, true, ct);
released.StatusCode.ShouldBe(StatusCodes.Good);
released.ContinuationPoint.IsEmpty.ShouldBeTrue();
(await PageAsync(session, point, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
(await PageAsync(session, ByteString.From(new byte[] { 1, 2, 3 }), false, ct))
.StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
}
[Fact]
public async Task Session_capacity_evicts_the_oldest_point_and_keeps_the_newest()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartHistoryAsync(ct);
using var session = await rig.ConnectAsync(anonymous: true, ct);
var first = (await PageAsync(session, default, false, ct)).ContinuationPoint;
ByteString latest = default;
// The production host retains the SDK's cap of 100; issue 101 distinct initial reads.
for (var i = 0; i < 100; i++) latest = (await PageAsync(session, default, false, ct)).ContinuationPoint;
latest.IsEmpty.ShouldBeFalse();
(await PageAsync(session, first, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
(await PageAsync(session, latest, false, ct)).StatusCode.ShouldBe(StatusCodes.Good);
}
[Fact]
public async Task Closed_session_tokens_do_not_resume_on_a_new_session()
{
var ct = TestContext.Current.CancellationToken;
await using var rig = await StartHistoryAsync(ct);
ByteString point;
using (var old = await rig.ConnectAsync(anonymous: true, ct))
{
point = (await PageAsync(old, default, false, ct)).ContinuationPoint;
point.IsEmpty.ShouldBeFalse();
await old.CloseAsync(ct);
}
using var fresh = await rig.ConnectAsync(anonymous: true, ct);
(await PageAsync(fresh, point, false, ct)).StatusCode.ShouldBe(StatusCodes.BadContinuationPointInvalid);
}
private static async Task StartHistoryAsync(CancellationToken ct)
{
var rig = await ServerEventRig.StartAsync("SessionHistoryLifecycle", encrypted: false, ct);
try
{
var events = Enumerable.Range(0, 25)
.Select(i => new HistoricalEvent($"e{i:00}", "src", Epoch.AddSeconds(i / 2), Epoch, "m", 500)).ToList();
rig.Nm.HistorianDataSource = new EventHistoryPagingOverTheWireTests.SqlLikeEventSource(events);
rig.MaterialiseNative();
rig.Enforce(false, [Key()]);
return rig;
}
catch
{
await rig.DisposeAsync();
throw;
}
}
private static async Task PageAsync(ISession session, ByteString point, bool release, CancellationToken ct)
{
var node = new NodeId(RawDeviceFolder, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
var filter = new EventFilter
{
SelectClauses = [new SimpleAttributeOperand(ObjectTypeIds.BaseEventType, QualifiedName.From(BrowseNames.EventId))],
};
var details = new ReadEventDetails
{
StartTime = Epoch.AddDays(-1), EndTime = Epoch.AddDays(1), NumValuesPerNode = 10, Filter = filter,
};
var response = await session.HistoryReadAsync(null, new ExtensionObject(details), TimestampsToReturn.Both, release,
[new HistoryReadValueId { NodeId = node, ContinuationPoint = point }], ct);
return response.Results.ToList().ShouldHaveSingleItem();
}
}