From 06de9b1c939de2ad5ba75d3f02ca7200298940b8 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Mon, 12 Oct 2026 21:24:56 +0800 Subject: [PATCH] test(selfop): verify publication transactions against opt-in TEST MySQL --- ...ventoryPublicationMySqlAcceptanceTest.java | 237 ++++++++++++++++++ 1 file changed, 237 insertions(+) create mode 100644 lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/inventory/InventoryPublicationMySqlAcceptanceTest.java diff --git a/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/inventory/InventoryPublicationMySqlAcceptanceTest.java b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/inventory/InventoryPublicationMySqlAcceptanceTest.java new file mode 100644 index 000000000..6c16d4a37 --- /dev/null +++ b/lastleg-service/src/test/java/com/baiyue/lastleg/selfoperated/analytics/inventory/InventoryPublicationMySqlAcceptanceTest.java @@ -0,0 +1,237 @@ +package com.baiyue.lastleg.selfoperated.analytics.inventory; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.toolkit.IdWorker; +import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean; +import com.baiyue.lastleg.selfoperated.analytics.mapper.*; +import com.baiyue.lastleg.selfoperated.analytics.model.*; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.ibatis.executor.Executor; +import org.apache.ibatis.logging.nologging.NoLoggingImpl; +import org.apache.ibatis.mapping.MappedStatement; +import org.apache.ibatis.plugin.*; +import org.junit.jupiter.api.*; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; +import org.mybatis.spring.SqlSessionTemplate; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.DriverManagerDataSource; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; + +import java.math.BigDecimal; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.*; +import java.util.*; +import static com.baiyue.lastleg.selfoperated.analytics.model.SelfopSupport.*; +import static org.junit.jupiter.api.Assertions.*; + +/** Opt-in TEST-only real MySQL transactions. No collector, HTTP client, scheduler or mocked SQL. */ +@EnabledIfEnvironmentVariable(named="SELFOP_TEST_MYSQL_PUBLICATION",matches="true") +@Timeout(90) +class InventoryPublicationMySqlAcceptanceTest { + private static final String ACCOUNT="SELFOP-TEST-TXN-20261012"; + private static final ZoneId ZONE=ZoneId.of("Asia/Shanghai"); + private static final String FAULT="SYNTH_PUBLICATION_FAULT_AFTER_RUN_WRITE"; + private final String token=UUID.randomUUID().toString().replace("-",""); + private final String warehouse="SELFOP-TXN-"+token.substring(0,12).toUpperCase(Locale.ROOT); + private final String owner="SYNTH-TXN-PUBLISH-"+token; + private final Map evidence=new LinkedHashMap<>(); + private JdbcTemplate jdbc; + private TransactionTemplate transactions; + private SelfopSyncScopeMapper scopes; + private SelfopSyncTaskMapper tasks; + private SelfopSyncRunMapper runs; + private SelfopInventorySnapshotMapper stocks; + private SelfopSkuProfileMapper profiles; + private SelfopSyncScope scope; + private InventorySnapshotService service; + private LocalDate day; + private WriteFault fault; + private SelfopAnalyticsProperties properties; + + @BeforeEach void initialize(TestInfo info) throws Exception { + assertEquals("test",requiredEnv("CHONGCHONG_ENV").toLowerCase(Locale.ROOT),"Only TEST is authorized"); + String host=requiredEnv("CHONGCHONG_DB_HOST"),port=requiredEnv("CHONGCHONG_DB_PORT"); + assertTrue(host.matches("[A-Za-z0-9.-]+")&&port.matches("[0-9]{1,5}"),"Invalid TEST connection fields"); + var source=new DriverManagerDataSource("jdbc:mysql://"+host+":"+port+"/baiyue_lastleg" + +"?useSSL=false&allowPublicKeyRetrieval=true&connectionTimeZone=UTC&forceConnectionTimeZoneToSession=true" + +"&characterEncoding=UTF-8&connectTimeout=10000&socketTimeout=15000", + requiredEnv("CHONGCHONG_DB_USERNAME"),requiredEnv("CHONGCHONG_DB_PASSWORD")); + source.setDriverClassName("com.mysql.cj.jdbc.Driver"); + try(var connection=source.getConnection()) { + assertEquals("MySQL",connection.getMetaData().getDatabaseProductName()); + assertEquals(8,connection.getMetaData().getDatabaseMajorVersion()); + assertEquals("baiyue_lastleg",connection.getCatalog()); + } catch(java.sql.SQLException unavailable) { + throw new IllegalStateException("TEST MySQL connection unavailable; root must inspect connection settings"); + } + jdbc=new JdbcTemplate(source);jdbc.setQueryTimeout(15); + var manager=new DataSourceTransactionManager(source);transactions=new TransactionTemplate(manager); + var configuration=new MybatisConfiguration();configuration.setMapUnderscoreToCamelCase(true); + configuration.setLogImpl(NoLoggingImpl.class);fault=new WriteFault();configuration.addInterceptor(fault); + for(var mapper:List.of(SelfopCoordinationMapper.class,SelfopSyncScopeMapper.class,SelfopSyncTaskMapper.class, + SelfopSyncRunMapper.class,SelfopSyncStageMapper.class,SelfopInventorySnapshotMapper.class,SelfopSkuProfileMapper.class)) + configuration.addMapper(mapper); + var factory=new MybatisSqlSessionFactoryBean();factory.setDataSource(source);factory.setConfiguration(configuration); + var session=new SqlSessionTemplate(Objects.requireNonNull(factory.getObject())); + scopes=session.getMapper(SelfopSyncScopeMapper.class);tasks=session.getMapper(SelfopSyncTaskMapper.class); + runs=session.getMapper(SelfopSyncRunMapper.class);stocks=session.getMapper(SelfopInventorySnapshotMapper.class); + profiles=session.getMapper(SelfopSkuProfileMapper.class);properties=new SelfopAnalyticsProperties(); + assertEquals("OFF",properties.getCollectionMode());assertFalse(properties.isExecutionEnabled()); + var target=new InventorySnapshotService(session.getMapper(SelfopCoordinationMapper.class),scopes,runs,tasks, + session.getMapper(SelfopSyncStageMapper.class),stocks,profiles,properties); + var proxy=new ProxyFactory(target);proxy.setProxyTargetClass(true); + proxy.addAdvice(new TransactionInterceptor(manager,new AnnotationTransactionAttributeSource())); + service=(InventorySnapshotService)proxy.getProxy();day=LocalDate.now(ZONE); + evidence.put("case",info.getDisplayName());evidence.put("token",token);evidence.put("environment","TEST"); + evidence.put("sourceAccount",ACCOUNT);evidence.put("warehouseCode",warehouse); + evidence.put("realMySqlTransactions",true);evidence.put("collectionMode","OFF"); + evidence.put("sourceHttpCalls",0);evidence.put("assertionsCompleted",false); + scope=new SelfopSyncScope();scope.setId(IdWorker.getId());scope.setSourceAccount(ACCOUNT); + scope.setWarehouseCode(warehouse);scope.setDataFamily("INVENTORY"); + scope.setScopeJson(json(Map.of("fixtureOrigin","OFFLINE_SYNTHETIC_TRANSACTION_ACCEPTANCE", + "quantityUnit","PCS","businessTimezone","Asia/Shanghai","sourceTimezone","Asia/Shanghai"))); + scope.setScopeHash(hash(scope.getScopeJson()));scope.setBusinessTimezone("Asia/Shanghai");scope.setSourceTimezone("Asia/Shanghai"); + scope.setScheduleCron("0 0 8 * * *");scope.setStaleAfterSeconds(93600); + scope.setFactRevision(0L);scope.setPublishedFactRevision(0L);scope.setRebuildRequired(0); + scope.setFenceToken(1L);scope.setLeaseOwner(owner);scope.setLeaseUntilUtc(utcNow().plusMinutes(30)); + scope.setReviewRequestBudget(1);scope.setReviewTimeBudgetSeconds(1);scope.setReviewTargetDays(7); + scope.setCreatedAt(utcNow());scope.setUpdatedAt(utcNow()); + transactions.executeWithoutResult(s->scopes.insert(scope));evidence.put("scopeId",scope.getId().toString()); + } + + @Test void pointerRunTaskSupersededAndDailyElectionRollBackAfterRealSqlFault() { + var old=candidate(day.minusDays(2),day.minusDays(1),false);publish(old); + var next=candidate(day,day,false);var before=snapshot(old,next); + fault.armed=true; + RuntimeException failure=assertThrows(RuntimeException.class,()->publish(next)); + boolean expected=false;for(Throwable cause=failure;cause!=null;cause=cause.getCause()) + if(FAULT.equals(cause.getMessage()))expected=true; + assertTrue(expected,"Expected injected failure after actual run SQL"); + assertTrue(fault.pointerWritten&&fault.realTransaction,"Actual pointer update and transaction must have occurred"); + var after=snapshot(old,next);assertEquals(before,after,"Rollback must restore every publication field"); + assertEquals("READY",runs.selectById(next.getId()).getStatus());assertNull(runs.selectById(next.getId()).getInventoryDailySampleDate()); + assertEquals(0L,sampleCount(day)); + evidence.put("before",before);evidence.put("afterRollback",after); + evidence.put("faultAfterRealScopeAndRunWrites",true);evidence.put("assertionsCompleted",true); + } + + @Test void successfulPublicationAndPublishedRecoveryKeepExactCommittedResult() { + var run=candidate(day,day,false);publish(run);var before=snapshot(run); + assertPublished(run,true);publish(run);assertEquals(before,snapshot(run)); + assertEquals(1L,sampleCount(day)); + evidence.put("afterFirstCommit",before);evidence.put("afterRecovery",snapshot(run)); + evidence.put("assertionsCompleted",true); + } + + @Test void laterSameDayPublicationPreservesFirstRepresentativeAndItsRows() { + var first=candidate(day,day,false);publish(first); + var second=candidate(day,day,false);publish(second); + assertPublished(second,false);assertEquals(first.getId(),dailyRun(day));assertEquals(1L,sampleCount(day)); + assertEquals(1L,rowCount("ll_selfop_inventory_snapshot","snapshot_run_id",first.getId())); + assertEquals(1L,rowCount("ll_selfop_sku_profile","profile_run_id",first.getId())); + assertNotNull(runs.selectById(first.getId()).getSupersededAtUtc()); + evidence.put("afterSecondCommit",snapshot(first,second));evidence.put("assertionsCompleted",true); + } + + @Test void crossMidnightPublicationHasNoDailyMarkerAndCannotBackfill() { + var first=candidate(day,day,false);publish(first); + var crossed=candidate(day.minusDays(1),day,true);publish(crossed); + assertPublished(crossed,false);assertEquals(first.getId(),dailyRun(day)); + assertEquals(0L,sampleCount(day.minusDays(1)));assertEquals(1L,sampleCount(day)); + evidence.put("afterCrossMidnightCommit",snapshot(first,crossed));evidence.put("assertionsCompleted",true); + } + + private SelfopSyncRun candidate(LocalDate planned,LocalDate observed,boolean crossed) { + LocalDateTime start=crossed?utcDate(observed).minusMinutes(1):utcDate(observed).plusHours(1); + LocalDateTime finish=crossed?utcDate(observed).plusMinutes(1):start.plusMinutes(1); + var task=new SelfopSyncTask();task.setId(IdWorker.getId());task.setTaskNo("SYNTH-TXN-"+UUID.randomUUID().toString().toUpperCase(Locale.ROOT)); + task.setExecutionPurpose("BUSINESS");task.setSourceAccount(ACCOUNT);task.setWarehouseCode(warehouse); + task.setTaskRole("WAREHOUSE_SYNC");task.setScopeId(scope.getId());task.setTaskType("SCHEDULED"); + task.setPlannedAtUtc(utcDate(planned).plusHours(1));task.setRequestKey("SYNTH-TXN-"+UUID.randomUUID()); + task.setRequestHash(scope.getScopeHash());task.setScopeHash(scope.getScopeHash());task.setQueryScopeJson(scope.getScopeJson()); + task.setStatus("RUNNING");task.setApplyState("NOT_APPLICABLE");task.setAttemptCount(1); + task.setCreatedAt(utcNow());task.setUpdatedAt(utcNow()); + var run=new SelfopSyncRun();run.setId(IdWorker.getId());run.setTaskId(task.getId());run.setScopeId(scope.getId()); + run.setExecutionPurpose("BUSINESS");run.setSourceAccount(ACCOUNT);run.setWarehouseCode(warehouse);run.setRunRole("WAREHOUSE_SYNC"); + run.setAttemptNo(1);run.setScopeHash(scope.getScopeHash());run.setScopeSnapshotJson(scope.getScopeJson());run.setFenceToken(1L); + run.setBasePublishedRunId(scopes.selectById(scope.getId()).getPublishedRunId());run.setStatus("READY");run.setLastPhase("BUILD"); + run.setFetchStartedAtUtc(start);run.setFetchFinishedAtUtc(finish);run.setPageCount(1);run.setSourceTotalCount(1L); + run.setUniqueObjectCount(1L);run.setCoverageStatus("COMPLETE");run.setCreatedAt(utcNow());run.setUpdatedAt(utcNow()); + var stock=new SelfopInventorySnapshot();stock.setId(IdWorker.getId());stock.setSnapshotRunId(run.getId()); + stock.setSourceAccount(ACCOUNT);stock.setWarehouseCode(warehouse);stock.setCustomerCode("SYNTH-TXN-OWNER"); + stock.setSkuCode("SYNTH-TXN-SKU");stock.setProductName("合成事务验收商品，非领星数据");stock.setProductQuality(0); + stock.setPhysicalQuantity(BigDecimal.TEN);stock.setAvailableQuantity(BigDecimal.TEN);stock.setFrozenQuantity(BigDecimal.ZERO); + stock.setTransitQuantity(BigDecimal.ZERO);stock.setObservedAtUtc(finish);stock.setQualityStatus("VALID"); + stock.setCreatedAt(utcNow());stock.setUpdatedAt(utcNow()); + var profile=new SelfopSkuProfile();profile.setId(IdWorker.getId());profile.setProfileRunId(run.getId()); + profile.setSourceAccount(ACCOUNT);profile.setWarehouseCode(warehouse);profile.setCustomerCode(stock.getCustomerCode()); + profile.setSkuCode(stock.getSkuCode());profile.setProductName(stock.getProductName());profile.setQuantityUnit("PCS"); + profile.setFirstSeenAtUtc(start);profile.setEarliestEvidenceAtUtc(start);profile.setEvidenceType("INVENTORY_OBSERVATION"); + profile.setEvidenceRunId(run.getId());profile.setLastSeenRunId(run.getId());profile.setCreatedAt(utcNow());profile.setUpdatedAt(utcNow()); + transactions.executeWithoutResult(s->{tasks.insert(task);runs.insert(run);stocks.insert(stock);profiles.insert(profile);}); + return run; + } + private void publish(SelfopSyncRun run){service.publish(scope.getId(),run.getId(),owner,1L);} + private void assertPublished(SelfopSyncRun candidate,boolean daily){ + var saved=runs.selectById(candidate.getId());var task=tasks.selectById(saved.getTaskId()); + assertEquals(candidate.getId(),scopes.selectById(scope.getId()).getPublishedRunId()); + assertEquals("PUBLISHED",saved.getStatus());assertNotNull(saved.getPublishedAtUtc()); + assertEquals("SUCCEEDED",task.getStatus());assertEquals(saved.getId(),task.getResultPublishedRunId()); + assertEquals(daily?day:null,saved.getInventoryDailySampleDate()); + assertEquals(daily?"DAILY_SCHEDULED_FIRST_V1":null,saved.getInventorySampleRuleVersion()); + assertEquals("OFF",properties.getCollectionMode());assertFalse(properties.isExecutionEnabled()); + } + /** JdbcTemplate uses independent connections after service transaction completes. */ + private Map snapshot(SelfopSyncRun... selected){ + assertFalse(TransactionSynchronizationManager.isActualTransactionActive()); + var result=new LinkedHashMap(); + result.put("scope",jdbc.queryForMap("SELECT CAST(published_run_id AS CHAR) published_run_id FROM ll_selfop_sync_scope WHERE id=?",scope.getId())); + for(var run:selected){ + result.put("run-"+run.getId(),jdbc.queryForMap("SELECT status,published_at_utc,superseded_at_utc,inventory_daily_sample_date,inventory_sample_rule_version FROM ll_selfop_sync_run WHERE id=?",run.getId())); + result.put("task-"+run.getTaskId(),jdbc.queryForMap("SELECT status,CAST(result_published_run_id AS CHAR) result_published_run_id FROM ll_selfop_sync_task WHERE id=?",run.getTaskId())); + }return result; + } + private long sampleCount(LocalDate date){return jdbc.queryForObject("SELECT COUNT(*) FROM ll_selfop_sync_run WHERE source_account=? AND warehouse_code=? AND inventory_daily_sample_date=?",Long.class,ACCOUNT,warehouse,date);} + private Long dailyRun(LocalDate date){return jdbc.queryForObject("SELECT id FROM ll_selfop_sync_run WHERE source_account=? AND warehouse_code=? AND inventory_daily_sample_date=?",Long.class,ACCOUNT,warehouse,date);} + private long rowCount(String table,String column,long id){return jdbc.queryForObject("SELECT COUNT(*) FROM "+table+" WHERE "+column+"=?",Long.class,id);} + private static LocalDateTime utcDate(LocalDate date){return date.atStartOfDay(ZONE).withZoneSameInstant(ZoneOffset.UTC).toLocalDateTime();} + private static String requiredEnv(String name){String value=System.getenv(name);if(value==null||value.isBlank())throw new IllegalStateException("Missing TEST variable "+name);return value;} + + @AfterEach void preserveEvidenceAndDisableRemainingDrafts() throws Exception { + if(jdbc!=null&&scope!=null){ + // Retain every row. Prevent later scheduler runs from consuming unsuccessful synthetic drafts. + transactions.executeWithoutResult(s->{ + jdbc.update("UPDATE ll_selfop_sync_task SET status='DEAD',next_retry_at_utc=NULL WHERE source_account=? AND scope_id=? AND status<>'SUCCEEDED'",ACCOUNT,scope.getId()); + jdbc.update("UPDATE ll_selfop_sync_scope SET lease_owner=NULL,lease_until_utc=NULL WHERE id=? AND source_account=? AND lease_owner=?",scope.getId(),ACCOUNT,owner); + }); + evidence.put("retainedRunHeads",jdbc.queryForList("SELECT CAST(id AS CHAR) id,status,inventory_daily_sample_date FROM ll_selfop_sync_run WHERE scope_id=? ORDER BY id",scope.getId())); + } + if(!evidence.isEmpty()){ + evidence.put("at",Instant.now().toString());evidence.put("noRowsDeleted",true); + var directory=Path.of("target");Files.createDirectories(directory); + new ObjectMapper().findAndRegisterModules().writerWithDefaultPrettyPrinter().writeValue( + directory.resolve("selfop-mysql-publication-"+token+".json").toFile(),evidence); + } + } + + @Intercepts(@Signature(type=Executor.class,method="update",args={MappedStatement.class,Object.class})) + public static class WriteFault implements Interceptor { + boolean armed,pointerWritten,realTransaction; + @Override public Object intercept(Invocation invocation) throws Throwable { + Object result=invocation.proceed(); // Every mapped SQL really executes before the fault hook. + if(armed){String id=((MappedStatement)invocation.getArgs()[0]).getId(); + if(id.equals(SelfopSyncScopeMapper.class.getName()+".updateById"))pointerWritten=true; + if(id.equals(SelfopSyncRunMapper.class.getName()+".updateById")){ + armed=false;realTransaction=TransactionSynchronizationManager.isActualTransactionActive(); + throw new IllegalStateException(FAULT); + } + }return result; + } + } +}