using System.Globalization; using System.Text.Json.Nodes; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts; namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests; /// /// The authored write block (lmxopcua#781 item 2) against real servers: Update with a stamp, Insert (and /// the duplicate-key / NOT NULL failure mapping), the dialect's Upsert, and a stored-procedure call — one suite, /// run per backend by the sealed subclasses below, each in its backend's existing collection and env gate. /// Every seed carries the same shape (Docker/*/seed-*.sql, and /// for SQL Server): write_targets(tag_name PK, sp_value NOT NULL, stamp_ts) holding one Present /// row, and set_target(p_value, p_key, p_ts), which journals its arguments into proc_calls. Each /// test writes a key of its own, so the suite needs no cleanup grant and tolerates re-runs. /// public abstract class SqlWriteModesLiveSuite { /// Why the backend is unavailable, or null. protected abstract string? SkipReason { get; } /// The suite login's connection string. protected abstract string ConnectionString { get; } /// The backend's dialect. protected abstract ISqlDialect Dialect { get; } /// The schema the seed lives in, as authored (the catalog gate folds its case). protected abstract string Schema { get; } /// The procedure as authored — deliberately in the case the backend does NOT store, so the gate's spelling substitution is exercised. protected abstract string AuthoredProcedure { get; } // ---- Update ---- [Fact] public async Task Update_withStamp_setsTheValueAndTheWriteTime_onThePresentRow() { SkipUnlessLive(); await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject { ["stampTimestamp"] = true })); var value = 100.0 + Random.Shared.Next(1, 1000); var before = DateTime.UtcNow.AddMinutes(-2); (await WriteAsync(driver, "Sp", value)).ShouldBe(SqlStatusCodes.Good); Convert.ToDouble(await ScalarAsync("SELECT sp_value FROM {0} WHERE tag_name = 'Present'", "write_targets"), CultureInfo.InvariantCulture).ShouldBe(value); var stamped = Convert.ToDateTime( await ScalarAsync("SELECT stamp_ts FROM {0} WHERE tag_name = 'Present'", "write_targets"), CultureInfo.InvariantCulture); stamped.ShouldBeGreaterThan(before); stamped.ShouldBeLessThan(DateTime.UtcNow.AddMinutes(2)); } [Fact] public async Task Update_toAnAbsentKey_isBadNoData_andInsertsNothing() { SkipUnlessLive(); var key = NewKey(); await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Update" })); (await WriteAsync(driver, "Sp", 1.0)).ShouldBe(SqlStatusCodes.BadNoData); (await CountAsync("write_targets", key)).ShouldBe(0); } // ---- Insert ---- [Fact] public async Task Insert_addsTheRow_andASecondInsertOfTheKey_isBadEntryExists() { SkipUnlessLive(); var key = NewKey(); await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Insert" })); (await WriteAsync(driver, "Sp", 5.25)).ShouldBe(SqlStatusCodes.Good); (await WriteAsync(driver, "Sp", 6.25)).ShouldBe(SqlStatusCodes.BadEntryExists); (await CountAsync("write_targets", key)).ShouldBe(1); Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "write_targets"), CultureInfo.InvariantCulture).ShouldBe(5.25); } [Fact] public async Task Insert_ofNull_intoANotNullColumn_isBadOutOfRange() { SkipUnlessLive(); var key = NewKey(); await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Insert" })); (await WriteAsync(driver, "Sp", null)).ShouldBe(SqlStatusCodes.BadOutOfRange); (await CountAsync("write_targets", key)).ShouldBe(0); } // ---- Upsert ---- [Fact] public async Task Upsert_insertsAnAbsentKey_thenUpdatesIt_andRepeatingTheSameValueStillSucceeds() { SkipUnlessLive(); var key = NewKey(); await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Upsert", ["stampTimestamp"] = true })); (await WriteAsync(driver, "Sp", 1.5)).ShouldBe(SqlStatusCodes.Good); (await WriteAsync(driver, "Sp", 2.5)).ShouldBe(SqlStatusCodes.Good); (await WriteAsync(driver, "Sp", 2.5)).ShouldBe(SqlStatusCodes.Good); // MySQL: unchanged row, found-rows = 1 (await CountAsync("write_targets", key)).ShouldBe(1); Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "write_targets"), CultureInfo.InvariantCulture).ShouldBe(2.5); Convert.ToDouble((await driver.ReadAsync(["Sp"], TestContext.Current.CancellationToken))[0].Value, CultureInfo.InvariantCulture).ShouldBe(2.5); } // ---- Procedure ---- [Fact] public async Task Procedure_isCalled_withTheValueKeyAndWriteTime() { SkipUnlessLive(); var key = NewKey(); await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject { ["mode"] = "Procedure", ["procedure"] = AuthoredProcedure, ["valueParam"] = "p_value", ["keyParam"] = "p_key", ["timestampParam"] = "p_ts", }, keyValue: key)); var before = DateTime.UtcNow.AddMinutes(-2); (await WriteAsync(driver, "Sp", 7.75)).ShouldBe(SqlStatusCodes.Good); (await CountAsync("proc_calls", key)).ShouldBe(1); Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "proc_calls"), CultureInfo.InvariantCulture).ShouldBe(7.75); var stamp = await ScalarAsync($"SELECT stamp_ts FROM {{0}} WHERE tag_name = '{key}'", "proc_calls"); (stamp is DateTimeOffset dto ? dto.UtcDateTime : Convert.ToDateTime(stamp, CultureInfo.InvariantCulture)) .ShouldBeGreaterThan(before); } [Fact] public async Task Procedure_thatDoesNotExist_isRejectedByTheCatalogGate() { SkipUnlessLive(); await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject { ["mode"] = "Procedure", ["procedure"] = Schema + ".no_such_proc", })); (await WriteAsync(driver, "Sp", 1.0)).ShouldBe(SqlStatusCodes.BadNodeIdUnknown); } // ---- helpers ---- private void SkipUnlessLive() { if (SkipReason is { } reason) Assert.Skip(reason); } private static string NewKey() => "W-" + Guid.NewGuid().ToString("N")[..12]; private RawTagEntry Target(string rawPath, string key, JsonObject write, string? keyValue = null) { var config = new JsonObject { ["driver"] = "Sql", ["model"] = "KeyValue", ["table"] = Schema + ".write_targets", ["keyColumn"] = "tag_name", ["keyValue"] = keyValue ?? key, ["valueColumn"] = "sp_value", ["timestampColumn"] = "stamp_ts", ["type"] = "Float64", ["writable"] = true, ["write"] = write, }; return new RawTagEntry(rawPath, config.ToJsonString(), WriteIdempotent: false); } private async Task StartAsync(params RawTagEntry[] tags) { var driver = new SqlDriver( new SqlDriverOptions { RawTags = tags, CommandTimeout = TimeSpan.FromSeconds(10), OperationTimeout = TimeSpan.FromSeconds(15), AllowWrites = true, }, driverInstanceId: "sql-write-modes-" + Dialect.Provider, dialect: Dialect, connectionString: ConnectionString); try { await driver.InitializeAsync("{}", TestContext.Current.CancellationToken); } catch { await driver.DisposeAsync(); throw; } return driver; } private static async Task WriteAsync(SqlDriver driver, string reference, object? value) => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken)) .ShouldHaveSingleItem().StatusCode; private async Task CountAsync(string table, string key) => Convert.ToInt64(await ScalarAsync($"SELECT COUNT(*) FROM {{0}} WHERE tag_name = '{key}'", table), CultureInfo.InvariantCulture); /// Runs a scalar as the suite login; {0} is replaced by the schema-qualified table. private async Task ScalarAsync(string sqlFormat, string table) { var connection = Dialect.Factory.CreateConnection()!; await using (connection.ConfigureAwait(false)) { connection.ConnectionString = ConnectionString; await connection.OpenAsync(TestContext.Current.CancellationToken).ConfigureAwait(false); var command = connection.CreateCommand(); await using (command.ConfigureAwait(false)) { command.CommandText = string.Format(CultureInfo.InvariantCulture, sqlFormat, Schema + "." + table); var value = await command.ExecuteScalarAsync(TestContext.Current.CancellationToken).ConfigureAwait(false); return value is DBNull ? null : value; } } } } /// SQL Server leg — the seeding fixture on the shared central server (its own database). [Collection(SqlPollServerCollection.Name)] public sealed class SqlServerWriteModesTests(SqlPollServerFixture fixture) : SqlWriteModesLiveSuite { protected override string? SkipReason => fixture.SkipReason; protected override string ConnectionString => fixture.ConnectionString; protected override ISqlDialect Dialect { get; } = new SqlServerDialect(); protected override string Schema => SqlPollServerFixture.Schema; protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET"; } /// PostgreSQL leg — the granted-not-owning login; upper-case authoring folds to the stored lower case. [Collection(PostgresCollection.Name)] public sealed class PostgresWriteModesTests(PostgresFixture fixture) : SqlWriteModesLiveSuite { protected override string? SkipReason => fixture.SkipReason; protected override string ConnectionString => fixture.ConnectionString; protected override ISqlDialect Dialect { get; } = new PostgresDialect(); protected override string Schema => PostgresFixture.Schema; protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET"; } /// MySQL leg — positional CALL, ON DUPLICATE KEY UPDATE's 1-or-2 row count. [Collection(MySqlCollection.Name)] public sealed class MySqlWriteModesTests(MySqlFixture fixture) : SqlWriteModesLiveSuite { protected override string? SkipReason => fixture.SkipReason; protected override string ConnectionString => fixture.ConnectionString; protected override ISqlDialect Dialect { get; } = new MySqlDialect(); protected override string Schema => MySqlFixture.Schema; protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET"; } /// Oracle leg — MERGE from DUAL, an anonymous PL/SQL block; lower-case authoring folds to the stored upper case. [Collection(OracleCollection.Name)] public sealed class OracleWriteModesTests(OracleFixture fixture) : SqlWriteModesLiveSuite { protected override string? SkipReason => fixture.SkipReason; protected override string ConnectionString => fixture.ConnectionString; protected override ISqlDialect Dialect { get; } = new OracleDialect(); protected override string Schema => OracleFixture.Schema; protected override string AuthoredProcedure => "otopcua.set_target"; }