|
|
@@ -0,0 +1,518 @@
|
|
|
+package com.malk.benteler.service;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
+import com.malk.benteler.config.BentelerYidaConf;
|
|
|
+import com.malk.benteler.dto.DingTalkSyncTask;
|
|
|
+import com.malk.server.aliwork.YDAuth;
|
|
|
+import com.malk.server.aliwork.YDConf;
|
|
|
+import com.malk.service.aliwork.YDClient_Form;
|
|
|
+import com.malk.service.dingtalk.DDClient_Personnel;
|
|
|
+import com.malk.service.dingtalk.DDClient_Contacts;
|
|
|
+import com.malk.service.dingtalk.DDService;
|
|
|
+import org.junit.Before;
|
|
|
+import org.junit.Test;
|
|
|
+import org.mockito.ArgumentCaptor;
|
|
|
+
|
|
|
+import java.time.Clock;
|
|
|
+import java.time.Instant;
|
|
|
+import java.time.ZoneOffset;
|
|
|
+import java.util.Arrays;
|
|
|
+import java.util.Collections;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.concurrent.CountDownLatch;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
+
|
|
|
+import static org.junit.Assert.assertEquals;
|
|
|
+import static org.junit.Assert.assertFalse;
|
|
|
+import static org.junit.Assert.assertTrue;
|
|
|
+import static org.mockito.ArgumentMatchers.any;
|
|
|
+import static org.mockito.ArgumentMatchers.anyMap;
|
|
|
+import static org.mockito.ArgumentMatchers.anyString;
|
|
|
+import static org.mockito.ArgumentMatchers.eq;
|
|
|
+import static org.mockito.ArgumentMatchers.isNull;
|
|
|
+import static org.mockito.Mockito.doThrow;
|
|
|
+import static org.mockito.Mockito.mock;
|
|
|
+import static org.mockito.Mockito.never;
|
|
|
+import static org.mockito.Mockito.times;
|
|
|
+import static org.mockito.Mockito.verify;
|
|
|
+import static org.mockito.Mockito.when;
|
|
|
+
|
|
|
+/**
|
|
|
+ * {@link BentelerDingTalkSyncTaskService} durable queue behavior tests.
|
|
|
+ */
|
|
|
+public class BentelerDingTalkSyncTaskServiceTest {
|
|
|
+
|
|
|
+ private static final Instant NOW = Instant.parse("2026-08-15T08:00:00Z");
|
|
|
+
|
|
|
+ private YDClient_Form ydClientForm;
|
|
|
+ private DDService ddService;
|
|
|
+ private DDClient_Personnel personnelClient;
|
|
|
+ private DDClient_Contacts contactsClient;
|
|
|
+ private BentelerYidaConf conf;
|
|
|
+ private BentelerDingTalkSyncTaskService service;
|
|
|
+
|
|
|
+ @Before
|
|
|
+ public void setUp() {
|
|
|
+ ydClientForm = mock(YDClient_Form.class);
|
|
|
+ ddService = mock(DDService.class);
|
|
|
+ personnelClient = mock(DDClient_Personnel.class);
|
|
|
+ contactsClient = mock(DDClient_Contacts.class);
|
|
|
+ conf = TestBentelerYidaConf.create();
|
|
|
+ YDConf ydConf = new YDConf();
|
|
|
+ ydConf.setAppType("APP_TEST");
|
|
|
+ ydConf.setSystemToken("SYSTEM_TEST");
|
|
|
+ service = new BentelerDingTalkSyncTaskService(ydClientForm, ydConf, ddService,
|
|
|
+ personnelClient, contactsClient, conf, Clock.fixed(NOW, ZoneOffset.UTC));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterUpdate_createsPendingTaskTenMinutesLater() {
|
|
|
+ service.enqueueRosterUpdate("source_1", 0, "user_1", "众川");
|
|
|
+
|
|
|
+ Map<String, Object> saved = capturedUpsertedForm();
|
|
|
+ assertEquals("source_1:0:ROSTER_FIELD_UPDATE:outsourcing_code", saved.get("task_key"));
|
|
|
+ assertEquals(DingTalkSyncTask.ROSTER_FIELD_UPDATE, saved.get("task_type"));
|
|
|
+ assertEquals(DingTalkSyncTask.PENDING, saved.get("task_status"));
|
|
|
+ assertEquals(0, ((Number) saved.get("retry_count")).intValue());
|
|
|
+ assertEquals(NOW.plusSeconds(600).toEpochMilli(),
|
|
|
+ ((Number) saved.get("next_execute_at")).longValue());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterUpdate_usesStructuredSearchCondition() {
|
|
|
+ service.enqueueRosterUpdate("source_1", 0, "user_1", "众川");
|
|
|
+
|
|
|
+ ArgumentCaptor<String> search = ArgumentCaptor.forClass(String.class);
|
|
|
+ verify(ydClientForm).upsertForm(any(YDAuth.class), eq("FORM_TASK"),
|
|
|
+ search.capture(), anyString(), anyMap());
|
|
|
+ List<Map> conditions = JSON.parseArray(search.getValue(), Map.class);
|
|
|
+ assertEquals(1, conditions.size());
|
|
|
+ assertEquals("task_key", conditions.get(0).get("key"));
|
|
|
+ assertEquals("source_1:0:ROSTER_FIELD_UPDATE:outsourcing_code",
|
|
|
+ conditions.get(0).get("value"));
|
|
|
+ assertEquals("TEXT", conditions.get(0).get("type"));
|
|
|
+ assertEquals("eq", conditions.get(0).get("operator"));
|
|
|
+ assertEquals("TextField", conditions.get(0).get("componentName"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterUpdate_blankValueDoesNotCreateTask() {
|
|
|
+ service.enqueueRosterUpdate("source_1", 0, "user_1", " ");
|
|
|
+
|
|
|
+ verify(ydClientForm, never()).upsertForm(any(), anyString(), anyString(),
|
|
|
+ anyString(), anyMap());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterClear_createsTaskWithEmptyFieldValue() {
|
|
|
+ service.enqueueRosterClear("source_1", 0, "user_1");
|
|
|
+
|
|
|
+ assertEquals("", capturedUpsertedForm().get("field_value"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_contactDepartmentPutsTargetDepartmentFirst() {
|
|
|
+ Map<String, Object> record = taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "100");
|
|
|
+ Map<String, Object> data = formData(record);
|
|
|
+ data.put("task_type", DingTalkSyncTask.CONTACT_PRIMARY_DEPARTMENT_UPDATE);
|
|
|
+ data.put("task_key", "task_1:0:CONTACT_PRIMARY_DEPARTMENT_UPDATE:100");
|
|
|
+ dueTasks(record);
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ when(contactsClient.getUserInfoById("cached_token", "user_1"))
|
|
|
+ .thenReturn(Collections.singletonMap("dept_id_list", Arrays.asList(200L, 100L)));
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ ArgumentCaptor<Map> bodyCaptor = ArgumentCaptor.forClass(Map.class);
|
|
|
+ verify(contactsClient).updateUser(eq("cached_token"), eq("user_1"), bodyCaptor.capture());
|
|
|
+ assertEquals(Arrays.asList(100L, 200L), bodyCaptor.getValue().get("dept_id_list"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterUpdate_existingUnfinishedTaskIsResetInsteadOfDuplicated() {
|
|
|
+ when(ydClientForm.searchForm(any(YDAuth.class), eq("FORM_TASK"), anyString(),
|
|
|
+ eq(1), eq(100), isNull())).thenReturn(page(Collections.singletonList(
|
|
|
+ taskRecord("task_1", DingTalkSyncTask.RETRYING, 3,
|
|
|
+ NOW.minusSeconds(60), "old_user", "old value"))));
|
|
|
+
|
|
|
+ service.enqueueRosterUpdate("source_1", 0, "new_user", "new value");
|
|
|
+
|
|
|
+ Map<String, Object> updated = capturedUpsertedForm();
|
|
|
+ assertEquals("new_user", updated.get("user_id"));
|
|
|
+ assertEquals("new value", updated.get("field_value"));
|
|
|
+ assertEquals(DingTalkSyncTask.PENDING, updated.get("task_status"));
|
|
|
+ assertEquals(0, ((Number) updated.get("retry_count")).intValue());
|
|
|
+ assertEquals(NOW.plusSeconds(600).toEpochMilli(),
|
|
|
+ ((Number) updated.get("next_execute_at")).longValue());
|
|
|
+ assertEquals("", updated.get("last_error"));
|
|
|
+ assertEquals("", updated.get("completed_at"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void enqueueRosterUpdate_concurrentCallsUseSameAtomicUpsertKey() throws Exception {
|
|
|
+ CountDownLatch start = new CountDownLatch(1);
|
|
|
+ Thread first = new Thread(() -> awaitAndEnqueue(start));
|
|
|
+ Thread second = new Thread(() -> awaitAndEnqueue(start));
|
|
|
+ first.start();
|
|
|
+ second.start();
|
|
|
+ start.countDown();
|
|
|
+ first.join(5000);
|
|
|
+ second.join(5000);
|
|
|
+
|
|
|
+ ArgumentCaptor<String> search = ArgumentCaptor.forClass(String.class);
|
|
|
+ verify(ydClientForm, times(2)).upsertForm(any(YDAuth.class), eq("FORM_TASK"),
|
|
|
+ search.capture(), anyString(), anyMap());
|
|
|
+ List<Map> expected = Collections.singletonList(new HashMap<String, Object>() {{
|
|
|
+ put("key", "task_key");
|
|
|
+ put("value", "source_1:0:ROSTER_FIELD_UPDATE:outsourcing_code");
|
|
|
+ put("type", "TEXT");
|
|
|
+ put("operator", "eq");
|
|
|
+ put("componentName", "TextField");
|
|
|
+ }});
|
|
|
+ assertEquals(expected, JSON.parseArray(search.getAllValues().get(0), Map.class));
|
|
|
+ assertEquals(expected, JSON.parseArray(search.getAllValues().get(1), Map.class));
|
|
|
+ verify(ydClientForm, never()).saveForm(any(), anyString(), anyString(), any());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_futureTaskIsNotExecuted() {
|
|
|
+ when(ydClientForm.searchForm(any(YDAuth.class), eq("FORM_TASK"), eq("{}"),
|
|
|
+ eq(1), eq(100), isNull())).thenReturn(page(Collections.singletonList(
|
|
|
+ taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW.plusSeconds(1), "user_1", "众川"))));
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient, never()).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ verify(ydClientForm, never()).updateForm(any(), anyString(), anyString(), anyMap());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_successMarksTaskCompleted() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川"));
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient).updateEmployeeRosterField("cached_token", 4784847516L,
|
|
|
+ "user_1", "sys00", "outsourcing_code", "众川", null);
|
|
|
+ Map<String, Object> updated = capturedUpdate("task_1");
|
|
|
+ assertEquals(DingTalkSyncTask.SUCCESS, updated.get("task_status"));
|
|
|
+ assertEquals(NOW.toEpochMilli(), ((Number) updated.get("completed_at")).longValue());
|
|
|
+ assertEquals("", updated.get("last_error"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_firstFailureRetriesAfterFifteenMinutes() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川"));
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ doThrow(new IllegalStateException("temporary failure"))
|
|
|
+ .when(personnelClient).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ Map<String, Object> updated = capturedUpdate("task_1");
|
|
|
+ assertEquals(DingTalkSyncTask.RETRYING, updated.get("task_status"));
|
|
|
+ assertEquals(1, ((Number) updated.get("retry_count")).intValue());
|
|
|
+ assertEquals(NOW.plusSeconds(900).toEpochMilli(),
|
|
|
+ ((Number) updated.get("next_execute_at")).longValue());
|
|
|
+ assertEquals("IllegalStateException: temporary failure", updated.get("last_error"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_fifthFailureMarksTaskFailed() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.RETRYING, 4,
|
|
|
+ NOW, "user_1", "众川"));
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ doThrow(new IllegalStateException("permanent failure"))
|
|
|
+ .when(personnelClient).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ Map<String, Object> updated = capturedUpdate("task_1");
|
|
|
+ assertEquals(DingTalkSyncTask.FAILED, updated.get("task_status"));
|
|
|
+ assertEquals(5, ((Number) updated.get("retry_count")).intValue());
|
|
|
+ assertEquals(NOW.toEpochMilli(), ((Number) updated.get("completed_at")).longValue());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_failureIsSanitizedAndDoesNotStopNextTask() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川"),
|
|
|
+ taskRecord("task_2", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_2", "外包二"));
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ doThrow(new IllegalStateException("access_token=secret\nrequest failed"))
|
|
|
+ .doNothing().when(personnelClient).updateEmployeeRosterField(
|
|
|
+ anyString(), any(), anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient, times(2)).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ List<Map<String, Object>> updates = capturedUpdates();
|
|
|
+ assertEquals("IllegalStateException: access_token=*** request failed",
|
|
|
+ updates.get(0).get("last_error"));
|
|
|
+ assertEquals(DingTalkSyncTask.SUCCESS, updates.get(1).get("task_status"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_failureStateWriteDoesNotStopNextTask() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川"),
|
|
|
+ taskRecord("task_2", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_2", "外包二"));
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ doThrow(new IllegalStateException("DingTalk failed"))
|
|
|
+ .doNothing().when(personnelClient).updateEmployeeRosterField(
|
|
|
+ anyString(), any(), anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ when(ydClientForm.updateForm(any(YDAuth.class), eq("task_1"), anyString(), anyMap()))
|
|
|
+ .thenReturn(Collections.emptyMap())
|
|
|
+ .thenThrow(new IllegalStateException("YiDa failed"));
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient, times(2)).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ verify(ydClientForm, org.mockito.Mockito.atLeastOnce()).updateForm(
|
|
|
+ any(YDAuth.class), eq("task_2"), anyString(), anyMap());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_malformedRecordDoesNotStopValidTask() {
|
|
|
+ Map<String, Object> malformed = new HashMap<>();
|
|
|
+ malformed.put("formInstanceId", "task_bad\naccess_token=secret");
|
|
|
+ malformed.put("formData", "{not-json");
|
|
|
+ dueTasks(malformed, taskRecord("task_2", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_2", "外包二"));
|
|
|
+ claimSucceeds("task_2");
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient).updateEmployeeRosterField("cached_token", 4784847516L,
|
|
|
+ "user_2", "sys00", "outsourcing_code", "外包二", null);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_competingClaimOwnerPreventsRosterCall() {
|
|
|
+ dueTasks(taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川"));
|
|
|
+ when(ydClientForm.getForm(any(YDAuth.class), eq("task_1"), isNull()))
|
|
|
+ .thenReturn(claimedTask("task_1", "CLAIM:other-instance",
|
|
|
+ NOW.plusSeconds(600)));
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient, never()).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ ArgumentCaptor<String> claimJson = ArgumentCaptor.forClass(String.class);
|
|
|
+ ArgumentCaptor<Map> claimExt = ArgumentCaptor.forClass(Map.class);
|
|
|
+ verify(ydClientForm).updateForm(any(YDAuth.class), eq("task_1"),
|
|
|
+ claimJson.capture(), claimExt.capture());
|
|
|
+ Map<String, Object> claim = JSON.parseObject(claimJson.getValue(), Map.class);
|
|
|
+ assertEquals(NOW.plusSeconds(600).toEpochMilli(),
|
|
|
+ ((Number) claim.get("next_execute_at")).longValue());
|
|
|
+ assertEquals(false, claimExt.getValue().get("useLatestVersion"));
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_expiredClaimIsRecoveredAndExecuted() {
|
|
|
+ Map<String, Object> abandoned = taskRecord("task_1", DingTalkSyncTask.RETRYING, 2,
|
|
|
+ NOW, "user_1", "众川");
|
|
|
+ formData(abandoned).put("last_error", "CLAIM:dead-instance");
|
|
|
+ dueTasks(abandoned);
|
|
|
+ claimSucceeds("task_1");
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient).updateEmployeeRosterField("cached_token", 4784847516L,
|
|
|
+ "user_1", "sys00", "outsourcing_code", "众川", null);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_overlappingLocalRunReturnsWithoutSecondScan() throws Exception {
|
|
|
+ CountDownLatch entered = new CountDownLatch(1);
|
|
|
+ CountDownLatch release = new CountDownLatch(1);
|
|
|
+ when(ydClientForm.searchForm(any(YDAuth.class), eq("FORM_TASK"), eq("{}"),
|
|
|
+ eq(1), eq(100), isNull())).thenAnswer(invocation -> {
|
|
|
+ entered.countDown();
|
|
|
+ release.await(5, TimeUnit.SECONDS);
|
|
|
+ return page(Collections.emptyList());
|
|
|
+ });
|
|
|
+ Thread first = new Thread(service::executeDueTasks);
|
|
|
+ first.start();
|
|
|
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+ release.countDown();
|
|
|
+ first.join(5000);
|
|
|
+
|
|
|
+ verify(ydClientForm, times(1)).searchForm(any(YDAuth.class), eq("FORM_TASK"),
|
|
|
+ eq("{}"), eq(1), eq(100), isNull());
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_retryDelaysCoverThirtyAndSixtyMinuteBranches() {
|
|
|
+ assertRetryDelay(1, 30);
|
|
|
+ assertRetryDelay(2, 60);
|
|
|
+ assertRetryDelay(3, 60);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test
|
|
|
+ public void executeDueTasks_duplicateUnfinishedKeyExecutesOnlyCanonicalRecord() {
|
|
|
+ Map<String, Object> first = taskRecord("task_1", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川");
|
|
|
+ Map<String, Object> duplicate = taskRecord("task_2", DingTalkSyncTask.PENDING, 0,
|
|
|
+ NOW, "user_1", "众川");
|
|
|
+ formData(duplicate).put("task_key", formData(first).get("task_key"));
|
|
|
+ dueTasks(first, duplicate);
|
|
|
+ claimSucceeds("task_1");
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ verify(personnelClient, times(1)).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+ Map<String, Object> duplicateUpdate = capturedUpdate("task_2");
|
|
|
+ assertEquals(DingTalkSyncTask.FAILED, duplicateUpdate.get("task_status"));
|
|
|
+ assertTrue(String.valueOf(duplicateUpdate.get("last_error")).contains("重复任务"));
|
|
|
+ }
|
|
|
+
|
|
|
+ private void assertRetryDelay(int existingRetryCount, long expectedMinutes) {
|
|
|
+ resetMocks();
|
|
|
+ dueTasks(taskRecord("task_retry", DingTalkSyncTask.RETRYING, existingRetryCount,
|
|
|
+ NOW, "user_1", "众川"));
|
|
|
+ claimSucceeds("task_retry");
|
|
|
+ when(ddService.getAccessToken()).thenReturn("cached_token");
|
|
|
+ doThrow(new IllegalStateException("temporary failure"))
|
|
|
+ .when(personnelClient).updateEmployeeRosterField(anyString(), any(),
|
|
|
+ anyString(), anyString(), anyString(), anyString(), any());
|
|
|
+
|
|
|
+ service.executeDueTasks();
|
|
|
+
|
|
|
+ List<Map<String, Object>> updates = capturedUpdates("task_retry");
|
|
|
+ Map<String, Object> retry = updates.get(updates.size() - 1);
|
|
|
+ assertEquals(NOW.plusSeconds(expectedMinutes * 60).toEpochMilli(),
|
|
|
+ ((Number) retry.get("next_execute_at")).longValue());
|
|
|
+ }
|
|
|
+
|
|
|
+ private void resetMocks() {
|
|
|
+ org.mockito.Mockito.reset(ydClientForm, ddService, personnelClient);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void awaitAndEnqueue(CountDownLatch start) {
|
|
|
+ try {
|
|
|
+ start.await(5, TimeUnit.SECONDS);
|
|
|
+ service.enqueueRosterUpdate("source_1", 0, "user_1", "众川");
|
|
|
+ } catch (InterruptedException ex) {
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ throw new IllegalStateException(ex);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void claimSucceeds(String formInstanceId) {
|
|
|
+ claimSucceeds(formInstanceId, taskRecord(formInstanceId, DingTalkSyncTask.PENDING,
|
|
|
+ 0, NOW, "user_1", "众川"));
|
|
|
+ }
|
|
|
+
|
|
|
+ private void claimSucceeds(String formInstanceId, Map<String, Object> original) {
|
|
|
+ when(ydClientForm.getForm(any(YDAuth.class), eq(formInstanceId), isNull()))
|
|
|
+ .thenAnswer(invocation -> {
|
|
|
+ List<Map<String, Object>> updates = capturedUpdates(formInstanceId);
|
|
|
+ Map<String, Object> claim = updates.get(0);
|
|
|
+ Map<String, Object> persisted = new HashMap<>(original);
|
|
|
+ Map<String, Object> data = new HashMap<>(formData(original));
|
|
|
+ data.put("task_status", DingTalkSyncTask.RETRYING);
|
|
|
+ data.put("last_error", claim.get("last_error"));
|
|
|
+ data.put("next_execute_at", claim.get("next_execute_at"));
|
|
|
+ persisted.put("formData", data);
|
|
|
+ return persisted;
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Object> claimedTask(String instanceId, String owner, Instant leaseUntil) {
|
|
|
+ Map<String, Object> record = taskRecord(instanceId, DingTalkSyncTask.RETRYING, 0,
|
|
|
+ leaseUntil, "user_1", "众川");
|
|
|
+ formData(record).put("last_error", owner);
|
|
|
+ return record;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void dueTasks(Map<String, Object>... records) {
|
|
|
+ when(ydClientForm.searchForm(any(YDAuth.class), eq("FORM_TASK"), eq("{}"),
|
|
|
+ eq(1), eq(100), isNull())).thenReturn(page(Arrays.asList(records)));
|
|
|
+ for (Map<String, Object> record : records) {
|
|
|
+ Object id = record.get("formInstanceId");
|
|
|
+ if (id != null) {
|
|
|
+ claimSucceeds(String.valueOf(id), record);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Object> capturedUpsertedForm() {
|
|
|
+ ArgumentCaptor<String> json = ArgumentCaptor.forClass(String.class);
|
|
|
+ verify(ydClientForm).upsertForm(any(YDAuth.class), eq("FORM_TASK"), anyString(),
|
|
|
+ json.capture(), anyMap());
|
|
|
+ return JSON.parseObject(json.getValue(), Map.class);
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Object> capturedUpdate(String formInstanceId) {
|
|
|
+ List<Map<String, Object>> updates = capturedUpdates(formInstanceId);
|
|
|
+ return updates.get(updates.size() - 1);
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<Map<String, Object>> capturedUpdates() {
|
|
|
+ return Arrays.asList(capturedUpdate("task_1"), capturedUpdate("task_2"));
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<Map<String, Object>> capturedUpdates(String formInstanceId) {
|
|
|
+ ArgumentCaptor<String> json = ArgumentCaptor.forClass(String.class);
|
|
|
+ verify(ydClientForm, org.mockito.Mockito.atLeastOnce()).updateForm(
|
|
|
+ any(YDAuth.class), eq(formInstanceId), json.capture(), anyMap());
|
|
|
+ List<Map<String, Object>> values = new java.util.ArrayList<>();
|
|
|
+ for (String value : json.getAllValues()) {
|
|
|
+ values.add(JSON.parseObject(value, Map.class));
|
|
|
+ }
|
|
|
+ return values;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Object> taskRecord(String instanceId, String status, int retryCount,
|
|
|
+ Instant nextExecuteAt, String userId, String value) {
|
|
|
+ Map<String, Object> formData = new HashMap<>();
|
|
|
+ formData.put("task_key", instanceId + ":0:ROSTER_FIELD_UPDATE:outsourcing_code");
|
|
|
+ formData.put("task_type", DingTalkSyncTask.ROSTER_FIELD_UPDATE);
|
|
|
+ formData.put("source_id", "source_1");
|
|
|
+ formData.put("source_row", 0);
|
|
|
+ formData.put("user_id", userId);
|
|
|
+ formData.put("field_code", "outsourcing_code");
|
|
|
+ formData.put("field_value", value);
|
|
|
+ formData.put("task_status", status);
|
|
|
+ formData.put("retry_count", retryCount);
|
|
|
+ formData.put("next_execute_at", nextExecuteAt.toEpochMilli());
|
|
|
+ Map<String, Object> record = new HashMap<>();
|
|
|
+ record.put("formInstanceId", instanceId);
|
|
|
+ record.put("formData", formData);
|
|
|
+ return record;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, Object> page(List<Map<String, Object>> records) {
|
|
|
+ Map<String, Object> page = new HashMap<>();
|
|
|
+ page.put("data", records);
|
|
|
+ page.put("totalCount", records.size());
|
|
|
+ return page;
|
|
|
+ }
|
|
|
+
|
|
|
+ @SuppressWarnings("unchecked")
|
|
|
+ private Map<String, Object> formData(Map<String, Object> record) {
|
|
|
+ return (Map<String, Object>) record.get("formData");
|
|
|
+ }
|
|
|
+}
|
lfx