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");
}
}