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(TestContext.Current.CancellationToken); 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(TestContext.Current.CancellationToken); (await r.ReadAsync(TestContext.Current.CancellationToken)).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(TestContext.Current.CancellationToken)).ShouldBeFalse(); await r.CloseAsync(); cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event"; Convert.ToInt32(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken)).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 An_event_reads_back_with_its_type_name_condition_name_and_alarm_id() { fx.SkipIfUnavailable(); var s = SqlHistorianFixture.UniqueSchema(); var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s }; var c = new SqlHistorianConnection(o); var boot = new SqlServerSchemaBootstrapper(c, o); var w = new SqlServerAlarmHistorianWriter(c, boot); (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]); using var src = new SqlServerHistorianDataSource(c, boot, new SqlServerTagCatalog(c, boot)); var e = (await src.ReadEventsAsync("Plant/Line/Eq", T.AddMinutes(-1), T.AddMinutes(1), 0, CancellationToken.None)) .Events.ShouldHaveSingleItem(); e.EventTypeName.ShouldBe("LimitAlarm"); e.ConditionName.ShouldBe("HiTemp"); e.AlarmId.ShouldBe("alarm-1"); } [Fact] public async Task A_batch_is_one_statement_per_chunk_and_a_redelivery_converges() { fx.SkipIfUnavailable(); var (w, s) = Make(); var events = Enumerable.Range(0, 400) .Select(i => new AlarmHistorianEvent("alarm-1", "Plant/Line/Eq", "HiTemp", "LimitAlarm", AlarmSeverity.High, "Activated", $"m{i}", "user", null, T.AddSeconds(i))) .ToList(); (await w.WriteBatchAsync(events, CancellationToken.None)).ShouldAllBe(o => o == HistorianWriteOutcome.Ack); w.Statements.ShouldBe(3, "400 events = 150 + 150 + 100, not 400 round trips"); // Redelivered (the at-least-once drain), with an in-batch duplicate: every event Acks, no row is added. var again = events.Append(events[0]).ToList(); var r = await w.WriteBatchAsync(again, CancellationToken.None); r.Count.ShouldBe(401); r.ShouldAllBe(o => o == HistorianWriteOutcome.Ack); await using var c = new SqlConnection(fx.ConnectionString); await c.OpenAsync(TestContext.Current.CancellationToken); await using var cmd = c.CreateCommand(); cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event"; Convert.ToInt32(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken)).ShouldBe(400); } [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(TestContext.Current.CancellationToken); await using var cmd = c.CreateCommand(); cmd.CommandText = $"SELECT COUNT(*) FROM [{s}].Event WHERE EventKind='Activated'"; Convert.ToInt32(await cmd.ExecuteScalarAsync(TestContext.Current.CancellationToken)).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(); } }