lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlWriteModesLiveTests.cs

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;

/// <summary>
///     The authored <c>write</c> 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.
///     <para>Every seed carries the same shape (<c>Docker/*/seed-*.sql</c>, and <see cref="SqlPollServerFixture"/>
///     for SQL Server): <c>write_targets(tag_name PK, sp_value NOT NULL, stamp_ts)</c> holding one <c>Present</c>
///     row, and <c>set_target(p_value, p_key, p_ts)</c>, which journals its arguments into <c>proc_calls</c>. Each
///     test writes a key of its own, so the suite needs no cleanup grant and tolerates re-runs.</para>
/// </summary>
public abstract class SqlWriteModesLiveSuite
{
    /// <summary>Why the backend is unavailable, or null.</summary>
    protected abstract string? SkipReason { get; }

    /// <summary>The suite login's connection string.</summary>
    protected abstract string ConnectionString { get; }

    /// <summary>The backend's dialect.</summary>
    protected abstract ISqlDialect Dialect { get; }

    /// <summary>The schema the seed lives in, as authored (the catalog gate folds its case).</summary>
    protected abstract string Schema { get; }

    /// <summary>The procedure as authored — deliberately in the case the backend does NOT store, so the gate's spelling substitution is exercised.</summary>
    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<SqlDriver> 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<uint> WriteAsync(SqlDriver driver, string reference, object? value)
        => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
            .ShouldHaveSingleItem().StatusCode;

    private async Task<long> CountAsync(string table, string key)
        => Convert.ToInt64(await ScalarAsync($"SELECT COUNT(*) FROM {{0}} WHERE tag_name = '{key}'", table),
            CultureInfo.InvariantCulture);

    /// <summary>Runs a scalar as the suite login; <c>{0}</c> is replaced by the schema-qualified table.</summary>
    private async Task<object?> 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;
            }
        }
    }
}

/// <summary>SQL Server leg — the seeding fixture on the shared central server (its own database).</summary>
[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";
}

/// <summary>PostgreSQL leg — the granted-not-owning login; upper-case authoring folds to the stored lower case.</summary>
[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";
}

/// <summary>MySQL leg — positional CALL, ON DUPLICATE KEY UPDATE's 1-or-2 row count.</summary>
[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";
}

/// <summary>Oracle leg — MERGE from DUAL, an anonymous PL/SQL block; lower-case authoring folds to the stored upper case.</summary>
[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";
}