using System.Text.RegularExpressions; using Admin.NET.Core; using Admin.NET.Plugin.AiDOP.DataPlatform; using Admin.NET.Plugin.AiDOP.DataPlatform.Executors; using Admin.NET.Plugin.AiDOP.Entity.DataPlatform; using Microsoft.Extensions.Caching.Memory; using Microsoft.Extensions.Logging.Abstractions; using SqlSugar; using Xunit; namespace Admin.NET.Plugin.AiDOP.Tests.DataPlatform; /// /// 隔离验收:只碰 DBSYNC_TEST_MYSQL / DBSYNC_TEST_SQLSERVER 与 dbsync_test_po。 /// 默认跳过。设置 AIDOP_DBSYNC_ACCEPT=1 后执行。 /// public class DbSyncIsolatedAcceptanceTests { private const long MysqlTenant = 990594001; private const long SqlServerTenant = 990594002; [Fact] public async Task Isolated_sources_cover_keyset_update_full_and_reset() { if (Environment.GetEnvironmentVariable("AIDOP_DBSYNC_ACCEPT") != "1") return; var root = FindRepoRoot(); using var db = OpenAidopdev(root); await RunSourceAsync(db, "DBSYNC_TEST_MYSQL", MysqlTenant, null); await RunNeutralProjectionAsync(db); await PrepareSqlServerSourceAsync(db, root); await RunSourceAsync(db, "DBSYNC_TEST_SQLSERVER", SqlServerTenant, OpenSqlServer(db, root)); } private static async Task RunNeutralProjectionAsync(ISqlSugarClient db) { const string entityCode = "DBSYNC_TEST_S3_PURCHASE"; var source = await db.Queryable() .Where(x => x.SourceCode == "DBSYNC_TEST_MYSQL") .FirstAsync(); var neutral = MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode); var oldIds = await db.Queryable() .Where(x => x.TenantId == MysqlTenant && x.EntityCode == entityCode) .Select(x => x.Id) .ToListAsync(); if (oldIds.Count > 0) await db.Deleteable().Where(x => oldIds.Contains(x.EntityId)).ExecuteCommandAsync(); await db.Deleteable() .Where(x => x.TenantId == MysqlTenant && x.EntityCode == entityCode) .ExecuteCommandAsync(); var now = DateTime.Now; var entity = new MdpEntity { TenantId = MysqlTenant, SourceId = source.Id, EntityCode = entityCode, EntityName = "隔离验收采购订单", EntityType = "TABLE", SourceTableName = "dbsync_test_po", TransformSourceTable = "S3_PURCHASE_ORDER", TargetTableName = "mdp_stg_purchase_order", SyncMode = "FULL", TieBreakerColumn = "id", BizKeyExpr = "PoNo,PoLine", BatchSize = 100, JobId = "DBSYNC_TEST_S3", Status = 1, CreateTime = now, UpdateTime = now }; entity.Id = await db.Insertable(entity).ExecuteReturnBigIdentityAsync(); await db.Insertable(new List { new() { EntityId = entity.Id, SourceField = "bill_no", TargetField = "PoNo", FieldType = "DIRECT", IsRequired = 1, SortOrder = 10, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "line_no", TargetField = "PoLine", FieldType = "DIRECT", IsRequired = 1, SortOrder = 20, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "qty", TargetField = "OrderQty", FieldType = "DIRECT", IsRequired = 1, SortOrder = 30, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "update_time", TargetField = "OrderDate", FieldType = "DIRECT", IsRequired = 1, SortOrder = 40, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "update_time", TargetField = "DueDate", FieldType = "DIRECT", IsRequired = 1, SortOrder = 50, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "", TargetField = "SupplierCode", FieldType = "CONST", ConstValue = "TEST_SUPPLIER", SortOrder = 60, CreateTime = now }, new() { EntityId = entity.Id, SourceField = "", TargetField = "ItemCode", FieldType = "CONST", ConstValue = "TEST_ITEM", SortOrder = 70, CreateTime = now } }).ExecuteCommandAsync(); await db.Ado.ExecuteCommandAsync( """ INSERT INTO mdp_tenant_std_source (tenant_id, std_object, source_system, source_scope, remark) VALUES (@t, 'PURCHASE', @s, 'DBSYNC_ACCEPT', '隔离验收') ON DUPLICATE KEY UPDATE source_system=VALUES(source_system), source_scope=VALUES(source_scope), remark=VALUES(remark) """, new SugarParameter("@t", MysqlTenant), new SugarParameter("@s", neutral)); await db.Ado.ExecuteCommandAsync( "DELETE FROM mdp_stg_purchase_order WHERE tenant_id=@t AND source_system=@s AND source_table='S3_PURCHASE_ORDER'", new SugarParameter("@t", MysqlTenant), new SugarParameter("@s", neutral)); await db.Ado.ExecuteCommandAsync( "DELETE FROM mdp_std_purchase_order WHERE tenant_id=@t AND source_system=@s", new SugarParameter("@t", MysqlTenant), new SugarParameter("@s", neutral)); var executor = BuildExecutor(db); var pull = await PullAllAsync(executor, db, source, entity.Id, full: true); Assert.True(pull.Written >= 29, "neutral staging written=" + pull.Written); var resolver = new MdpTenantStdSourceResolver(db); var gate = new MdpNeutralSourceGate(resolver, db, NullLoggerFactory.Instance); var projection = new InboundNeutralProjectionService(db, gate); await projection.ProjectAsync("S3_PURCHASE_ORDER", MysqlTenant, neutral); var standardRows = await db.Ado.GetIntAsync( "SELECT COUNT(1) FROM mdp_std_purchase_order WHERE tenant_id=@t AND source_system=@s", new SugarParameter("@t", MysqlTenant), new SugarParameter("@s", neutral)); Assert.True(standardRows >= 29, "neutral standard rows=" + standardRows); } private static async Task RunSourceAsync(ISqlSugarClient db, string sourceCode, long tenantId, ISqlSugarClient sourceDb) { await db.Ado.ExecuteCommandAsync( "DELETE FROM mdp_stg_dbsync_test_po WHERE source_system=@s", new SugarParameter("@s", sourceCode)); await db.Ado.ExecuteCommandAsync( "UPDATE mdp_entity SET last_cursor=NULL WHERE tenant_id=@t AND entity_code='DBSYNC_TEST_PO'", new SugarParameter("@t", tenantId)); var entity = await db.Queryable() .Where(x => x.TenantId == tenantId && x.EntityCode == "DBSYNC_TEST_PO" && x.Status == 1) .FirstAsync(); var source = await db.Queryable().Where(x => x.Id == entity.SourceId).FirstAsync(); var cursorBefore = entity.LastCursor; var previewDb = sourceDb ?? db; var preview = await previewDb.Ado.GetIntAsync("SELECT COUNT(1) FROM dbsync_test_po"); var cursorAfterPreview = await db.Queryable().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync(); Assert.True(preview >= 30); Assert.Equal(cursorBefore, cursorAfterPreview); var executor = BuildExecutor(db); var first = await PullAllAsync(executor, db, source, entity.Id, full: false); Assert.True(first.Written >= 29, sourceCode + " written=" + first.Written); Assert.True(first.Errors >= 1, sourceCode + " errors=" + first.Errors); var staged = await db.Ado.GetIntAsync( "SELECT COUNT(1) FROM mdp_stg_dbsync_test_po WHERE source_system=@s AND source_table='dbsync_test_po'", new SugarParameter("@s", sourceCode)); var keys = await db.Ado.GetIntAsync( "SELECT COUNT(DISTINCT source_biz_key) FROM mdp_stg_dbsync_test_po WHERE source_system=@s AND source_table='dbsync_test_po'", new SugarParameter("@s", sourceCode)); Assert.Equal(staged, keys); await TouchRowAsync(previewDb, source.DbType); var second = await PullAllAsync(executor, db, source, entity.Id, full: false); Assert.True(second.Written >= 1, sourceCode + " update written=" + second.Written); var full = await PullAllAsync(executor, db, source, entity.Id, full: true); Assert.True(full.Read >= 30, sourceCode + " full read=" + full.Read); var current = await db.Queryable().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync(); await db.Ado.ExecuteCommandAsync( "UPDATE mdp_entity SET last_cursor=NULL, update_time=NOW() WHERE id=@id AND ((last_cursor IS NULL AND @old IS NULL) OR last_cursor=@old)", new SugarParameter("@id", entity.Id), new SugarParameter("@old", current)); await db.Ado.ExecuteCommandAsync( @"INSERT INTO mdp_entity_cursor_audit (tenant_id, entity_id, entity_code, old_cursor, new_cursor, action_type, reason, create_time) VALUES (@t, @id, @code, @old, NULL, 'RESET', '隔离验收重置水位', NOW())", new SugarParameter("@t", tenantId), new SugarParameter("@id", entity.Id), new SugarParameter("@code", entity.EntityCode), new SugarParameter("@old", current)); var third = await PullAllAsync(executor, db, source, entity.Id, full: false); Assert.True(third.Read >= 30, sourceCode + " reset read=" + third.Read); var logs = await db.Ado.GetIntAsync( "SELECT COUNT(1) FROM mdp_sync_log WHERE source_code=@s AND entity_id=@id", new SugarParameter("@s", sourceCode), new SugarParameter("@id", entity.Id)); Assert.True(logs >= 3, sourceCode + " logs=" + logs); } private static async Task TouchRowAsync(ISqlSugarClient sourceDb, string dbType) { var sql = dbType != null && dbType.Replace("_", "").Contains("SQLSERVER", StringComparison.OrdinalIgnoreCase) ? "UPDATE dbsync_test_po SET qty=qty+1, update_time=DATEADD(second,1,(SELECT MAX(update_time) FROM dbsync_test_po)) WHERE id=2" : "UPDATE dbsync_test_po SET qty=qty+1, update_time=(SELECT next_time FROM (SELECT DATE_ADD(MAX(update_time), INTERVAL 1 SECOND) next_time FROM dbsync_test_po) x) WHERE id=2"; await sourceDb.Ado.ExecuteCommandAsync(sql); } private static async Task<(int Read, int Written, int Errors)> PullAllAsync( MdpDbPullExecutor executor, ISqlSugarClient db, MdpSource source, long entityId, bool full) { var read = 0; var written = 0; var errors = 0; var ctx = new MdpPullContext { TenantId = source.SourceCode == "DBSYNC_TEST_SQLSERVER" ? SqlServerTenant : MysqlTenant, FullRefresh = full, SyncWindowType = full ? "FULL" : "INCR", RequireConfiguredDbSync = true, TriggerType = "MANUAL", BatchId = "DBSYNC_ACCEPT_" + DateTime.Now.ToString("yyyyMMddHHmmssfff") }; for (var page = 0; page < 20; page++) { var entity = await db.Queryable().Where(x => x.Id == entityId).FirstAsync(); if (!full) { ctx.UseKeysetCursor = false; ctx.CursorValue = null; ctx.TieBreakerValue = null; ctx.Offset = 0; } MdpSourcePullDispatcher.ApplyDbSyncDefaults(entity, source, ctx); var result = await executor.PullAsync(source, entity, ctx); read += result.RowsPulled; written += result.RowsWritten; errors += result.RowsError; if (result.RowsPulled <= 0 || result.RowsPulled < entity.BatchSize) break; if (full) ctx.Offset += result.RowsPulled; } return (read, written, errors); } private static MdpDbPullExecutor BuildExecutor(ISqlSugarClient db) { var passwords = new MdpSourcePasswordResolver(NullLogger.Instance); var scopes = new MdpSourceScopeFactory(db, passwords, NullLogger.Instance); var maps = new MdpPullFieldMappingService(db, new MemoryCache(new MemoryCacheOptions())); var writer = new MdpStagingWriter(db, NullLogger.Instance); return new MdpDbPullExecutor(scopes, db, writer, maps); } private static async Task PrepareSqlServerSourceAsync(ISqlSugarClient mysql, string root) { using var db = OpenSqlServer(mysql, root); await db.Ado.ExecuteCommandAsync(""" IF OBJECT_ID('dbo.dbsync_test_po','U') IS NULL CREATE TABLE dbo.dbsync_test_po ( id bigint NOT NULL PRIMARY KEY, bill_no nvarchar(40) NULL, line_no int NOT NULL, qty decimal(18,4) NOT NULL, update_time datetime2(3) NOT NULL, row_status nvarchar(20) NOT NULL) """); await db.Ado.ExecuteCommandAsync("DELETE FROM dbo.dbsync_test_po"); await db.Ado.ExecuteCommandAsync(""" INSERT INTO dbo.dbsync_test_po (id, bill_no, line_no, qty, update_time, row_status) VALUES (1,'PO-001',1,1.5,'2026-10-01T08:00:00','OPEN'), (2,'PO-002',2,2.5,'2026-10-01T08:00:00','OPEN'), (3,'PO-003',3,3.5,'2026-10-01T08:00:00','OPEN'), (4,'PO-004',4,4.5,'2026-10-01T08:00:00','OPEN'), (5,'PO-005',5,5.5,'2026-10-01T08:00:00','DELETED'), (6,'PO-006',6,6.5,'2026-10-01T08:00:00','OPEN'), (7,'PO-007',7,7.5,'2026-10-01T08:00:00','OPEN'), (8,'PO-008',8,8.5,'2026-10-01T08:00:00','OPEN'), (9,'PO-009',9,9.5,'2026-10-01T08:00:00','OPEN'), (10,'PO-010',10,10.5,'2026-10-01T08:00:00','OPEN'), (11,'PO-011',11,11.5,'2026-10-02T08:00:00','OPEN'), (12,'PO-012',12,12.5,'2026-10-02T08:00:00','OPEN'), (13,'PO-013',13,13.5,'2026-10-02T08:00:00','OPEN'), (14,'PO-014',14,14.5,'2026-10-02T08:00:00','OPEN'), (15,'PO-015',15,15.5,'2026-10-02T08:00:00','OPEN'), (16,'PO-016',16,16.5,'2026-10-02T08:00:00','OPEN'), (17,'PO-017',17,17.5,'2026-10-02T08:00:00','OPEN'), (18,'PO-018',18,18.5,'2026-10-02T08:00:00','OPEN'), (19,'PO-019',19,19.5,'2026-10-02T08:00:00','OPEN'), (20,'PO-020',20,20.5,'2026-10-02T08:00:00','OPEN'), (21,'PO-021',21,21.5,'2026-10-03T08:00:00','OPEN'), (22,'PO-022',22,22.5,'2026-10-03T08:00:00','OPEN'), (23,'PO-023',23,23.5,'2026-10-03T08:00:00','OPEN'), (24,'PO-024',24,24.5,'2026-10-03T08:00:00','OPEN'), (25,'PO-025',25,25.5,'2026-10-03T08:00:00','OPEN'), (26,'PO-026',26,26.5,'2026-10-03T08:00:00','OPEN'), (27,'PO-027',27,27.5,'2026-10-03T08:00:00','OPEN'), (28,'PO-028',28,28.5,'2026-10-03T08:00:00','OPEN'), (29,'PO-029',29,29.5,'2026-10-03T08:00:00','OPEN'), (30,NULL,30,30.5,'2026-10-03T08:00:00','OPEN') """); } private static SqlSugarClient OpenSqlServer(ISqlSugarClient mysql, string root) { var cipher = mysql.Ado.GetString("SELECT db_password_enc FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'"); var key = Regex.Match(File.ReadAllText(Path.Combine(root, "server", "Admin.NET.Application", "Configuration", "App.json")), "\"PrivateKey\"\\s*:\\s*\"([^\"]+)\"").Groups[1].Value; string password; try { password = GMUtil.SM2Decrypt(key, cipher); } catch (Exception) { password = cipher; } Environment.SetEnvironmentVariable(MdpSourcePasswordResolver.EnvironmentVariableName("DBSYNC_TEST_SQLSERVER"), password); var host = mysql.Ado.GetString("SELECT db_host FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'"); var port = mysql.Ado.GetInt("SELECT db_port FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'"); var name = mysql.Ado.GetString("SELECT db_name FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'"); var user = mysql.Ado.GetString("SELECT db_user FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'"); return new SqlSugarClient(new ConnectionConfig { ConfigId = "dbsync-test-sqlserver", DbType = DbType.SqlServer, ConnectionString = $"Server={host},{port};Database={name};User Id={user};Password={password};TrustServerCertificate=true;Encrypt=false", IsAutoCloseConnection = true }); } private static SqlSugarScope OpenAidopdev(string root) { var cs = File.ReadLines(Path.Combine(root, "server", "Admin.NET.Application", "Configuration", "Database.json")) .Select(x => x.Trim()) .Where(x => !x.StartsWith("//") && x.Contains("\"ConnectionString\"") && x.Contains("Database=aidopdev")) .Select(x => Regex.Match(x, "\"ConnectionString\"\\s*:\\s*\"([^\"]+)\"").Groups[1].Value) .First(x => !string.IsNullOrEmpty(x)); return new SqlSugarScope(new ConnectionConfig { ConfigId = "dbsync-accept", DbType = DbType.MySql, ConnectionString = cs, IsAutoCloseConnection = true }); } private static string FindRepoRoot() { var dir = new DirectoryInfo(AppContext.BaseDirectory); while (dir != null) { if (File.Exists(Path.Combine(dir.FullName, "server", "Admin.NET.Application", "Configuration", "Database.json"))) return dir.FullName; dir = dir.Parent; } throw new InvalidOperationException("未找到 Database.json"); } }