package jnpf.dmsService;
|
|
import jnpf.base.ActionResult;
|
import jnpf.dmsEntity.storage.approval.DmsApprovedBatch;
|
import jnpf.dmsMapper.DmsApprovedBatchMapper;
|
import jnpf.dmsService.impl.DmsWenjianBanbenServiceImpl;
|
import jnpf.exception.DataException;
|
import org.junit.jupiter.api.Test;
|
import java.util.Date;
|
import java.util.UUID;
|
import static org.mockito.Mockito.*;
|
import static org.junit.jupiter.api.Assertions.*;
|
|
class DmsApprovedFileRetryServiceTest {
|
DmsApprovedBatchMapper mapper = mock(DmsApprovedBatchMapper.class);
|
DmsApprovedFileAcceptanceService acceptance = mock(DmsApprovedFileAcceptanceService.class);
|
DmsApprovedContentService files = mock(DmsApprovedContentService.class);
|
DmsApprovedFileActivationService activation = mock(DmsApprovedFileActivationService.class);
|
DmsWenjianBanbenServiceImpl business = mock(DmsWenjianBanbenServiceImpl.class);
|
DmsApprovedDisableProcessor disables = mock(DmsApprovedDisableProcessor.class);
|
DmsInvalidationProperties invalidation = new DmsInvalidationProperties();
|
DmsApprovedFileRetryService service = new DmsApprovedFileRetryService(mapper,acceptance,files,activation,business,disables,invalidation);
|
DmsApprovedBatch batch(String status) {
|
DmsApprovedBatch b = new DmsApprovedBatch(); b.setId(UUID.randomUUID()); b.setTenantId("tenant");
|
b.setTaskId("task"); b.setFlowId("flow"); b.setAuditType("xinzeng_shenhe"); b.setActorId("actor");
|
b.setStatus(status); b.setApprovedAt(new Date()); b.setAttempts(1);
|
when(mapper.claim(eq("tenant"),eq(b.getId()),any(),anyBoolean())).thenAnswer(i -> { b.setLeaseToken(i.getArgument(2)); return 1; });
|
when(mapper.find("tenant","task","xinzeng_shenhe")).thenReturn(b);
|
return b;
|
}
|
@Test void unclaimedBatchHasNoSideEffects() {
|
DmsApprovedBatch b = batch("REGISTERING"); when(mapper.claim(any(),any(),any(),anyBoolean())).thenReturn(0);
|
service.process(b); verifyNoInteractions(acceptance,files,activation,business,disables);
|
}
|
@Test void sourceFailureLeavesApprovalQueuedAndSchedulesRetry() {
|
DmsApprovedBatch b = batch("PREPARING");
|
when(acceptance.stage(any(),any(),any(),any(),any(),any())).thenThrow(new DataException("source unavailable"));
|
assertThrows(DataException.class, () -> service.process(b));
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),any(),eq("source unavailable"),eq(60));
|
verifyNoInteractions(files,activation,business,disables);
|
}
|
@Test void pushFailureCanBeRetriedWithoutLosingReservedBatch() {
|
DmsApprovedBatch b = batch("REGISTERING");
|
when(acceptance.stage(any(),any(),any(),any(),any(),any())).thenReturn(b);
|
when(files.register(any())).thenThrow(new DataException("storage unavailable"));
|
assertThrows(DataException.class, () -> service.process(b));
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),any(),eq("storage unavailable"),eq(60));
|
ActionResult<DmsApprovedBatch> ok = new ActionResult<>(); ok.setCode(200);
|
DmsApprovedBatch result = new DmsApprovedBatch(); org.springframework.beans.BeanUtils.copyProperties(b,result); result.setStatus("APPLIED"); ok.setData(result);
|
doReturn(result).when(files).register(any());
|
when(activation.activate("tenant", b.getId())).thenReturn(result);
|
service.process(b);
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),any(),isNull(),eq(0));
|
}
|
@Test void lostReplyAfterCommittedRegistrationResumesAtActivationWithoutResendingFiles() {
|
DmsApprovedBatch b = batch("WAITING"); when(activation.activate("tenant",b.getId())).thenAnswer(i -> { b.setStatus("EFFECTIVE"); return b; });
|
service.process(b); verify(business).applyApproved(b); verifyNoInteractions(files,acceptance);
|
}
|
@Test void failedBusinessWriteRetriesProjectionOnly() {
|
DmsApprovedBatch b = batch("EFFECTIVE"); when(activation.activate("tenant",b.getId())).thenReturn(b);
|
doThrow(new DataException("business unavailable")).doNothing().when(business).applyApproved(b);
|
assertThrows(DataException.class, () -> service.process(b)); service.process(b);
|
verify(business,times(2)).applyApproved(b); verifyNoInteractions(files,acceptance);
|
verify(mapper,never()).fail(any(),any());
|
}
|
@Test void failedReceiptWriteStillPreservesOriginalFailureForLeaseRecovery() {
|
DmsApprovedBatch b = batch("PREPARING");
|
when(acceptance.stage(any(),any(),any(),any(),any(),any())).thenThrow(new DataException("source"));
|
when(mapper.finish(any(),any(),any(),any(),anyInt())).thenThrow(new DataException("database"));
|
RuntimeException error=assertThrows(DataException.class,()->service.process(b)); assertEquals(1,error.getSuppressed().length);
|
}
|
DmsApprovedBatch stopBatch() {
|
DmsApprovedBatch b=batch("WAITING");b.setAuditType("shixiao_shenqing");invalidation.setEnabled(true);invalidation.setAllBatches(true);
|
when(mapper.find("tenant","task","shixiao_shenqing")).thenReturn(b);
|
return b;
|
}
|
@Test void stopTasksUseSharedLeaseAndProcessorOnly() {
|
DmsApprovedBatch b=stopBatch();when(disables.isEnabled()).thenReturn(true);
|
service.process(b);verify(disables).process(b);
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),anyString(),isNull(),eq(0));
|
verifyNoInteractions(files,acceptance,activation,business);
|
}
|
@Test void stopFailureUsesSharedBackoffAndPreservesRowProgress() {
|
DmsApprovedBatch b=stopBatch();when(disables.isEnabled()).thenReturn(true);
|
doThrow(new DataException("row failed")).when(disables).process(b);
|
assertThrows(DataException.class,()->service.process(b));
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),anyString(),eq("row failed"),eq(60));
|
verify(mapper,never()).fail(any(),any());verifyNoInteractions(files,acceptance,activation,business);
|
}
|
@Test void disabledStopFeatureDoesNotClaimOrSendStopTask() {
|
DmsApprovedBatch b=stopBatch();service.process(b);
|
verify(mapper,never()).claim(any(),any(),any(),anyBoolean());verify(disables,never()).process(any());
|
}
|
@Test void disabledBackoffRetriesExistingDelayedTaskAndLeavesFailureReadyForNextScan() {
|
org.springframework.test.util.ReflectionTestUtils.setField(service,"retryBackoffEnabled",false);
|
DmsApprovedBatch b = batch("REGISTERING");
|
b.setAttempts(7);
|
b.setNextAttemptAt(new Date(System.currentTimeMillis() + 3600000));
|
when(acceptance.stage(any(),any(),any(),any(),any(),any())).thenThrow(new DataException("source unavailable"));
|
assertThrows(DataException.class, () -> service.process(b));
|
verify(mapper).claim(eq("tenant"),eq(b.getId()),anyString(),eq(false));
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),anyString(),eq("source unavailable"),eq(0));
|
verifyNoInteractions(files,activation,business,disables);
|
}
|
@Test void delayGrowsAndIsBounded() {
|
assertEquals(60,DmsApprovedFileRetryService.retryDelay(1)); assertEquals(120,DmsApprovedFileRetryService.retryDelay(2));
|
assertEquals(240,DmsApprovedFileRetryService.retryDelay(3)); assertEquals(3600,DmsApprovedFileRetryService.retryDelay(1000));
|
}
|
@Test void approvedRevocationUsesSharedProcessorWithoutRolloutConfiguration() {
|
DmsApprovedBatch b=batch("WAITING");b.setAuditType("chexiao_shenqing");
|
when(mapper.find("tenant","task","chexiao_shenqing")).thenReturn(b);
|
when(disables.isEnabled()).thenReturn(true);
|
service.process(b);
|
verify(mapper).claim(eq("tenant"),eq(b.getId()),anyString(),anyBoolean());
|
verify(disables).process(b);
|
verify(mapper).finish(eq("tenant"),eq(b.getId()),anyString(),isNull(),eq(0));
|
verifyNoInteractions(files,acceptance,activation,business);
|
}
|
}
|
