Files
StoreWeb/Tests/CloudProduct.Checks/IntegrationChecks.cs
T

330 lines
24 KiB
C#

using System.Reflection;
using System.Text;
using System.Text.Json;
using MAX.Framework.Commons;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging.Abstractions;
using OpenSearch.Client;
using OpenSearch.Net;
using Store.Modules.CloudProduct.Dtos;
using Store.Modules.CloudProduct.Entities;
using Store.Modules.CloudProduct.Repositorys;
using Store.Modules.CloudProduct.Services;
internal static class IntegrationChecks
{
public static async Task RunAsync(bool applyMigration = false, bool migrateViolationRows = false)
{
var root = Directory.GetCurrentDirectory();
var config = new ConfigurationBuilder().SetBasePath(root)
.AddJsonFile("Store.API/appsettings.json")
.AddJsonFile("Store.API/appsettings.Development.json", optional: true)
.AddJsonFile("Modules/CloudProduct/appsettings.CloudProduct.json").Build();
using var db = new FreeSql.FreeSqlBuilder()
.UseConnectionString(FreeSql.DataType.MySql, config["CloudProductConnection"])
.UseAdoConnectionPool(true)
.Build();
if (applyMigration)
{
var sql = await File.ReadAllTextAsync(Path.Combine(root, "Modules/CloudProduct/Sql/20261030_violation_reason_is_violation.sql"));
var present = Convert.ToInt32(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='violation_reason' AND column_name='is_violation'"));
if (present > 0) sql = sql[(sql.IndexOf(';') + 1)..];
await db.Ado.ExecuteNonQueryAsync(sql);
Console.WriteLine("MIGRATION PASS: is_violation initialized");
}
var run = Guid.NewGuid().ToString("N");
var index = "violation-check-" + run;
var originalIndex = config["Search:IndexName"]!;
config["Search:IndexName"] = index;
config["Indexing:Mode"] = "Immediate";
var client = new OpenSearchClient(new ConnectionSettings(new Uri(config["Search:Url"]!)).DefaultIndex(index));
var type = typeof(IOpenSearchService).Assembly.GetType("Store.Modules.CloudProduct.Services.OpenSearchService")!;
var logger = Activator.CreateInstance(typeof(NullLogger<>).MakeGenericType(type));
var gate = new GoodsIndexLock(config);
var search = (IOpenSearchService)Activator.CreateInstance(type, config, logger, db, gate)!;
GoodsViolationService Service(IOpenSearchService? target = null) => new(
new GoodsViolationRepository(db), gate, target ?? search, NullLogger<GoodsViolationService>.Instance);
var violations = Service();
if (migrateViolationRows)
{
await using var lease = await gate.AcquireAsync();
var invalid = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM goods_violation WHERE JSON_TYPE(reason_ids) <> 'ARRAY' OR JSON_LENGTH(reason_ids)=0"));
if (invalid != 0) throw new Exception("Migration requires nonempty reason arrays; inspect legacy records first");
var expected = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(DISTINCT v.goods_id, r.reason_id) FROM goods_violation v JOIN JSON_TABLE(v.reason_ids, '$[*]' COLUMNS (reason_id BIGINT PATH '$' ERROR ON ERROR)) r ON TRUE"));
var migration = await File.ReadAllTextAsync(Path.Combine(root, "Modules/CloudProduct/Sql/20261030_goods_violation_reason_id.sql"));
var switchAt = migration.IndexOf("RENAME TABLE", StringComparison.Ordinal);
await db.Ado.ExecuteNonQueryAsync(migration[..switchAt]);
var copied = Convert.ToInt64(await db.Ado.ExecuteScalarAsync("SELECT COUNT(*) FROM goods_violation_new"));
if (expected != copied) throw new Exception("Migration row count mismatch; original table not switched");
await db.Ado.ExecuteNonQueryAsync(migration[switchAt..]);
Console.WriteLine($"MIGRATION PASS: {copied} reason rows migrated; original table retained as goods_violation_json_backup");
}
var goodsType = type.Assembly.GetType("Store.Modules.CloudProduct.Services.GoodsService")!;
var goods = (IGoodsService)Activator.CreateInstance(goodsType,
new GoodsRepository(db), search, new NoRawData(), config, violations)!;
var first = 8_000_000_000_000_000L + Random.Shared.NextInt64(1_000_000_000);
long[] ids = [first, first + 1, first + 2, first + 3];
// Exercise full-scan jobs against only this run's records in the shared test database.
db.GlobalFilter.Apply<GoodsEntity>("integration-goods", x => ids.Contains(x.GoodsId));
db.GlobalFilter.Apply<GoodsViolationEntity>("integration-violations", x => ids.Contains(x.GoodsId));
var reasons = new List<uint>();
string[] names = ["发布不合规信息", "出售假冒商品", "价格违规", "发布违禁信息"];
var ignoredName = "其他原因-" + run;
string[] allowedNames = [
"发布不合规信息",
"出售假冒商品",
"价格违规",
"发布违禁信息",
"材质夸大宣传",
"不当使用他人权利",
"发布虚假宣传信息",
"商品价格违规",
"出售盗版商品"
];
var ownsIds = false;
var checks = 0;
void Check(bool condition, string name)
{
if (!condition) throw new Exception("FAIL: " + name);
Console.WriteLine("PASS: " + name);
checks++;
}
async Task<bool> Exists(long id)
{
var response = await client.DocumentExistsAsync<GoodsSearchDocument>(id, d => d.Index(index));
if (!response.IsValid && response.ApiCall?.HttpStatusCode != 404)
throw new Exception("Document existence request failed");
return response.Exists;
}
async Task<uint[]> Reasons(long id) => (await db.Select<GoodsViolationEntity>()
.Where(x => x.GoodsId == id).OrderBy(x => x.ReasonId).ToListAsync(x => x.ReasonId)).ToArray();
async Task<GoodsViolationEntity> Record(long id) => await db.Select<GoodsViolationEntity>()
.Where(x => x.GoodsId == id).FirstAsync();
try
{
Check(await db.Select<GoodsViolationEntity>().CountAsync() >= 0, "read existing violation table");
Check(await db.Select<ViolationReasonEntity>().CountAsync() >= 0, "read existing reason dictionary");
Check(!await db.Select<GoodsEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
&& !await db.Select<GoodsViolationEntity>().AnyAsync(x => ids.Contains(x.GoodsId)), "test IDs are unused");
ownsIds = true;
var enabledReasons = await db.Select<ViolationReasonEntity>().Where(x => x.IsViolation).ToListAsync();
Check(enabledReasons.Select(x => x.Name).ToHashSet(StringComparer.Ordinal).SetEquals(allowedNames), "exactly the specified 9 dictionary reasons enabled");
var existingReasonId = (await db.Select<ViolationReasonEntity>().Where(x => x.Name == names[0]).FirstAsync()).Id;
Check(await search.CreateIndexAsync(), "create isolated index on configured OpenSearch server");
await search.BulkIndexGoodsAsync(ids.Select(id => new GoodsSearchDocument { Id = id, Title = "integration " + run }));
Check(await Exists(first), "initial product is indexed");
var result = await violations.ReportAsync([
new() { GoodsId = first, Reasons = " " + names[0] + " " },
new() { GoodsId = first, Reasons = names[0] },
new() { GoodsId = first, Reasons = names[1] },
new() { GoodsId = first, Reasons = ignoredName }
]);
Check(result.Count == 1 && result[0].GoodsId == first && result[0].IsOffShelf, "batch report and deduplication");
Check(!await Exists(first), "report deletes real OpenSearch document");
reasons.Add((await db.Select<ViolationReasonEntity>().Where(x => x.Name == names[0]).FirstAsync()).Id);
reasons.Add((await db.Select<ViolationReasonEntity>().Where(x => x.Name == names[1]).FirstAsync()).Id);
Check(reasons[0] == existingReasonId && reasons[1] > 0, "reuse enabled dictionary reason IDs");
var record = await Record(first);
Check(record.IsIndexDeleted && (await Reasons(first)).Order().SequenceEqual(reasons.Take(2).Order()), "separate reasons and is_index_deleted persisted");
await violations.ReportAsync([new() { GoodsId = first, Reasons = names[1] }, new() { GoodsId = first, Reasons = names[2] }]);
reasons.Add((await db.Select<ViolationReasonEntity>().Where(x => x.Name == names[2]).FirstAsync()).Id);
Check(await db.Select<ViolationReasonEntity>().Where(x => names.Contains(x.Name)).CountAsync() == 4, "repeated reasons reuse dictionary entries");
record = await Record(first);
Check((await Reasons(first)).Order().SequenceEqual(reasons.Order()), "repeat report merges previous reasons");
Check(await db.Select<GoodsViolationEntity>().Where(x => x.GoodsId == first).CountAsync() == 3, "one row per distinct product/reason pair");
try
{
await db.Insert(await Record(first)).ExecuteAffrowsAsync();
throw new Exception("Duplicate product/reason pair accepted");
}
catch (Exception ex) when (ex.GetBaseException() is MySql.Data.MySqlClient.MySqlException { Number: 1062 })
{
Check(true, "database composite primary key rejects duplicate product/reason pairs");
}
await db.Update<GoodsViolationEntity>().Set(x => x.IsIndexDeleted, false)
.Where(x => x.GoodsId == first && x.ReasonId == reasons[0]).ExecuteAffrowsAsync();
var violationRepository = new GoodsViolationRepository(db);
Check(!(await violationRepository.GetOffShelfGoodsIdsAsync([first])).Contains(first),
"one pending reason prevents reporting the product as fully off shelf");
var pendingBatch = await violationRepository.GetBatchAsync(0, 1, true);
Check(pendingBatch.SequenceEqual(new[] { first }), "pending deletion pages distinct product IDs");
await violationRepository.MarkDeletedAsync([first]);
var ignored = await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = ignoredName }]);
Check(ignored.Count == 1 && ignored[0].GoodsId == first + 1 && !ignored[0].IsOffShelf && await Record(first + 1) == null && await Exists(first + 1),
"unlisted reason records dictionary only and leaves index untouched");
var disabledReport = await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = "品牌资质问题" }]);
Check(disabledReport.Count == 1 && disabledReport[0].GoodsId == first + 1 && !disabledReport[0].IsOffShelf
&& await Record(first + 1) == null && await Exists(first + 1), "previously enabled reason outside the nine does not ban goods");
Check(!((await db.Select<ViolationReasonEntity>().Where(x => x.Name == ignoredName).FirstAsync()).IsViolation),
"new dictionary reason defaults to not a violation");
var before = await Reasons(first);
ignored = await violations.ReportAsync([new() { GoodsId = first, Reasons = ignoredName }]);
Check(ignored.Single(x => x.GoodsId == first).IsOffShelf && (await Reasons(first)).SequenceEqual(before) && !await Exists(first),
"ignored reason does not clear or alter existing violation");
// A failing search transport proves dictionary-only reports never access OpenSearch.
var status = await violations.GetStatusAsync([first, first + 1, first, first + 100]);
Check(status.Count == 3 && status[0].IsViolation && !status[1].IsViolation && !status[2].IsViolation, "batch status including missing goods");
Check((await Reasons(first)).Length == 3, "scalar reasons stored in separate rows");
Check((await violations.GetGoodsIdsByReasonAsync(reasons[2], first - 1)).Contains(first), "scalar reason query");
try
{
await violations.ReportAsync([new() { GoodsId = first + 3, Reasons = " " }]);
throw new Exception("Invalid reason accepted");
}
catch (AppException) { Check(await Record(first + 3) == null, "blank reason creates no row"); }
try
{
await search.AddDocumentAsync(new() { Id = first, Title = "blocked" });
throw new Exception("Manual write accepted a violation");
}
catch (AppException) { Check(!await Exists(first), "manual indexing rejects violation"); }
var indexed = await search.BulkIndexGoodsAsync([
new() { Id = first, Title = "blocked" }, new() { Id = first + 1, Title = "allowed" }
]);
Check(indexed.SequenceEqual(new[] { first + 1 }) && !await Exists(first) && await Exists(first + 1), "bulk indexing filters violations");
var imported = await goods.ImportGoodsAsync([
new() { Goods = new() { GoodsId = first, Platform = 1, MinPrice = 0, MaxPrice = 0, CatId1 = 0, CatId2 = 0, CatId3 = 0, CatId4 = 0, ThumbUrl = "", Description = "", Title = "integration " + run } },
new() { Goods = new() { GoodsId = first + 1, Platform = 1, MinPrice = 0, MaxPrice = 0, CatId1 = 0, CatId2 = 0, CatId3 = 0, CatId4 = 0, ThumbUrl = "", Description = "", Title = "integration " + run } }
]);
Check(imported.SuccessCount == 2 && imported.FailedCount == 0 && !await Exists(first), "immediate import retains data but excludes violation: " + string.Join("; ", imported.Errors));
// Inject a stale document to prove rebuild actively removes old violations.
var stale = await client.IndexAsync(new GoodsSearchDocument { Id = first, Title = "stale" }, i => i.Index(index).Id(first));
Check(stale.IsValid, "seed stale violation document");
var count = await goods.RebuildSearchIndexAsync([first, first + 1]);
Check(count == 1 && !await Exists(first) && await Exists(first + 1), "targeted rebuild removes stale violation and indexes eligible goods");
await db.Update<GoodsEntity>().Set(x => x.IsSegmented, (byte)0)
.Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
await search.DeleteDocumentsAsync([first + 1]);
Check(await goods.IndexPendingGoodsAsync(1) == 1 && await Exists(first + 1) && !await Exists(first),
"pending job filters violations before LIMIT 1");
Check(await goods.RebuildSearchIndexAsync() == 1 && !await Exists(first), "full rebuild excludes violations");
// Real DB persistence with a deterministic failed search transport.
var failed = (IOpenSearchService)Activator.CreateInstance(type, config, logger, db, gate)!;
using var transport = new InMemoryConnection(Encoding.UTF8.GetBytes("{}"), 503);
var failedClient = new OpenSearchClient(new ConnectionSettings(new SingleNodeConnectionPool(new Uri("http://localhost:9200")), transport));
type.GetField("_client", BindingFlags.Instance | BindingFlags.NonPublic)!.SetValue(failed, failedClient);
var oldTime = new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc);
await db.Update<GoodsViolationEntity>().Set(x => x.UpdatedTime, oldTime)
.Where(x => x.GoodsId == first).ExecuteAffrowsAsync();
var unchangedBefore = await Record(first);
var unchangedReasons = await Reasons(first);
var unchanged = await Service(failed).ReportAsync([
new() { GoodsId = first, Reasons = names[2] },
new() { GoodsId = first, Reasons = " " + names[0] + " " },
new() { GoodsId = first, Reasons = names[2] },
new() { GoodsId = first, Reasons = ignoredName }
]);
var unchangedAfter = await Record(first);
Check(unchanged.Count == 1 && unchanged[0].GoodsId == first && unchanged[0].IsOffShelf,
"duplicate/subset reasons skip unavailable OpenSearch and remain reported as off shelf");
Check(unchangedBefore.UpdatedTime == unchangedAfter.UpdatedTime
&& unchangedReasons.SequenceEqual(await Reasons(first)) && unchangedAfter.IsIndexDeleted,
"unchanged report preserves timestamp, reasons and deletion status");
var added = await Service(failed).ReportAsync([new() { GoodsId = first, Reasons = names[3] }]);
Check(!added.Single(x => x.GoodsId == first).IsOffShelf
&& !(await Record(first)).IsIndexDeleted
&& (await Reasons(first)).Length == 4,
"new valid reason updates record and attempts deletion again");
var retry = await violations.ReportAsync([new() { GoodsId = first, Reasons = names[3] }]);
Check(retry.Single(x => x.GoodsId == first).IsOffShelf
&& (await Record(first)).IsIndexDeleted, "unchanged reasons still retry a previously failed deletion");
ignored = await Service(failed).ReportAsync([new() { GoodsId = first + 1, Reasons = ignoredName }]);
Check(!ignored.Single(x => x.GoodsId == first + 1).IsOffShelf, "ignored report does not require available OpenSearch");
result = await Service(failed).ReportAsync([
new() { GoodsId = first, Reasons = names[0] },
new() { GoodsId = first + 2, Reasons = names[0] }
]);
Check(result.Count == 2 && result[0].GoodsId == first && result[0].IsOffShelf
&& result[1].GoodsId == first + 2 && !result[1].IsOffShelf,
"mixed batch separates unchanged successes from pending deletion");
Check(!result.Single(x => x.GoodsId == first + 2).IsOffShelf && !(await Record(first + 2)).IsIndexDeleted,
"search failure persists pending deletion in real DB");
await violations.DeleteViolationIndexesAsync();
Check((await Record(first + 2)).IsIndexDeleted && !await Exists(first + 2), "retry deletes document and marks completed");
await Task.WhenAll(
search.BulkIndexGoodsAsync([new() { Id = first + 3, Title = "concurrent write" }]),
Service().ReportAsync([new() { GoodsId = first + 3, Reasons = names[0] }, new() { GoodsId = first + 3, Reasons = names[3] }]),
Service().ReportAsync([new() { GoodsId = first + 3, Reasons = names[1] }, new() { GoodsId = first + 3, Reasons = names[3] }]));
Check((await Reasons(first + 3)).Length == 3,
"concurrent reports preserve both reasons");
Check(await db.Select<ViolationReasonEntity>().Where(x => x.Name == names[3]).CountAsync() == 1, "concurrent report reuses dictionary entry");
Check(!await Exists(first + 3), "concurrent indexing cannot restore reported violation");
var removed = await violations.RemoveAsync([first, first, first + 2]);
Check(removed.RemovedCount == 2 && removed.RestoredGoodsIds.SequenceEqual(new[] { first })
&& removed.SkippedGoodsIds.SequenceEqual(new[] { first + 2 }), "remove duplicates and skip goods absent from database");
Check(await Record(first) == null && await Exists(first), "remove violation restores real index");
Check(!(await violations.GetStatusAsync([first]))[0].IsViolation, "restored product is no longer a violation");
Check(await db.Select<ViolationReasonEntity>().Where(x => names.Contains(x.Name)).CountAsync() == 4,
"removal preserves shared reason dictionary");
removed = await violations.RemoveAsync([first]);
Check(removed.RemovedCount == 0 && removed.RestoredGoodsIds.Contains(first), "repeat removal is idempotent");
await violations.ReportAsync([new() { GoodsId = first, Reasons = names[0] }]);
removed = await Service(failed).RemoveAsync([first]);
Check(removed.PendingRestoreGoodsIds.Contains(first) && await Record(first) == null
&& await db.Select<GoodsEntity>().AnyAsync(x => x.GoodsId == first && x.IsSegmented == 0),
"failed restore commits removal and leaves durable pending index flag");
Check(await goods.IndexPendingGoodsAsync(1000) == 1 && await Exists(first), "pending job retries failed restoration");
await db.Update<GoodsEntity>().Set(x => x.Status, (byte)1).Where(x => x.GoodsId == first + 1).ExecuteAffrowsAsync();
await violations.ReportAsync([new() { GoodsId = first + 1, Reasons = names[0] }]);
removed = await violations.RemoveAsync([first + 1]);
Check(removed.SkippedGoodsIds.Contains(first + 1) && await Record(first + 1) == null && !await Exists(first + 1),
"soft-deleted product is unblocked without resurrecting index");
await Task.WhenAll(violations.RemoveAsync([first]),
Service().ReportAsync([new() { GoodsId = first, Reasons = names[1] }]));
Check(((await Record(first)) != null) != await Exists(first), "concurrent removal and reporting leave consistent index state");
var allReasonsResult = await violations.ReportAsync(allowedNames.Select(name =>
new ReportGoodsViolationItem { GoodsId = first + 3, Reasons = name }).ToArray());
Check(allReasonsResult.Count == 1 && allReasonsResult[0].GoodsId == first + 3 && allReasonsResult[0].IsOffShelf
&& (await Reasons(first + 3)).ToHashSet()
.SetEquals(enabledReasons.Select(x => x.Id)), "all 9 reasons enter the product violation record");
Console.WriteLine($"INTEGRATION PASS: {checks} checks");
}
finally
{
// Only remove this run's isolated index and uniquely identified DB records.
var deleted = await client.Indices.DeleteAsync(index);
if (!deleted.IsValid && deleted.ApiCall?.HttpStatusCode != 404)
throw new Exception("Cleanup failed for test index: " + index);
// A separately running index job may have picked up test goods in the configured index.
var originalClient = new OpenSearchClient(new ConnectionSettings(new Uri(config["Search:Url"]!)).DefaultIndex(originalIndex));
var existing = await originalClient.Indices.ExistsAsync(originalIndex);
if (existing.Exists && ownsIds)
{
var cleanup = await originalClient.BulkAsync(b => b.Index(originalIndex).DeleteMany<GoodsSearchDocument>(ids.Select(id => new GoodsSearchDocument { Id = id })));
if (cleanup.ApiCall?.HttpStatusCode is not (>= 200 and < 300)
|| cleanup.Items.Count != ids.Length
|| cleanup.Items.Any(x => x.Status != 404 && (x.Status < 200 || x.Status >= 300)))
throw new Exception("Cleanup failed for test documents in configured index");
await originalClient.Indices.RefreshAsync(originalIndex);
}
if (ownsIds)
{
await db.Delete<GoodsEntity>().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
await db.Delete<GoodsViolationEntity>().Where(x => ids.Contains(x.GoodsId)).ExecuteAffrowsAsync();
}
await db.Delete<ViolationReasonEntity>().Where(x => x.Name == ignoredName).ExecuteAffrowsAsync();
Check(!await db.Select<GoodsViolationEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
&& !await db.Select<GoodsEntity>().AnyAsync(x => ids.Contains(x.GoodsId))
&& !await db.Select<ViolationReasonEntity>().AnyAsync(x => x.Name == ignoredName), "test database records cleaned");
Console.WriteLine("CLEANUP PASS: test index and records removed");
}
}
private sealed class NoRawData : IRawDataStorage
{
public bool Exists(long id) => false;
public Task<string?> ReadAsync(long id) => Task.FromResult<string?>(null);
public Task<string> SaveAsync(long id, string raw) => throw new InvalidOperationException("Integration test does not write raw files");
}
}