using System.Data.Common;
using System.Globalization;
using System.Text.Json.Nodes;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Browser;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests;
///
/// The Sql driver against a real MySQL 8 (lmxopcua#753), over the MySqlConnector provider
/// MySqlDialect ships — the MySQL counterpart of , narrowed the same way
/// to what a dialect can get wrong: liveness, the information_schema catalog SQL (a MySQL database
/// playing the schema), backtick quoting, the trailing LIMIT, an INT key compared against authored
/// text, case folding against case-SENSITIVE table names, the browse walk, the factory/probe provider gate —
/// and the named-query model, including the rolled-back transaction proven against a real side effect.
/// Env-gated; skipping is the normal outcome. See .
///
[Collection(MySqlCollection.Name)]
public sealed class MySqlReadTests
{
private readonly MySqlFixture _my;
/// Initializes a new instance of the class.
/// The seeded live-server fixture (collection-scoped).
public MySqlReadTests(MySqlFixture my) => _my = my;
/// Liveness + catalog load succeed for a SELECT-only login, and the endpoint is described credential-free.
[Fact]
public async Task InitializeAsync_realMySql_reportsHealthyAndDescribesTheEndpoint()
{
SkipUnlessLive();
await using var driver = await StartAsync(KeyValue("Sql/Line1/Speed", MySqlFixture.PresentKey));
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.Endpoint.ShouldContain(MySqlFixture.Schema);
driver.Endpoint.ShouldNotContain("password", Case.Insensitive);
driver.GetHostStatuses().ShouldHaveSingleItem().State.ShouldBe(HostState.Running);
}
/// A seeded key-value row round-trips Good, with its datetime(3) source timestamp stamped UTC.
[Fact]
public async Task ReadAsync_realMySql_roundTripsSeededKeyValue()
{
SkipUnlessLive();
await using var driver = await StartAsync(
KeyValue("Sql/Line1/Speed", MySqlFixture.PresentKey),
KeyValue("Sql/Line1/Temp", MySqlFixture.NullValueKey));
var snapshots = await driver.ReadAsync(["Sql/Line1/Speed", "Sql/Line1/Temp"], TestContext.Current.CancellationToken);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture).ShouldBe(MySqlFixture.PresentValue);
snapshots[0].SourceTimestampUtc.ShouldBe(MySqlFixture.PresentKeyTimestampUtc);
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Uncertain); // a present row with a NULL cell
}
///
/// topByTimestamp resolves to the newest row through the dialect's trailing LIMIT 1. The seed
/// makes the newest row neither first-inserted nor lowest-keyed, so losing either the limit or the
/// ORDER BY … DESC yields a wrong value.
///
[Fact]
public async Task ReadAsync_realMySql_topByTimestampSelectsTheNewestRow()
{
SkipUnlessLive();
await using var driver = await StartAsync(Tag("Sql/Newest/OvenTemp", new JsonObject
{
["model"] = "WideRow",
["table"] = MySqlFixture.Qualified("latest_status"),
["columnName"] = "oven_temp",
["rowSelector"] = new JsonObject { ["topByTimestamp"] = "sample_ts" },
}));
var snapshot = (await driver.ReadAsync(["Sql/Newest/OvenTemp"], TestContext.Current.CancellationToken))
.ShouldHaveSingleItem();
snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture).ShouldBe(210.25);
}
/// Authored key text compares against an INT key column, and against an INT where column.
[Fact]
public async Task ReadAsync_realMySql_integerKeyAndWhereColumnsCompareAgainstAuthoredText()
{
SkipUnlessLive();
await using var driver = await StartAsync(
Tag("Sql/IntKey/Tag101", new JsonObject
{
["model"] = "KeyValue",
["table"] = MySqlFixture.Qualified("int_tag_values"),
["keyColumn"] = "tag_id",
["keyValue"] = "101",
["valueColumn"] = "num_value",
}),
Tag("Sql/Station7/OvenTemp", new JsonObject
{
["model"] = "WideRow",
["table"] = MySqlFixture.Qualified("latest_status"),
["columnName"] = "oven_temp",
["rowSelector"] = new JsonObject { ["whereColumn"] = "station_id", ["whereValue"] = 7 },
}));
var snapshots = await driver.ReadAsync(
["Sql/IntKey/Tag101", "Sql/Station7/OvenTemp"], TestContext.Current.CancellationToken);
snapshots.ShouldAllBe(s => s.StatusCode == SqlStatusCodes.Good);
Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture).ShouldBe(77.5);
Convert.ToDouble(snapshots[1].Value, CultureInfo.InvariantCulture).ShouldBe(180.5);
}
///
/// An upper-cased authored table/column resolves to the catalog's lower-case spelling. MySQL table names are
/// case-SENSITIVE on Linux, so the backtick-quoted `TAG_VALUES` would name no table: the gate's
/// substitution is what makes mixed-case authoring work.
///
[Fact]
public async Task CatalogGate_realMySql_foldsAnUpperCasedAuthoredNameOntoTheCatalogSpelling()
{
SkipUnlessLive();
await using var driver = await StartAsync(Tag("Sql/Line1/Speed", new JsonObject
{
["model"] = "KeyValue",
["table"] = MySqlFixture.Qualified("tag_values").ToUpperInvariant(),
["keyColumn"] = "TAG_NAME",
["keyValue"] = MySqlFixture.PresentKey,
["valueColumn"] = "NUM_VALUE",
}));
var snapshot = (await driver.ReadAsync(["Sql/Line1/Speed"], TestContext.Current.CancellationToken))
.ShouldHaveSingleItem();
snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture).ShouldBe(MySqlFixture.PresentValue);
}
/// A table the login cannot see is rejected by the catalog gate (BadNodeIdUnknown), not queried.
[Fact]
public async Task CatalogGate_realMySql_rejectsATableTheCatalogDoesNotHold()
{
SkipUnlessLive();
await using var driver = await StartAsync(Tag("Sql/Nope", new JsonObject
{
["model"] = "KeyValue",
["table"] = MySqlFixture.Qualified("no_such_table"),
["keyColumn"] = "k",
["keyValue"] = "x",
["valueColumn"] = "v",
}));
(await driver.ReadAsync(["Sql/Nope"], TestContext.Current.CancellationToken))[0].StatusCode
.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
}
/// The AdminUI picker's walk — schemas → tables → columns → attributes — over a real MySQL.
[Fact]
public async Task Browse_realMySql_walksSchemaToTableToColumnAndTypesTheLeaf()
{
SkipUnlessLive();
var config = new JsonObject
{
["provider"] = nameof(SqlProvider.MySql),
["connectionString"] = _my.ConnectionString,
}.ToJsonString();
await using var session = await new SqlDriverBrowser().OpenAsync(config, TestContext.Current.CancellationToken);
var schema = (await session.RootAsync(TestContext.Current.CancellationToken)).ShouldHaveSingleItem(
"the fixture login holds SELECT on exactly one database");
schema.DisplayName.ShouldBe(MySqlFixture.Schema);
var tables = await session.ExpandAsync(schema.NodeId, TestContext.Current.CancellationToken);
tables.Select(t => t.DisplayName).ShouldContain("tag_values");
var columns = await session.ExpandAsync(
tables.First(t => t.DisplayName == "tag_values").NodeId, TestContext.Current.CancellationToken);
columns.Select(c => c.DisplayName).ShouldBe(["tag_name", "num_value", "sample_ts"]);
columns.ShouldAllBe(c => c.Kind == BrowseNodeKind.Leaf);
var attribute = (await session.AttributesAsync(
columns.Single(c => c.DisplayName == "num_value").NodeId, TestContext.Current.CancellationToken))
.ShouldHaveSingleItem();
attribute.DriverDataType.ShouldBe(nameof(DriverDataType.Float64)); // double
}
///
/// The production factory and Test Connect probe both admit MySql — the provider gate the three
/// switches used to disagree on — and a factory-built driver reads.
///
[Fact]
public async Task FactoryAndProbe_realMySql_constructAndConnect()
{
SkipUnlessLive();
var config = new JsonObject
{
["provider"] = nameof(SqlProvider.MySql),
["connectionStringRef"] = _my.ConnectionStringRef,
}.ToJsonString();
var probe = await new SqlDriverProbe().ProbeAsync(config, TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
probe.Ok.ShouldBeTrue(probe.Message);
await using var driver = await StartAsync(KeyValue("Sql/Line1/Speed", MySqlFixture.PresentKey));
(await driver.ReadAsync(["Sql/Line1/Speed"], TestContext.Current.CancellationToken))[0].StatusCode
.ShouldBe(SqlStatusCodes.Good);
}
// ---- named queries (design §5.4) against a real server ----
/// A named query with a bound parameter and a rowKey reads its row; the query bypasses the catalog.
[Fact]
public async Task NamedQuery_realMySql_readsTheKeyedRowWithABoundParameter()
{
SkipUnlessLive();
await using var driver = await StartAsync(
"""{"stations":{"sql":"SELECT station_id AS Station, oven_temp AS Temp FROM sqlpoll.latest_status WHERE pressure > @minPressure","params":{"minPressure":1.3}}}""",
Tag("Sql/Q/Newest", new JsonObject
{
["model"] = "Query", ["query"] = "stations", ["column"] = "temp",
["rowKey"] = new JsonObject { ["column"] = "station", ["value"] = 8 },
}),
Tag("Sql/Q/Filtered", new JsonObject
{
["model"] = "Query", ["query"] = "stations", ["column"] = "Temp",
["rowKey"] = new JsonObject { ["column"] = "Station", ["value"] = "7" },
}));
var snapshots = await driver.ReadAsync(["Sql/Q/Newest", "Sql/Q/Filtered"], TestContext.Current.CancellationToken);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture).ShouldBe(210.25);
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData); // pressure 1.2 is filtered out by the bind
}
///
/// The backstop proven against a real side effect: bump_counter() INSERTs a row and returns 1. A
/// SELECT bump_counter() passes the read-only guard (nothing lexical says "write"), the read is Good
/// (positive control: the function really ran), and the row is gone afterwards — the driver's transaction
/// was rolled back, not committed.
///
[Fact]
public async Task NamedQuery_realMySql_aWritingFunctionsEffectIsRolledBack()
{
SkipUnlessLive();
await using var driver = await StartAsync(
"""{"bump":{"sql":"SELECT sqlpoll.bump_counter() AS v"}}""",
Tag("Sql/Q/Bump", new JsonObject { ["model"] = "Query", ["query"] = "bump", ["column"] = "v" }));
var snapshot = (await driver.ReadAsync(["Sql/Q/Bump"], TestContext.Current.CancellationToken))[0];
snapshot.StatusCode.ShouldBe(SqlStatusCodes.Good);
Convert.ToInt64(snapshot.Value, CultureInfo.InvariantCulture).ShouldBe(1L);
(await _my.ScalarAsync("SELECT COUNT(*) FROM sqlpoll.side_effects")).ShouldBe(0L);
}
// ---- writes (design §3.4) against a real server ----
/// All gates open: the UPDATE lands on exactly the keyed row, and the driver reads it back.
[Fact]
public async Task Write_realMySql_updatesTheKeyedRowOnly()
{
SkipUnlessLive();
await using var driver = await StartAsync(null, allowWrites: true,
Setpoint("Sql/Sp/Speed", "Line1.SpeedSp", writable: true));
var target = 10.0 + Random.Shared.Next(1, 1000);
(await WriteAsync(driver, "Sql/Sp/Speed", target)).ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.SpeedSp'"),
CultureInfo.InvariantCulture).ShouldBe(target);
Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.TempSp'"),
CultureInfo.InvariantCulture).ShouldBe(20.0);
Convert.ToDouble((await driver.ReadAsync(["Sql/Sp/Speed"], TestContext.Current.CancellationToken))[0].Value,
CultureInfo.InvariantCulture).ShouldBe(target);
}
/// A key matching two rows is rolled back on InnoDB — neither row changes.
[Fact]
public async Task Write_realMySql_aKeyMatchingTwoRowsIsRolledBack()
{
SkipUnlessLive();
await using var driver = await StartAsync(null, allowWrites: true, Setpoint("Sql/Sp/Dup", "dup", writable: true));
(await WriteAsync(driver, "Sql/Sp/Dup", 99.0)).ShouldBe(SqlStatusCodes.BadConfigurationError);
(await _my.ScalarAsync("SELECT COUNT(*) FROM sqlpoll.setpoints WHERE tag_name = 'dup' AND sp_value = 99")).ShouldBe(0L);
}
///
/// Defence in depth: with every driver gate open, a table the login holds no UPDATE on is refused by the
/// server itself (the provider error surfaces as BadCommunicationError) and the row is unchanged.
///
[Fact]
public async Task Write_realMySql_aTableWithoutAnUpdateGrantIsRefusedByTheServer()
{
SkipUnlessLive();
await using var driver = await StartAsync(null, allowWrites: true, Tag("Sql/Kv/Speed", new JsonObject
{
["model"] = "KeyValue", ["table"] = MySqlFixture.Qualified("tag_values"), ["keyColumn"] = "tag_name",
["keyValue"] = MySqlFixture.PresentKey, ["valueColumn"] = "num_value", ["writable"] = true,
}));
(await WriteAsync(driver, "Sql/Kv/Speed", 1.0)).ShouldBe(SqlStatusCodes.BadCommunicationError);
Convert.ToDouble(await _my.ScalarRawAsync("SELECT num_value FROM sqlpoll.tag_values WHERE tag_name = 'Line1.Speed'"),
CultureInfo.InvariantCulture).ShouldBe(MySqlFixture.PresentValue);
}
/// The default: no allowWrites ⇒ refused before any connection, whatever the tag says.
[Fact]
public async Task Write_realMySql_withoutAllowWritesIsRefused()
{
SkipUnlessLive();
await using var driver = await StartAsync(null, allowWrites: false, Setpoint("Sql/Sp/Temp", "Line1.TempSp", writable: true));
(await WriteAsync(driver, "Sql/Sp/Temp", 77.0)).ShouldBe(SqlStatusCodes.BadNotWritable);
Convert.ToDouble(await _my.ScalarRawAsync("SELECT sp_value FROM sqlpoll.setpoints WHERE tag_name = 'Line1.TempSp'"),
CultureInfo.InvariantCulture).ShouldBe(20.0);
}
private static RawTagEntry Setpoint(string rawPath, string key, bool writable) => Tag(rawPath, new JsonObject
{
["model"] = "KeyValue",
["table"] = MySqlFixture.Qualified("setpoints"),
["keyColumn"] = "tag_name",
["keyValue"] = key,
["valueColumn"] = "sp_value",
["type"] = "Float64",
["writable"] = writable,
});
private static async Task WriteAsync(SqlDriver driver, string reference, object? value)
=> (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
.ShouldHaveSingleItem().StatusCode;
// ---- helpers ----
private void SkipUnlessLive()
{
if (_my.SkipReason is { } reason) Assert.Skip(reason);
}
private Task StartAsync(params RawTagEntry[] tags) => StartAsync(queriesJson: null, tags);
private Task StartAsync(string? queriesJson, params RawTagEntry[] tags)
=> StartAsync(queriesJson, allowWrites: false, tags);
///
/// Builds and initializes a driver through the PRODUCTION factory (provider: MySql, the connection
/// string resolved from the fixture's connectionStringRef environment variable), so the suite
/// exercises MySqlConnectorFactory.Instance and the factory's own named-query validation exactly as a
/// deployed config would.
///
private async Task StartAsync(string? queriesJson, bool allowWrites, params RawTagEntry[] tags)
{
var rawTags = new JsonArray();
foreach (var tag in tags)
{
rawTags.Add(new JsonObject
{
["rawPath"] = tag.RawPath,
["tagConfig"] = tag.TagConfig,
["writeIdempotent"] = tag.WriteIdempotent,
});
}
var config = new JsonObject
{
["provider"] = nameof(SqlProvider.MySql),
["connectionStringRef"] = _my.ConnectionStringRef,
["commandTimeout"] = "00:00:10",
["operationTimeout"] = "00:00:15",
["rawTags"] = rawTags,
};
if (queriesJson is not null) config["queries"] = JsonNode.Parse(queriesJson);
if (allowWrites) config["allowWrites"] = true;
var json = config.ToJsonString();
var driver = SqlDriverFactoryExtensions.CreateInstance("sql-mysql-integration", json);
try
{
await driver.InitializeAsync(json, TestContext.Current.CancellationToken);
}
catch
{
await driver.DisposeAsync();
throw;
}
return driver;
}
private static RawTagEntry KeyValue(string rawPath, string key) => Tag(rawPath, new JsonObject
{
["model"] = "KeyValue",
["table"] = MySqlFixture.Qualified("tag_values"),
["keyColumn"] = "tag_name",
["keyValue"] = key,
["valueColumn"] = "num_value",
["timestampColumn"] = "sample_ts",
});
private static RawTagEntry Tag(string rawPath, JsonObject config)
{
config["driver"] = "Sql";
return new RawTagEntry(rawPath, config.ToJsonString(), WriteIdempotent: false);
}
}
///
/// A real, seeded MySQL 8 database (Docker/mysql/docker-compose.yml + seed-mysql.sql).
/// Env-gated, checked before any socket is touched: nothing opens unless
/// and are both set. Like the Postgres fixture it
/// does not seed — the compose stack does, as root, so the suite's own login stays SELECT-only.
///
public sealed class MySqlFixture : IAsyncLifetime
{
/// Server endpoint in host:port form (e.g. 127.0.0.1:13306). Absent ⇒ skip.
public const string EndpointEnvVar = "SQL_MYSQL_ENDPOINT";
/// The login. Defaults to .
public const string UserEnvVar = "SQL_MYSQL_USER";
/// The password. Required — never defaulted.
public const string PasswordEnvVar = "SQL_MYSQL_PASSWORD";
/// The SELECT-only login the seed creates.
public const string DefaultUser = "otopcua";
/// MySQL's default port.
public const int DefaultPort = 3306;
/// The database (MySQL's "schema") every seeded object lives in.
public const string Schema = "sqlpoll";
/// A key whose row exists with .
public const string PresentKey = "Line1.Speed";
/// A key whose row exists with a NULL value cell.
public const string NullValueKey = "Line1.Temp";
/// The value seeded for .
public const double PresentValue = 42.0;
/// The source timestamp seeded for , read back as UTC.
public static readonly DateTime PresentKeyTimestampUtc = new(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc);
private const int ProbeConnectTimeoutSeconds = 5;
/// Why the suite is skipping, or null when the server is reachable.
public string? SkipReason { get; private set; }
/// The connection string handed to the code under test.
public string ConnectionString { get; private set; } = string.Empty;
///
/// The connectionStringRef the suite authors: the fixture publishes
/// under its environment variable for the session, exactly as a deployment provisions one.
///
public string ConnectionStringRef { get; } = $"MySqlIt{Guid.NewGuid():N}";
/// The server, for diagnostics; never carries credentials.
public string Server { get; private set; } = string.Empty;
/// Two-part name of a seeded object.
/// The bare table name.
/// sqlpoll.table.
public static string Qualified(string table) => $"{Schema}.{table}";
///
public async ValueTask InitializeAsync()
{
if (ResolveConnectionString() is not { } resolved)
{
SkipReason =
$"Set {EndpointEnvVar} (host:port) and {PasswordEnvVar} (and optionally {UserEnvVar}, default " +
$"'{DefaultUser}') to run the Sql driver's live MySQL suite. Bring the fixture up from " +
"Docker/mysql/docker-compose.yml first. No connection is attempted without these.";
return;
}
ConnectionString = resolved;
Environment.SetEnvironmentVariable(
SqlConnectionStringResolver.EnvironmentVariableFor(ConnectionStringRef), ConnectionString);
try
{
await ScalarAsync(new MySqlDialect().LivenessSql).ConfigureAwait(false);
}
catch (Exception ex)
{
SkipReason =
$"MySQL at '{Server}' was not reachable ({ex.GetType().Name}: {ex.Message}). Check " +
$"{EndpointEnvVar}/{UserEnvVar}/{PasswordEnvVar} and that the fixture stack is up, or unset them to skip.";
}
}
///
public ValueTask DisposeAsync()
{
Environment.SetEnvironmentVariable(SqlConnectionStringResolver.EnvironmentVariableFor(ConnectionStringRef), null);
return ValueTask.CompletedTask;
}
/// Runs a scalar query as the fixture login, for assertions on database state.
/// The statement.
/// The scalar, as a 64-bit integer when numeric.
public async Task