Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/SqlServerConformanceBackend.cs
T
Joseph Doherty 0658d244d0 feat(historian): #674 HistoryUpdate posture — future-skew bound, Primary gate, SqlServer v1 accepted, QuestDB TTL pin
- HistoryUpdate items with SourceTimestamp > now + ServerHistorian:HistoryUpdateMaxFutureSkew
  (default 1 day, <=0 disables) => BadOutOfRange in the node manager (every backend, never sent);
  clock injected via OtOpcUaNodeManager.Clock (host TimeProvider).
- HistoryUpdate is Primary-only: off-Primary => BadInvalidState per node, after authorization,
  same IDriverPrimaryRoleSource verdict as the native-ack gate; HistoryRead ungated.
- SqlServer: base schema requirement back to v1; SupportsModifiedTrail (v>=2). On v1 with
  AutoCreateSchema=false: writes/reads/events/retention work, modified reads + HistoryUpdate =>
  BadHistoryOperationUnsupported with nothing sent, retention skips SampleModified, one Warning per boot.
- QuestDB integration test pinning wall-clock TTL (far-future row evicts only partitions older
  than the TTL by the wall clock), with a positive control.
- Docs: Historian.md, Configuration.md, security.md, CLAUDE.md (incl. QuestDB retry duplicates,
  #674.4 anonymous-readable trail with ACL off, #674.5 not implemented).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-11 11:18:44 -04:00

81 lines
3.9 KiB
C#

using Microsoft.Data.SqlClient;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
/// <summary>
/// The SQL Server historian as a conformance backend: a fresh unique schema per instance (inside the fixture's
/// throwaway database), seeded by plain INSERTs through <see cref="ReadTestDb"/>.
/// </summary>
internal sealed class SqlServerConformanceBackend : IConformanceBackend
{
private readonly ReadTestDb _db;
private SqlServerConformanceBackend(ReadTestDb db) => _db = db;
/// <summary>Creates a backend over a fresh schema; skips the calling test when the gate is closed.</summary>
public static async Task<IConformanceBackend> CreateAsync(SqlHistorianFixture fx) =>
new SqlServerConformanceBackend(await ReadTestDb.CreateAsync(fx));
/// <inheritdoc/>
public long ResolutionTicks => 1;
/// <inheritdoc/>
public IHistorianDataSource CreateDataSource(TimeProvider? clock = null) => _db.Source(clock);
/// <inheritdoc/>
public IHistorianDataSource CreateDataSource(int maxTimeWeightedRawRows) => _db.Source(null, maxTimeWeightedRawRows);
/// <inheritdoc/>
public IHistorianValueWriter CreateValueWriter(Func<DateTime>? utcNow = null) =>
new SqlServerHistorianValueWriter(_db.Conn, _db.Boot, new SqlServerTagCatalog(_db.Conn, _db.Boot), utcNow: utcNow);
/// <inheritdoc/>
public IHistorianDataUpdater CreateUpdater() =>
new SqlServerHistorianDataUpdater(_db.Conn, _db.Boot, new SqlServerTagCatalog(_db.Conn, _db.Boot));
/// <inheritdoc/>
public async Task SeedAsync(string tag, params (DateTime T, HistorianValue V, uint Status)[] samples) =>
await _db.SeedAsync(tag, samples);
/// <inheritdoc/>
public async Task<IReadOnlyList<(DateTime T, HistorianValue V, uint Status)>> ReadBackAsync(string tag)
{
await using var c = await _db.Conn.OpenAsync(CancellationToken.None);
await using var cmd = _db.Conn.CreateCommand(c,
$"SELECT s.SourceTimeUtc, s.ValueKind, s.NumValue, s.IntValue, s.StrValue, s.StatusCode FROM {_db.Conn.Q("Sample")} s " +
$"JOIN {_db.Conn.Q("Tag")} t ON t.TagId = s.TagId WHERE t.Name = @n ORDER BY s.SourceTimeUtc");
cmd.Parameters.Add(new SqlParameter("@n", System.Data.SqlDbType.NVarChar, -1) { Value = tag });
var list = new List<(DateTime, HistorianValue, uint)>();
await using var r = await cmd.ExecuteReaderAsync();
while (await r.ReadAsync())
{
var v = SampleColumnMapper.FromColumns(r.GetByte(1), r.IsDBNull(2) ? null : r.GetDouble(2),
r.IsDBNull(3) ? null : r.GetInt64(3), r.IsDBNull(4) ? null : r.GetString(4));
list.Add((SqlHistorianConnection.GetUtc(r, 0), v, SampleColumnMapper.FromDbStatus(r.GetInt32(5))));
}
return list;
}
/// <inheritdoc/>
public async Task<bool> TagExistsAsync(string tag) => await ScalarTagAsync("COUNT(*)", tag) is > 0;
/// <inheritdoc/>
public async Task<DriverDataType?> TagDataTypeAsync(string tag) =>
await ScalarTagAsync("MAX(CONVERT(bigint, DataType))", tag) is { } v ? (DriverDataType)(byte)v : null; // MAX over no row is NULL
private async Task<long?> ScalarTagAsync(string expr, string tag)
{
await using var c = await _db.Conn.OpenAsync(CancellationToken.None);
await using var cmd = _db.Conn.CreateCommand(c, $"SELECT {expr} FROM {_db.Conn.Q("Tag")} WHERE Name = @n");
cmd.Parameters.Add(new SqlParameter("@n", System.Data.SqlDbType.NVarChar, -1) { Value = tag });
var o = await cmd.ExecuteScalarAsync();
return o is null or DBNull ? null : Convert.ToInt64(o);
}
/// <inheritdoc/>
public ValueTask DisposeAsync() => ValueTask.CompletedTask; // the fixture drops the whole database
}