test(selfop): verify publication transactions against opt-in TEST MySQL
This commit is contained in:
+237
@@ -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<String,Object> 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<String,Object> snapshot(SelfopSyncRun... selected){
|
||||
assertFalse(TransactionSynchronizationManager.isActualTransactionActive());
|
||||
var result=new LinkedHashMap<String,Object>();
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user