Add test cases on PipelineJobProgressPersistService

kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/persist/PipelineJobProgressPersistService.java

@@ -82,12 +82,7 @@ public static void remove(final String jobId) {
      * @param shardingItem sharding item
      */
     public static void notifyPersist(final String jobId, final int shardingItem) {
-        getPersistContext(jobId, shardingItem).ifPresent(persistContext -> persistContext.getUnhandledEventCount().incrementAndGet());
-    }
-
-    private static Optional<PipelineJobProgressPersistContext> getPersistContext(final String jobId, final int shardingItem) {
-        Map<Integer, PipelineJobProgressPersistContext> persistContextMap = JOB_PROGRESS_PERSIST_MAP.getOrDefault(jobId, Collections.emptyMap());
-        return Optional.ofNullable(persistContextMap.get(shardingItem));
+        getPersistContext(jobId, shardingItem).ifPresent(optional -> optional.getUnhandledEventCount().incrementAndGet());
     }
 
     /**
@@ -97,7 +92,11 @@ private static Optional<PipelineJobProgressPersistContext> getPersistContext(fin
      * @param shardingItem sharding item
      */
     public static void persistNow(final String jobId, final int shardingItem) {
-        getPersistContext(jobId, shardingItem).ifPresent(persistContext -> PersistJobContextRunnable.persist(jobId, shardingItem, persistContext));
+        getPersistContext(jobId, shardingItem).ifPresent(optional -> PersistJobContextRunnable.persist(jobId, shardingItem, optional));
+    }
+
+    private static Optional<PipelineJobProgressPersistContext> getPersistContext(final String jobId, final int shardingItem) {
+        return Optional.ofNullable(JOB_PROGRESS_PERSIST_MAP.getOrDefault(jobId, Collections.emptyMap()).get(shardingItem));
     }
 
     private static final class PersistJobContextRunnable implements Runnable {

kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobItemIncrementalTasksProgressTest.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.job.progress;
+
+import org.apache.shardingsphere.data.pipeline.core.task.progress.IncrementalTaskProgress;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class JobItemIncrementalTasksProgressTest {
+
+    @Test
+    void assertGetIncrementalPositionWithNullIncrementalTaskProgress() {
+        assertFalse(new JobItemIncrementalTasksProgress(null).getIncrementalPosition().isPresent());
+    }
+
+    @Test
+    void assertGetIncrementalPositionWithNotNullIncrementalTaskProgress() {
+        assertTrue(new JobItemIncrementalTasksProgress(mock(IncrementalTaskProgress.class, RETURNS_DEEP_STUBS)).getIncrementalPosition().isPresent());
+    }
+
+    @Test
+    void assertGetIncrementalLatestActiveTimeMillisWithNullIncrementalTaskProgress() {
+        assertThat(new JobItemIncrementalTasksProgress(null).getIncrementalLatestActiveTimeMillis(), is(0L));
+    }
+
+    @Test
+    void assertGetIncrementalLatestActiveTimeMillisWithNotNullIncrementalTaskProgress() {
+        IncrementalTaskProgress incrementalTaskProgress = mock(IncrementalTaskProgress.class, RETURNS_DEEP_STUBS);
+        when(incrementalTaskProgress.getIncrementalTaskDelay().getLatestActiveTimeMillis()).thenReturn(10L);
+        assertThat(new JobItemIncrementalTasksProgress(incrementalTaskProgress).getIncrementalLatestActiveTimeMillis(), is(10L));
+    }
+}

kernel/data-pipeline/core/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/persist/PipelineJobProgressPersistServiceTest.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.core.job.progress.persist;
+
+import lombok.SneakyThrows;
+import org.junit.jupiter.api.Test;
+import org.mockito.internal.configuration.plugins.Plugins;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+class PipelineJobProgressPersistServiceTest {
+
+    @Test
+    void assertAdd() {
+        PipelineJobProgressPersistService.add("foo_id", 1);
+        Map<String, Map<Integer, PipelineJobProgressPersistContext>> jobProgressPersistMap = getJobProgressPersistMap();
+        assertThat(jobProgressPersistMap.get("foo_id").get(1).getUnhandledEventCount().get(), is(0L));
+    }
+
+    @Test
+    void assertRemove() {
+        PipelineJobProgressPersistService.add("foo_id", 1);
+        PipelineJobProgressPersistService.remove("foo_id");
+        Map<String, Map<Integer, PipelineJobProgressPersistContext>> jobProgressPersistMap = getJobProgressPersistMap();
+        assertFalse(jobProgressPersistMap.containsKey("foo_id"));
+    }
+
+    @Test
+    void assertNotifyPersist() {
+        PipelineJobProgressPersistService.add("foo_id", 1);
+        PipelineJobProgressPersistService.notifyPersist("foo_id", 1);
+        Map<String, Map<Integer, PipelineJobProgressPersistContext>> jobProgressPersistMap = getJobProgressPersistMap();
+        assertThat(jobProgressPersistMap.get("foo_id").get(1).getUnhandledEventCount().get(), is(1L));
+    }
+
+    @SuppressWarnings("unchecked")
+    @SneakyThrows(ReflectiveOperationException.class)
+    private Map<String, Map<Integer, PipelineJobProgressPersistContext>> getJobProgressPersistMap() {
+        Field field = PipelineJobProgressPersistService.class.getDeclaredField("JOB_PROGRESS_PERSIST_MAP");
+        return (Map<String, Map<Integer, PipelineJobProgressPersistContext>>) Plugins.getMemberAccessor().get(field, PipelineJobProgressPersistService.class);
+    }
+
+    @Test
+    void assertPersistNow() {
+        PipelineJobProgressPersistService.add("foo_id", 1);
+        assertDoesNotThrow(() -> PipelineJobProgressPersistService.persistNow("foo_id", 1));
+    }
+}