package com.yunhai.question.service; import com.fasterxml.jackson.databind.ObjectMapper; import com.yunhai.question.entity.PullCursor; import com.yunhai.question.entity.Questionnaire; import com.yunhai.question.fc.FormCenterClient; import com.yunhai.question.repository.AnswerProjectionRepository; import com.yunhai.question.repository.PullCursorRepository; import com.yunhai.question.repository.QuestionnaireRepository; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.support.TransactionTemplate; import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** * 投影服务单测(H2 真库 + mock FormCenterClient)——任务书点名的四件之二/三: * 游标推进、投影幂等 upsert(重放同窗口不产生重复行、不重复累计 seen_count)、墓碑翻位。 */ @DataJpaTest class ProjectionServiceTest { private static final Long SCH_ID = 990001L; private static final String FORM_UUID = "f0123456789abcdef0123456789abcde"; @Autowired private QuestionnaireRepository questionnaireRepository; @Autowired private PullCursorRepository pullCursorRepository; @Autowired private AnswerProjectionRepository answerProjectionRepository; @Autowired private PlatformTransactionManager txManager; private FormCenterClient fc; private ProjectionService service; @BeforeEach void setUp() { fc = mock(FormCenterClient.class); service = new ProjectionService(questionnaireRepository, pullCursorRepository, answerProjectionRepository, fc, new ObjectMapper(), new TransactionTemplate(txManager), 10, 100); } private Questionnaire questionnaire(String formUuid) { Questionnaire q = new Questionnaire(); q.setSchId(SCH_ID); q.setFormUuid(formUuid); q.setFormCode("q-001"); q.setTitleSnapshot("测试问卷"); q.setStatus(Questionnaire.STATUS_COLLECTING); q.setAnonymous(false); return questionnaireRepository.save(q); } /** FC 行(snake_case 闭集) */ private static Map fcRow(String uuid, boolean deleted, Object... valuePairs) { Map values = new LinkedHashMap<>(); for (int i = 0; i + 1 < valuePairs.length; i += 2) { values.put(String.valueOf(valuePairs[i]), valuePairs[i + 1]); } Map row = new HashMap<>(); row.put("answer_uuid", uuid); row.put("form_uuid", FORM_UUID); row.put("submitter_id", 101); row.put("subject_id", null); row.put("status", "待处理"); row.put("values", values); row.put("create_time", "2026-10-21 10:00:00"); row.put("update_time", "2026-10-21 10:00:00"); row.put("is_deleted", deleted ? 1 : 0); return row; } /** after=afterArg 的页(rows 可空) */ private void stubPage(String afterArg, String nextCursor, Map... rows) { when(fc.pullAnswers(any(), anyString(), eq(afterArg), anyInt())) .thenReturn(new FormCenterClient.AnswersPage(Arrays.asList(rows), nextCursor)); } @Test void advancesCursorAcrossPages() { questionnaire(FORM_UUID); stubPage(null, "a1", fcRow("a1", false, "q1", "m")); stubPage("a1", "a2", fcRow("a2", false, "q1", "f")); stubPage("a2", null); service.pullOnce(SCH_ID, FORM_UUID); assertEquals("a2", pullCursorRepository.findById(FORM_UUID).get().getCursor()); assertEquals(2, answerProjectionRepository.countByFormUuid(FORM_UUID)); // 第二轮:游标从 a2 起,只拉增量 stubPage("a2", "a3", fcRow("a3", false, "q1", "m")); stubPage("a3", null); int fresh = service.pullOnce(SCH_ID, FORM_UUID); assertEquals(1, fresh); assertEquals("a3", pullCursorRepository.findById(FORM_UUID).get().getCursor()); assertEquals(3, answerProjectionRepository.countByFormUuid(FORM_UUID)); // 增量轮从 a2 起——验证游标确已推进(a2 之前的窗口未再请求) verify(fc, times(1)).pullAnswers(any(), anyString(), eq((String) null), anyInt()); verify(fc, times(1)).pullAnswers(any(), anyString(), eq("a1"), anyInt()); verify(fc, times(2)).pullAnswers(any(), anyString(), eq("a2"), anyInt()); } @Test void replaySameWindowIsIdempotent() { questionnaire(FORM_UUID); stubPage(null, "a1", fcRow("a1", false, "q1", "m")); stubPage("a1", null); service.pullOnce(SCH_ID, FORM_UUID); // 模拟游标回拨(运维/异常后重放):从头再拉同窗口 + 新行 a2——a1 重放不算新行 PullCursor rollback = pullCursorRepository.findById(FORM_UUID).orElseThrow(AssertionError::new); rollback.setCursor(null); pullCursorRepository.saveAndFlush(rollback); stubPage(null, "a2", fcRow("a1", false, "q1", "m"), fcRow("a2", false, "q1", "f")); stubPage("a2", null); int fresh = service.pullOnce(SCH_ID, FORM_UUID); assertEquals(1, fresh); assertEquals(2, answerProjectionRepository.countByFormUuid(FORM_UUID)); PullCursor cursor = pullCursorRepository.findById(FORM_UUID).get(); assertEquals(2L, cursor.getSeenCount(), "重放行不重复累计 seen_count"); assertEquals("a2", cursor.getCursor()); } @Test void tombstoneFlipsIsDeletedWithoutDroppingRow() { questionnaire(FORM_UUID); stubPage(null, "a1", fcRow("a1", false, "q1", "m")); stubPage("a1", null); service.pullOnce(SCH_ID, FORM_UUID); assertEquals(1, answerProjectionRepository.countByFormUuidAndIsDeletedFalse(FORM_UUID)); // FC 侧软删 → 墓碑行:同一 answer_uuid,is_deleted=1 stubPage("a1", null, fcRow("a1", true, "q1", "m")); service.pullOnce(SCH_ID, FORM_UUID); assertEquals(0, answerProjectionRepository.countByFormUuidAndIsDeletedFalse(FORM_UUID)); assertEquals(1, answerProjectionRepository.countByFormUuid(FORM_UUID), "墓碑不删行"); assertTrue(answerProjectionRepository.findByAnswerUuid("a1").getIsDeleted()); assertEquals(1L, pullCursorRepository.findById(FORM_UUID).get().getSeenCount(), "墓碑是已见行的翻位,不计新行"); } @Test void updatedRowKeepsSingleRow() { questionnaire(FORM_UUID); stubPage(null, "a1", fcRow("a1", false, "q1", "m")); stubPage("a1", null); service.pullOnce(SCH_ID, FORM_UUID); // 状态被 FC 侧更新后重放:同 uuid 更新而非插入 Map updated = fcRow("a1", false, "q1", "f"); updated.put("status", "已处理"); stubPage("a1", null, updated); service.pullOnce(SCH_ID, FORM_UUID); assertEquals(1, answerProjectionRepository.countByFormUuid(FORM_UUID)); assertEquals("已处理", answerProjectionRepository.findByAnswerUuid("a1").getStatus()); assertEquals(1L, pullCursorRepository.findById(FORM_UUID).get().getSeenCount()); } @Test void firstPullWithEmptyWindowInitializesCursor() { questionnaire(FORM_UUID); stubPage(null, null); int fresh = service.pullOnce(SCH_ID, FORM_UUID); assertEquals(0, fresh); PullCursor cursor = pullCursorRepository.findById(FORM_UUID).get(); assertNull(cursor.getCursor(), "空窗口不推进游标(null=从未有行)"); assertEquals("OK", cursor.getLastStatus()); } @Test void reconciliationWarnsOnSeenCountDrift() { questionnaire(FORM_UUID); stubPage(null, "a1", fcRow("a1", false, "q1", "m")); stubPage("a1", null); service.pullOnce(SCH_ID, FORM_UUID); // 模拟投影表被外部删行:seen_count=1 但投影总量=0 → 漂移告警路径(不抛错) answerProjectionRepository.deleteAll(); service.checkReconciliation(FORM_UUID, pullCursorRepository.findById(FORM_UUID).get()); assertEquals(0, answerProjectionRepository.countByFormUuid(FORM_UUID)); } }