Merge ph-T9 (pluggable historian)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+89
@@ -0,0 +1,89 @@
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
|
||||
|
||||
[Collection(SqlHistorianCollection.Name)]
|
||||
public sealed class AlarmWriterTests(SqlHistorianFixture fx)
|
||||
{
|
||||
private static readonly DateTime T = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1234567);
|
||||
|
||||
private (SqlServerAlarmHistorianWriter W, string S) Make(string? connection = null, bool auto = true)
|
||||
{
|
||||
var s = SqlHistorianFixture.UniqueSchema();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = connection ?? fx.ConnectionString, Schema = s, AutoCreateSchema = auto };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
return (new SqlServerAlarmHistorianWriter(c, new SqlServerSchemaBootstrapper(c, o)), s);
|
||||
}
|
||||
|
||||
private static AlarmHistorianEvent Ev(string kind = "Activated", string type = "LimitAlarm", string? comment = "c") =>
|
||||
new("alarm-1", "Plant/Line/Eq", "HiTemp", type, AlarmSeverity.High, kind, "msg", "user", comment, T);
|
||||
|
||||
[Fact]
|
||||
public async Task Event_round_trips_and_a_repeat_is_one_row_with_both_Ack()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
var first = await w.WriteBatchAsync([Ev(), Ev()], CancellationToken.None);
|
||||
first.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]);
|
||||
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
|
||||
(await w.WriteBatchAsync([Ev(kind: "Cleared")], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
|
||||
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, Message, UserName, Comment, EventTimeUtc, ReceivedTimeUtc FROM [{s}].Event WHERE EventKind='Activated'";
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
(await r.ReadAsync()).ShouldBeTrue();
|
||||
r.GetString(0).ShouldBe("Plant/Line/Eq");
|
||||
r.GetString(1).ShouldBe("alarm-1");
|
||||
r.GetString(2).ShouldBe("HiTemp");
|
||||
r.GetString(3).ShouldBe("LimitAlarm");
|
||||
r.GetInt32(4).ShouldBe(700);
|
||||
r.GetString(6).ShouldBe("msg");
|
||||
r.GetString(7).ShouldBe("user");
|
||||
r.GetString(8).ShouldBe("c");
|
||||
SqlHistorianConnection.GetUtc(r, 9).Ticks.ShouldBe(T.Ticks);
|
||||
(DateTime.UtcNow - SqlHistorianConnection.GetUtc(r, 10)).ShouldBeLessThan(TimeSpan.FromMinutes(1));
|
||||
(await r.ReadAsync()).ShouldBeFalse();
|
||||
await r.CloseAsync();
|
||||
cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event";
|
||||
Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(2);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Over_long_AlarmTypeName_is_PermanentFail_and_siblings_still_Ack()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", type: new string('t', 101)), Ev(kind: "C", comment: null)], CancellationToken.None);
|
||||
r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
await w.WriteBatchAsync([Ev(kind: "warm")], CancellationToken.None);
|
||||
var rs = await Task.WhenAll(Enumerable.Range(0, 12).Select(_ => Task.Run(() => w.WriteBatchAsync([Ev()], CancellationToken.None))));
|
||||
rs.ShouldAllBe(x => x[0] == HistorianWriteOutcome.Ack);
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event WHERE EventKind='Activated'";
|
||||
Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Unreachable_database_is_RetryPlease_not_an_exception_and_empty_batch_is_empty()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, _) = Make(fx.ConnectionString + ";Connect Timeout=1", auto: false); // schema missing
|
||||
(await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.RetryPlease]);
|
||||
(await w.WriteBatchAsync([], CancellationToken.None)).ShouldBeEmpty();
|
||||
}
|
||||
}
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
|
||||
|
||||
[Collection(SqlHistorianCollection.Name)]
|
||||
public sealed class ProvisioningTests(SqlHistorianFixture fx)
|
||||
{
|
||||
private (SqlServerTagProvisioner P, string S) Make(string? schema = null)
|
||||
{
|
||||
var s = schema ?? SqlHistorianFixture.UniqueSchema();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
return (new SqlServerTagProvisioner(c, new SqlServerSchemaBootstrapper(c, o)), s);
|
||||
}
|
||||
|
||||
private async Task<(byte Type, string? Eu, string? Desc)> TagAsync(string s, string name)
|
||||
{
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT DataType, EngineeringUnit, Description FROM [{s}].Tag WHERE Name=@n";
|
||||
cmd.Parameters.AddWithValue("@n", name);
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
(await r.ReadAsync()).ShouldBeTrue();
|
||||
return (r.GetByte(0), r.IsDBNull(1) ? null : r.GetString(1), r.IsDBNull(2) ? null : r.GetString(2));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Upsert_creates_then_changes_DataType_on_rerun_and_every_type_is_accepted()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, s) = Make();
|
||||
var r1 = await p.EnsureTagsAsync([new("a", DriverDataType.Boolean, "degC", "desc"), new("s", DriverDataType.String, null, null)], CancellationToken.None);
|
||||
r1.ShouldBe(new HistorianProvisionResult(2, 2, 0, 0));
|
||||
(await TagAsync(s, "a")).ShouldBe(((byte)DriverDataType.Boolean, "degC", "desc"));
|
||||
var r2 = await p.EnsureTagsAsync([new("a", DriverDataType.Float64, null, null)], CancellationToken.None);
|
||||
r2.ShouldBe(new HistorianProvisionResult(1, 1, 0, 0));
|
||||
(await TagAsync(s, "a")).ShouldBe(((byte)DriverDataType.Float64, "degC", "desc")); // null keeps stored metadata
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Duplicate_names_and_bad_names_are_tallied_not_thrown()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, _) = Make();
|
||||
var r = await p.EnsureTagsAsync([
|
||||
new("d", DriverDataType.Int32, null, null), new("d", DriverDataType.Int64, null, null),
|
||||
new("", DriverDataType.Int32, null, null), new(new string('x', 401), DriverDataType.Int32, null, null)], CancellationToken.None);
|
||||
r.ShouldBe(new HistorianProvisionResult(4, 2, 0, 2));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task More_tags_than_one_chunk_all_land()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (p, s) = Make();
|
||||
var reqs = Enumerable.Range(0, 1000).Select(i => new HistorianTagProvisionRequest($"bulk{i}", DriverDataType.Float32, null, null)).ToList();
|
||||
(await p.EnsureTagsAsync(reqs, CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1000, 1000, 0, 0));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Two_concurrent_provisioners_of_the_same_new_tags_both_report_Ensured_with_no_Failed()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var s = SqlHistorianFixture.UniqueSchema();
|
||||
var (p0, _) = Make(s);
|
||||
await p0.EnsureTagsAsync([new("warm", DriverDataType.Int32, null, null)], CancellationToken.None); // bootstrap once
|
||||
for (var round = 0; round < 20; round++)
|
||||
{
|
||||
var ps = Enumerable.Range(0, 16).Select(_ => Make(s).P).ToList();
|
||||
// Warm each instance (its own bootstrap-once) and release all together, else bootstrap serialisation staggers the race away.
|
||||
foreach (var p in ps) await p.EnsureTagsAsync([new("warm", DriverDataType.Int32, null, null)], CancellationToken.None);
|
||||
var gate = new TaskCompletionSource();
|
||||
// Staggered windows over one pool: every worker's FIRST name differs, so a MERGE without HOLDLOCK cannot hide behind
|
||||
// a lock convoy on a shared first key and the overlap really races (union per round = 6 names; the single/few-row shape is the one T8 proved races).
|
||||
var tasks = ps.Select((p, k) => Task.Run(async () =>
|
||||
{
|
||||
var reqs = Enumerable.Range(k % 4, 3).Select(i => new HistorianTagProvisionRequest($"r{round}.t{i}", DriverDataType.Float64, "u", null)).ToList();
|
||||
await gate.Task;
|
||||
return await p.EnsureTagsAsync(reqs, CancellationToken.None);
|
||||
})).ToList();
|
||||
gate.SetResult();
|
||||
var rs = await Task.WhenAll(tasks);
|
||||
foreach (var r in rs) r.ShouldBe(new HistorianProvisionResult(3, 3, 0, 0));
|
||||
// The production dup-key retry would mask a missing HOLDLOCK; HOLDLOCK must mean it never fires.
|
||||
ps.Sum(p => p.DupKeyRetryCount).ShouldBe(0);
|
||||
}
|
||||
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Tag";
|
||||
Convert.ToInt32(await cmd.ExecuteScalarAsync()).ShouldBe(20 * 6 + 1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_connection_failure_is_a_Failed_tally_not_an_exception()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString + ";Connect Timeout=1", Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
var p = new SqlServerTagProvisioner(c, new SqlServerSchemaBootstrapper(c, o));
|
||||
(await p.EnsureTagsAsync([new("x", DriverDataType.Int32, null, null)], CancellationToken.None)).ShouldBe(new HistorianProvisionResult(1, 0, 0, 1));
|
||||
}
|
||||
}
|
||||
+173
@@ -0,0 +1,173 @@
|
||||
using System.Diagnostics;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
|
||||
|
||||
[Collection(SqlHistorianCollection.Name)]
|
||||
public sealed class WriteTests(SqlHistorianFixture fx)
|
||||
{
|
||||
// Non-ms-aligned: sub-millisecond ticks must round-trip exactly.
|
||||
private static readonly DateTime T0 = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1234567);
|
||||
|
||||
private (SqlServerHistorianValueWriter W, string S) Make()
|
||||
{
|
||||
var s = SqlHistorianFixture.UniqueSchema();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
var b = new SqlServerSchemaBootstrapper(c, o);
|
||||
return (new SqlServerHistorianValueWriter(c, b, new SqlServerTagCatalog(c, b)), s);
|
||||
}
|
||||
|
||||
private async Task<List<(DateTime T, byte Kind, double? N, long? I, string? S, int Status, string Tag)>> RowsAsync(string schema, string tag)
|
||||
{
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT s.SourceTimeUtc, s.ValueKind, s.NumValue, s.IntValue, s.StrValue, s.StatusCode, t.Name FROM [{schema}].Sample s JOIN [{schema}].Tag t ON t.TagId=s.TagId WHERE t.Name=@n ORDER BY s.SourceTimeUtc";
|
||||
cmd.Parameters.AddWithValue("@n", tag);
|
||||
var l = new List<(DateTime, byte, double?, long?, string?, int, string)>();
|
||||
await using var r = await cmd.ExecuteReaderAsync();
|
||||
while (await r.ReadAsync())
|
||||
l.Add((SqlHistorianConnection.GetUtc(r, 0), r.GetByte(1), r.IsDBNull(2) ? null : r.GetDouble(2), r.IsDBNull(3) ? null : r.GetInt64(3),
|
||||
r.IsDBNull(4) ? null : r.GetString(4), r.GetInt32(5), r.GetString(6)));
|
||||
return l;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Every_kind_round_trips_with_exact_ticks_and_status()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
var dt = new DateTime(2026, 1, 2, 3, 4, 5, DateTimeKind.Utc).AddTicks(77);
|
||||
HistorizationValue[] vals =
|
||||
[
|
||||
new(T0, HistorianValue.FromDouble(1.25), 0),
|
||||
new(T0.AddTicks(1), HistorianValue.FromInt64(long.MaxValue), 0x40000000u),
|
||||
new(T0.AddTicks(2), HistorianValue.FromBoolean(true), 0),
|
||||
new(T0.AddTicks(3), HistorianValue.FromString("héllo"), 0),
|
||||
new(T0.AddTicks(4), HistorianValue.FromDateTime(dt), 0),
|
||||
new(T0.AddTicks(5), HistorianValue.Empty, 0x80310000u),
|
||||
];
|
||||
(await w.WriteLiveValuesAsync("tag.a", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(s, "tag.a");
|
||||
rows.Count.ShouldBe(6);
|
||||
for (var i = 0; i < 6; i++) rows[i].T.Ticks.ShouldBe(vals[i].TimestampUtc!.Value.Ticks);
|
||||
var back = rows.Select(r => SampleRowMapper.FromColumns(r.Kind, r.N, r.I, r.S)).ToList();
|
||||
back.ShouldBe(vals.Select(v => v.Value).ToList());
|
||||
(rows[5].N, rows[5].I, rows[5].S).ShouldBe((null, null, null));
|
||||
rows[1].Status.ShouldBe(unchecked((int)0x40000000u));
|
||||
SampleRowMapper.FromDbStatus(rows[5].Status).ShouldBe(0x80310000u);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Writing_the_same_batch_twice_is_idempotent()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
HistorizationValue[] vals = [new(T0, HistorianValue.FromDouble(1), 0), new(T0.AddSeconds(1), HistorianValue.FromDouble(2), 0)];
|
||||
(await w.WriteLiveValuesAsync("tag.b", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.b", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
// Overlapping batch: only the new row lands, the stored first value is untouched.
|
||||
(await w.WriteLiveValuesAsync("tag.b", [new(T0, HistorianValue.FromDouble(99), 0), new(T0.AddSeconds(2), HistorianValue.FromDouble(3), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(s, "tag.b");
|
||||
rows.Count.ShouldBe(3);
|
||||
rows[0].N.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Within_batch_duplicate_timestamps_do_not_fail_and_first_wins()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
(await w.WriteLiveValuesAsync("tag.c", [new(T0, HistorianValue.FromDouble(1), 0), new(T0, HistorianValue.FromDouble(2), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(s, "tag.c");
|
||||
rows.Count.ShouldBe(1);
|
||||
rows[0].N.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Non_finite_doubles_store_as_Empty_with_their_status_and_do_not_fail_the_write()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
(await w.WriteLiveValuesAsync("tag.d", [
|
||||
new(T0, HistorianValue.FromDouble(double.NaN), 0x40000000u),
|
||||
new(T0.AddTicks(1), HistorianValue.FromDouble(double.PositiveInfinity), 0),
|
||||
new(T0.AddTicks(2), HistorianValue.FromDouble(7), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(s, "tag.d");
|
||||
rows.Count.ShouldBe(3);
|
||||
rows[0].Kind.ShouldBe((byte)HistorianValueKind.Empty);
|
||||
rows[0].N.ShouldBeNull();
|
||||
rows[0].Status.ShouldBe(unchecked((int)0x40000000u));
|
||||
rows[2].N.ShouldBe(7);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Empty_batch_succeeds_without_a_call_and_null_timestamp_is_stamped()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
(await w.WriteLiveValuesAsync("tag.e", [], CancellationToken.None)).ShouldBeTrue();
|
||||
(await w.WriteLiveValuesAsync("tag.e", [new(null, HistorianValue.FromInt64(5), 0)], CancellationToken.None)).ShouldBeTrue();
|
||||
var rows = await RowsAsync(s, "tag.e");
|
||||
rows.Count.ShouldBe(1);
|
||||
(DateTime.UtcNow - rows[0].T).ShouldBeLessThan(TimeSpan.FromMinutes(1));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Tag_is_created_with_the_first_typed_values_data_type()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
await w.WriteLiveValuesAsync("tag.f", [new(T0, HistorianValue.Empty, 0x80310000u), new(T0.AddTicks(1), HistorianValue.FromString("x"), 0)], CancellationToken.None);
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"SELECT DataType FROM [{s}].Tag WHERE Name='tag.f'";
|
||||
Convert.ToByte(await cmd.ExecuteScalarAsync()).ShouldBe((byte)DriverDataType.String);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Ten_thousand_values_write_in_under_five_seconds()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
await w.WriteLiveValuesAsync("tag.warm", [new(T0, HistorianValue.FromDouble(0), 0)], CancellationToken.None); // bootstrap
|
||||
var vals = Enumerable.Range(0, 10_000).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromDouble(i), 0)).ToArray();
|
||||
var sw = Stopwatch.StartNew();
|
||||
(await w.WriteLiveValuesAsync("tag.big", vals, CancellationToken.None)).ShouldBeTrue();
|
||||
sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5));
|
||||
(await RowsAsync(s, "tag.big")).Count.ShouldBe(10_000);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Concurrent_overlapping_writers_all_succeed_and_leave_one_row_per_timestamp()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (w, s) = Make();
|
||||
var vals = Enumerable.Range(0, 200).Select(i => new HistorizationValue(T0.AddMilliseconds(i), HistorianValue.FromDouble(i), 0)).ToArray();
|
||||
// A deadlock victim (1205) legitimately returns false and the recorder retries; model that, and require convergence.
|
||||
var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => Task.Run(async () =>
|
||||
{
|
||||
for (var i = 0; i < 5; i++)
|
||||
if (await w.WriteLiveValuesAsync("tag.race", vals, CancellationToken.None)) return true;
|
||||
return false;
|
||||
})));
|
||||
results.ShouldAllBe(r => r);
|
||||
(await RowsAsync(s, "tag.race")).Count.ShouldBe(200);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_SQL_failure_returns_false_instead_of_throwing()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString + ";Connect Timeout=1", Schema = SqlHistorianFixture.UniqueSchema(), AutoCreateSchema = false };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
var b = new SqlServerSchemaBootstrapper(c, o);
|
||||
var w = new SqlServerHistorianValueWriter(c, b, new SqlServerTagCatalog(c, b));
|
||||
(await w.WriteLiveValuesAsync("x", [new(T0, HistorianValue.FromDouble(1), 0)], CancellationToken.None)).ShouldBeFalse();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests;
|
||||
|
||||
public sealed class AlarmEventIdTests
|
||||
{
|
||||
private static readonly DateTime T = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc);
|
||||
|
||||
private static AlarmHistorianEvent Ev(string path = "P", string id = "A", string kind = "Activated", string user = "u",
|
||||
string? comment = "c", string msg = "m", DateTime? t = null) =>
|
||||
new(id, path, "n", "LimitAlarm", AlarmSeverity.High, kind, msg, user, comment, t ?? T);
|
||||
|
||||
[Fact]
|
||||
public void Id_is_stable_lowercase_hex_sha256()
|
||||
{
|
||||
var a = SqlServerAlarmHistorianWriter.ComputeEventId(Ev());
|
||||
a.ShouldBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev()));
|
||||
a.Length.ShouldBe(64);
|
||||
a.ShouldBe(a.ToLowerInvariant());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Field_boundaries_are_unambiguous()
|
||||
{
|
||||
// The delimiter-based scheme collided on these; length prefixes must not.
|
||||
SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a|b", id: "c"))
|
||||
.ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a", id: "b|c")));
|
||||
SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "ab", id: "c"))
|
||||
.ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(path: "a", id: "bc")));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Null_and_empty_differ_and_every_field_participates()
|
||||
{
|
||||
var basis = SqlServerAlarmHistorianWriter.ComputeEventId(Ev());
|
||||
SqlServerAlarmHistorianWriter.ComputeEventId(Ev(comment: null)).ShouldNotBe(SqlServerAlarmHistorianWriter.ComputeEventId(Ev(comment: "")));
|
||||
foreach (var other in new[]
|
||||
{
|
||||
Ev(path: "Q"), Ev(id: "B"), Ev(kind: "Cleared"), Ev(user: "v"), Ev(comment: "d"), Ev(msg: "x"),
|
||||
Ev(t: T.AddTicks(1)),
|
||||
})
|
||||
SqlServerAlarmHistorianWriter.ComputeEventId(other).ShouldNotBe(basis);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(typeof(InvalidOperationException))]
|
||||
[InlineData(typeof(ArgumentException))]
|
||||
[InlineData(typeof(FormatException))]
|
||||
public void Non_SQL_exceptions_retry_rather_than_dead_letter(Type type) =>
|
||||
SqlServerAlarmHistorianWriter.Classify((Exception)Activator.CreateInstance(type)!)
|
||||
.ShouldBe(HistorianWriteOutcome.RetryPlease);
|
||||
}
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests;
|
||||
|
||||
public sealed class ValueWriterPrepareTests
|
||||
{
|
||||
private static readonly DateTime T = new(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc);
|
||||
|
||||
[Fact]
|
||||
public void Non_finite_doubles_become_Empty_keeping_status()
|
||||
{
|
||||
const uint status = 0x40000000u;
|
||||
var rows = SqlServerHistorianValueWriter.Prepare(
|
||||
[
|
||||
new(T, HistorianValue.FromDouble(double.NaN), status),
|
||||
new(T.AddTicks(1), HistorianValue.FromDouble(double.PositiveInfinity), 0),
|
||||
new(T.AddTicks(2), HistorianValue.FromDouble(double.NegativeInfinity), 0),
|
||||
new(T.AddTicks(3), HistorianValue.FromDouble(1.5), 0),
|
||||
], () => T);
|
||||
rows.Select(r => r.Value.Kind).ShouldBe([HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Empty, HistorianValueKind.Double]);
|
||||
rows[0].StatusCode.ShouldBe(status);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Duplicate_timestamps_keep_the_first_and_null_timestamp_uses_the_clock()
|
||||
{
|
||||
var rows = SqlServerHistorianValueWriter.Prepare(
|
||||
[
|
||||
new(T, HistorianValue.FromInt64(1), 0), new(T, HistorianValue.FromInt64(2), 0),
|
||||
new(null, HistorianValue.FromInt64(3), 0),
|
||||
], () => T.AddDays(1));
|
||||
rows.Count.ShouldBe(2);
|
||||
rows[0].Value.Integer.ShouldBe(1);
|
||||
rows[1].TimeUtc.ShouldBe(T.AddDays(1));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Data_type_mapping_is_one_table()
|
||||
{
|
||||
SqlHistorianDataTypes.FromKind(HistorianValueKind.Double).ShouldBe((byte)DriverDataType.Float64);
|
||||
SqlHistorianDataTypes.FromKind(HistorianValueKind.Boolean).ShouldBe((byte)DriverDataType.Boolean);
|
||||
SqlHistorianDataTypes.FromKind(HistorianValueKind.Empty).ShouldBe((byte)DriverDataType.Float64);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user