Skip to content

Commit 3950d10

Browse files
author
Aditya Pratap Singh
committed
added some unit tests for gobblin temporal module
1 parent 72306bb commit 3950d10

File tree

5 files changed

+293
-8
lines changed

5 files changed

+293
-8
lines changed

‎gobblin-temporal/build.gradle‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,11 +67,14 @@ dependencies {
6767
testCompile project(":gobblin-example")
6868

6969
testCompile externalDependency.testng
70-
testCompile externalDependency.mockito
70+
testCompile externalDependency.mockitoInline
71+
testCompile externalDependency.powerMockApi
72+
testCompile externalDependency.powerMockModule
7173
testCompile externalDependency.hadoopYarnMiniCluster
7274
testCompile externalDependency.curatorFramework
7375
testCompile externalDependency.curatorTest
7476

77+
7578
testCompile ('com.google.inject:guice:3.0') {
7679
force = true
7780
}

‎gobblin-temporal/src/main/java/org/apache/gobblin/temporal/util/nesting/workflow/AbstractNestingExecWorkflowImpl.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@
3434
import io.temporal.workflow.Workflow;
3535
import lombok.extern.slf4j.Slf4j;
3636

37+
import com.google.common.annotations.VisibleForTesting;
38+
39+
3740
import org.apache.gobblin.temporal.util.nesting.work.WorkflowAddr;
3841
import org.apache.gobblin.temporal.util.nesting.work.Workload;
3942

@@ -115,8 +118,9 @@ protected NestingExecWorkflow<WORK_ITEM> createChildWorkflow(final WorkflowAddr
115118
return Workflow.newChildWorkflowStub(NestingExecWorkflow.class, childOpts);
116119
}
117120

121+
@VisibleForTesting
118122
/** @return how long to pause prior to creating a child workflow, based on `numDirectLeavesChildMayHave` */
119-
protected Duration calcPauseDurationBeforeCreatingSubTree(int numDirectLeavesChildMayHave) {
123+
public Duration calcPauseDurationBeforeCreatingSubTree(int numDirectLeavesChildMayHave) {
120124
// (only pause when an appreciable number of leaves)
121125
// TODO: use a configuration value, for simpler adjustment, rather than hard-code
122126
return numDirectLeavesChildMayHave > MAX_CHILD_SUB_TREE_LEAVES_BEFORE_SHOULD_PAUSE_DEFAULT
@@ -130,11 +134,9 @@ protected Duration calcPauseDurationBeforeCreatingSubTree(int numDirectLeavesChi
130134
* List<Integer> naiveUniformity = Collections.nCopies(numSubTreesPerSubTree, numSubTreeChildren);
131135
* @return each sub-tree's desired size, in ascending sub-tree order
132136
*/
133-
protected static List<Integer> consolidateSubTreeGrandChildren(
134-
final int numSubTreesPerSubTree,
135-
final int numChildrenTotal,
136-
final int numSubTreeChildren
137-
) {
137+
@VisibleForTesting
138+
public static List<Integer> consolidateSubTreeGrandChildren(final int numSubTreesPerSubTree,
139+
final int numChildrenTotal, final int numSubTreeChildren) {
138140
if (numSubTreesPerSubTree <= 0) {
139141
return Lists.newArrayList();
140142
} else if (isSqrt(numSubTreeChildren, numChildrenTotal)) {

‎gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/activity/impl/GenerateWorkUnitsImplTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,4 +94,4 @@ public void testFetchesUniqueWorkDirsFromMultiWorkUnits() {
9494
Set<String> output = GenerateWorkUnitsImpl.calculateWorkDirsToCleanup(workUnitStream);
9595
Assert.assertEquals(output.size(), 11);
9696
}
97-
}
97+
}

‎gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/utils/JobStateUtilTest.java‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.gobblin.temporal.ddm.utils;
18+
19+
import java.io.DataOutputStream;
20+
import java.io.IOException;
21+
import java.net.URI;
22+
import java.net.URISyntaxException;
23+
import java.util.Properties;
24+
25+
import org.mockito.Mockito;
26+
import org.testng.Assert;
27+
import org.testng.annotations.BeforeMethod;
28+
import org.testng.annotations.Test;
29+
import org.apache.hadoop.fs.FSDataOutputStream;
30+
import org.apache.hadoop.fs.FileSystem;
31+
import org.apache.hadoop.fs.Path;
32+
33+
import org.apache.gobblin.runtime.JobState;
34+
import org.apache.gobblin.runtime.TaskState;
35+
import org.apache.gobblin.source.Source;
36+
import org.apache.gobblin.metastore.StateStore;
37+
import org.apache.gobblin.temporal.ddm.util.JobStateUtils;
38+
39+
40+
public class JobStateUtilTest {
41+
42+
private JobState jobState;
43+
private FileSystem fileSystem;
44+
45+
@BeforeMethod
46+
public void setUp() {
47+
jobState = Mockito.mock(JobState.class);
48+
fileSystem = Mockito.mock(FileSystem.class);
49+
}
50+
51+
@Test
52+
public void testOpenFileSystem() throws IOException {
53+
54+
Mockito.when(jobState.getProp(Mockito.anyString(), Mockito.anyString())).thenReturn("file:///test");
55+
Mockito.when(jobState.getProperties()).thenReturn(new Properties());
56+
57+
FileSystem fs = JobStateUtils.openFileSystem(jobState);
58+
59+
Assert.assertNotNull(fs);
60+
Mockito.verify(jobState,Mockito.times(1)).getProp(Mockito.anyString(), Mockito.anyString());
61+
}
62+
63+
@Test
64+
public void testCreateSource() throws ReflectiveOperationException {
65+
Mockito.when(jobState.getProp(Mockito.anyString()))
66+
.thenReturn("org.apache.gobblin.source.extractor.filebased.TextFileBasedSource");
67+
Source<?, ?> source = JobStateUtils.createSource(jobState);
68+
Assert.assertNotNull(source);
69+
}
70+
71+
@Test
72+
public void testOpenTaskStateStoreUncached() throws URISyntaxException {
73+
Mockito.when(jobState.getProp(Mockito.anyString())).thenReturn("file:///test");
74+
Mockito.when(jobState.getJobId()).thenReturn("testJobId");
75+
Mockito.when(jobState.getJobName()).thenReturn("testJobName");
76+
Mockito.when(fileSystem.makeQualified(Mockito.any()))
77+
.thenReturn(new Path("file:///test/testJobName/testJobId/output"));
78+
Mockito.when(fileSystem.getUri()).thenReturn(new URI("file:///test/testJobName/testJobId/output"));
79+
80+
StateStore<TaskState> stateStore = JobStateUtils.openTaskStateStoreUncached(jobState, fileSystem);
81+
82+
Assert.assertNotNull(stateStore);
83+
}
84+
85+
@Test
86+
public void testGetFileSystemUri() {
87+
Mockito.when(jobState.getProp(Mockito.anyString(), Mockito.anyString())).thenReturn("file:///test");
88+
URI fsUri = JobStateUtils.getFileSystemUri(jobState);
89+
Assert.assertEquals(URI.create("file:///test"), fsUri);
90+
Mockito.verify(jobState).getProp(Mockito.anyString(), Mockito.anyString());
91+
}
92+
93+
@Test
94+
public void testGetWorkDirRoot() {
95+
Mockito.when(jobState.getProp(Mockito.anyString())).thenReturn("/tmp");
96+
Mockito.when(jobState.getJobName()).thenReturn("testJob");
97+
Mockito.when(jobState.getJobId()).thenReturn("jobId123");
98+
Path rootPath = JobStateUtils.getWorkDirRoot(jobState);
99+
Assert.assertEquals(new Path("/tmp/testJob/jobId123"), rootPath);
100+
Mockito.verify(jobState, Mockito.times(1)).getProp(Mockito.anyString());
101+
}
102+
103+
@Test
104+
public void testGetWorkUnitsPath() {
105+
Mockito.when(jobState.getProp(Mockito.anyString())).thenReturn("/tmp");
106+
Mockito.when(jobState.getJobName()).thenReturn("testJob");
107+
Mockito.when(jobState.getJobId()).thenReturn("jobId123");
108+
Path workUnitsPath = JobStateUtils.getWorkUnitsPath(jobState);
109+
Assert.assertEquals(new Path("/tmp/testJob/jobId123/input"), workUnitsPath);
110+
}
111+
112+
@Test
113+
public void testGetTaskStateStorePath() throws IOException {
114+
Mockito.when(fileSystem.makeQualified(Mockito.any(Path.class))).thenReturn(new Path("/qualified/path"));
115+
Mockito.when(jobState.getProp(Mockito.anyString())).thenReturn("/tmp");
116+
Mockito.when(jobState.getJobName()).thenReturn("testJob");
117+
Mockito.when(jobState.getJobId()).thenReturn("jobId123");
118+
Path taskStateStorePath = JobStateUtils.getTaskStateStorePath(jobState, fileSystem);
119+
Assert.assertEquals(new Path("/qualified/path"), taskStateStorePath);
120+
}
121+
122+
@Test
123+
public void testWriteJobState() throws IOException {
124+
Path workDirRootPath = new Path("/tmp");
125+
FSDataOutputStream dos = Mockito.mock(FSDataOutputStream.class);
126+
Mockito.when(fileSystem.create(Mockito.any(Path.class))).thenReturn(dos);
127+
128+
JobStateUtils.writeJobState(jobState, workDirRootPath, fileSystem);
129+
130+
Mockito.verify(fileSystem).create(Mockito.any(Path.class));
131+
Mockito.verify(jobState).write(Mockito.any(DataOutputStream.class), Mockito.anyBoolean(), Mockito.anyBoolean());
132+
}
133+
134+
@Test
135+
public void testGetSharedResourcesBroker() {
136+
Mockito.when(jobState.getProperties()).thenReturn(System.getProperties());
137+
Mockito.when(jobState.getJobName()).thenReturn("testJob");
138+
Mockito.when(jobState.getJobId()).thenReturn("jobId123");
139+
Assert.assertNotNull(JobStateUtils.getSharedResourcesBroker(jobState));
140+
}
141+
}

‎gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/workflow/impl/AbstractNestingExecWorkflowImplTest.java‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.gobblin.temporal.ddm.workflow.impl;
19+
20+
import java.time.Duration;
21+
import java.util.List;
22+
import java.util.Optional;
23+
24+
import org.junit.runner.RunWith;
25+
import org.mockito.Mock;
26+
import org.mockito.Mockito;
27+
import org.powermock.core.classloader.annotations.PrepareForTest;
28+
import org.powermock.modules.junit4.PowerMockRunner;
29+
import org.testng.Assert;
30+
import org.testng.annotations.BeforeClass;
31+
import org.testng.annotations.Test;
32+
33+
import io.temporal.workflow.Async;
34+
import io.temporal.workflow.Promise;
35+
import io.temporal.workflow.Workflow;
36+
37+
import org.apache.gobblin.temporal.util.nesting.work.WorkflowAddr;
38+
import org.apache.gobblin.temporal.util.nesting.work.Workload;
39+
import org.apache.gobblin.temporal.util.nesting.workflow.AbstractNestingExecWorkflowImpl;
40+
41+
42+
@RunWith(PowerMockRunner.class)
43+
@PrepareForTest(Workflow.class)
44+
public class AbstractNestingExecWorkflowImplTest {
45+
46+
@Mock
47+
private Workload<String> mockWorkload;
48+
49+
@Mock
50+
private WorkflowAddr mockWorkflowAddr;
51+
52+
@Mock
53+
private Workload.WorkSpan<String> mockWorkSpan;
54+
55+
@Mock
56+
private Promise<Object> mockPromise;
57+
58+
private AbstractNestingExecWorkflowImpl<String, Object> workflow;
59+
60+
@BeforeClass
61+
public void setup() {
62+
// PowerMockito is required to mock static methods in the Workflow class
63+
Mockito.mockStatic(Workflow.class);
64+
Mockito.mockStatic(Async.class);
65+
Mockito.mockStatic(Promise.class);
66+
this.mockWorkload = Mockito.mock(Workload.class);
67+
this.mockWorkflowAddr = Mockito.mock(WorkflowAddr.class);
68+
this.mockWorkSpan = Mockito.mock(Workload.WorkSpan.class);
69+
this.mockPromise = Mockito.mock(Promise.class);
70+
71+
workflow = new AbstractNestingExecWorkflowImpl<String, Object>() {
72+
@Override
73+
protected Promise<Object> launchAsyncActivity(String task) {
74+
return mockPromise;
75+
}
76+
};
77+
}
78+
79+
@Test
80+
public void testPerformWorkload_NoWorkSpan() {
81+
// Arrange
82+
Mockito.when(mockWorkload.getSpan(Mockito.anyInt(), Mockito.anyInt())).thenReturn(Optional.empty());
83+
84+
// Act
85+
int result = workflow.performWorkload(mockWorkflowAddr, mockWorkload, 0, 10, 5, Optional.empty());
86+
87+
// Assert
88+
Assert.assertEquals(0, result);
89+
Mockito.verify(mockWorkload, Mockito.times(2)).getSpan(0, 5);
90+
}
91+
92+
@Test
93+
public void testCalcPauseDurationBeforeCreatingSubTree_NoPause() {
94+
// Act
95+
Duration result = workflow.calcPauseDurationBeforeCreatingSubTree(50);
96+
97+
// Assert
98+
Assert.assertEquals(Duration.ZERO, result);
99+
}
100+
101+
@Test
102+
public void testCalcPauseDurationBeforeCreatingSubTree_PauseRequired() {
103+
// Act
104+
Duration result = workflow.calcPauseDurationBeforeCreatingSubTree(150);
105+
106+
// Assert
107+
Assert.assertEquals(
108+
Duration.ofSeconds(AbstractNestingExecWorkflowImpl.NUM_SECONDS_TO_PAUSE_BEFORE_CREATING_SUB_TREE_DEFAULT),
109+
result);
110+
}
111+
112+
@Test
113+
public void testConsolidateSubTreeGrandChildren() {
114+
// Act
115+
List<Integer> result = AbstractNestingExecWorkflowImpl.consolidateSubTreeGrandChildren(3, 10, 2);
116+
117+
// Assert
118+
Assert.assertEquals(3, result.size());
119+
Assert.assertEquals(Integer.valueOf(0), result.get(0));
120+
Assert.assertEquals(Integer.valueOf(0), result.get(1));
121+
Assert.assertEquals(Integer.valueOf(6), result.get(2));
122+
}
123+
124+
@Test(expectedExceptions = AssertionError.class)
125+
public void testPerformWorkload_LaunchesChildWorkflows() {
126+
// Arrange
127+
Mockito.when(mockWorkload.getSpan(Mockito.anyInt(), Mockito.anyInt())).thenReturn(Optional.of(mockWorkSpan));
128+
Mockito.when(mockWorkSpan.getNumElems()).thenReturn(5);
129+
Mockito.when(mockWorkSpan.next()).thenReturn("task1");
130+
Mockito.when(mockWorkload.isIndexKnownToExceed(Mockito.anyInt())).thenReturn(false);
131+
132+
// Mock the child workflow
133+
Mockito.when(Async.function(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.anyInt(),
134+
Mockito.anyInt(), Mockito.any())).thenReturn(mockPromise);
135+
Mockito.when(mockPromise.get()).thenReturn(5);
136+
// Act
137+
int result = workflow.performWorkload(mockWorkflowAddr, mockWorkload, 0, 10, 5, Optional.empty());
138+
}
139+
}

0 commit comments

Comments
 (0)