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 ScalarAsync(string sql) { var value = await ScalarRawAsync(sql).ConfigureAwait(false); return value is IConvertible and not string ? Convert.ToInt64(value, CultureInfo.InvariantCulture) : value; } /// Runs a scalar query as the fixture login and returns the provider's value unconverted. /// The statement. /// The scalar. public async Task ScalarRawAsync(string sql) { var connection = new MySqlDialect().Factory.CreateConnection() ?? throw new InvalidOperationException("MySqlConnector returned no connection object."); await using (connection.ConfigureAwait(false)) { connection.ConnectionString = ConnectionString; await connection.OpenAsync().ConfigureAwait(false); var command = connection.CreateCommand(); await using (command.ConfigureAwait(false)) { command.CommandText = sql; return await command.ExecuteScalarAsync().ConfigureAwait(false); } } } /// /// Composes the connection string through the dialect's own factory builder (no direct MySqlConnector /// reference in this project — see the csproj note), or returns null when unconfigured. /// private string? ResolveConnectionString() { if (Environment.GetEnvironmentVariable(EndpointEnvVar) is not { Length: > 0 } endpoint) return null; if (Environment.GetEnvironmentVariable(PasswordEnvVar) is not { Length: > 0 } password) return null; var user = Environment.GetEnvironmentVariable(UserEnvVar) is { Length: > 0 } named ? named : DefaultUser; var (host, port) = PostgresFixture.SplitEndpoint(endpoint, DefaultPort); Server = $"{host}:{port}"; DbConnectionStringBuilder builder = new MySqlDialect().Factory.CreateConnectionStringBuilder() ?? throw new InvalidOperationException("MySqlConnector returned no connection-string builder."); builder["Server"] = host; builder["Port"] = port; builder["Database"] = Schema; builder["User ID"] = user; builder["Password"] = password; builder["Connection Timeout"] = ProbeConnectTimeoutSeconds; return builder.ConnectionString; } } /// Collection definition so the gate is resolved once per session. [CollectionDefinition(Name)] public sealed class MySqlCollection : ICollectionFixture { /// The collection name test classes reference. public const string Name = "SqlMySqlServer"; }