Skip to content

Commit d8d656e

Browse files
oeyhoeyh
committed
Add more tests
Signed-off-by: Hai Yan <oeyh@amazon.com>
1 parent d94a81b commit d8d656e

2 files changed

Lines changed: 121 additions & 2 deletions

File tree

‎data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogClientWrapperTest.java‎

Lines changed: 67 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
package org.opensearch.dataprepper.plugins.source.rds.stream;
1111

1212
import com.github.shyiko.mysql.binlog.BinaryLogClient;
13+
import com.github.shyiko.mysql.binlog.network.AuthenticationException;
1314
import io.micrometer.core.instrument.Counter;
1415
import org.junit.jupiter.api.BeforeEach;
1516
import org.junit.jupiter.api.Test;
@@ -18,7 +19,6 @@
1819
import org.mockito.Mock;
1920
import org.mockito.junit.jupiter.MockitoExtension;
2021
import org.opensearch.dataprepper.plugins.source.rds.utils.RdsSourceAggregateMetrics;
21-
import com.github.shyiko.mysql.binlog.network.AuthenticationException;
2222

2323
import java.io.IOException;
2424
import java.util.List;
@@ -29,6 +29,9 @@
2929
import static org.mockito.Mockito.mock;
3030
import static org.mockito.Mockito.verify;
3131
import static org.mockito.Mockito.when;
32+
import static org.opensearch.dataprepper.plugins.source.rds.stream.BinlogClientWrapper.ACCESS_DENIED;
33+
import static org.opensearch.dataprepper.plugins.source.rds.stream.BinlogClientWrapper.CONNECTION_REFUSED;
34+
import static org.opensearch.dataprepper.plugins.source.rds.stream.BinlogClientWrapper.FAILED_TO_DETERMINE_BINLOG_FILENAME;
3235

3336

3437
@ExtendWith(MockitoExtension.class)
@@ -48,6 +51,18 @@ class BinlogClientWrapperTest {
4851
@Mock
4952
private Counter stream5xxErrors;
5053

54+
@Mock
55+
private Counter streamAuthErrors;
56+
57+
@Mock
58+
private Counter streamServerNotFoundErrors;
59+
60+
@Mock
61+
private Counter streamReplicationNotEnabledErrors;
62+
63+
@Mock
64+
private Counter streamAccessDeniedErrors;
65+
5166
private BinlogClientWrapper binlogClientWrapper;
5267

5368
@BeforeEach
@@ -56,6 +71,10 @@ void setUp() {
5671
lenient().when(rdsSourceAggregateMetrics.getStreamApiInvocations()).thenReturn(streamApiInvocations);
5772
lenient().when(rdsSourceAggregateMetrics.getStream4xxErrors()).thenReturn(stream4xxErrors);
5873
lenient().when(rdsSourceAggregateMetrics.getStream5xxErrors()).thenReturn(stream5xxErrors);
74+
lenient().when(rdsSourceAggregateMetrics.getStreamAuthErrors()).thenReturn(streamAuthErrors);
75+
lenient().when(rdsSourceAggregateMetrics.getStreamServerNotFoundErrors()).thenReturn(streamServerNotFoundErrors);
76+
lenient().when(rdsSourceAggregateMetrics.getStreamReplicationNotEnabledErrors()).thenReturn(streamReplicationNotEnabledErrors);
77+
lenient().when(rdsSourceAggregateMetrics.getStreamAccessDeniedErrors()).thenReturn(streamAccessDeniedErrors);
5978
}
6079

6180
@Test
@@ -66,14 +85,60 @@ void test_connect_calls_client_connect() throws IOException {
6685
}
6786

6887
@Test
69-
void test_connect_with_4xx_exception() throws IOException {
88+
void test_connect_with_4xx_auth_exception() throws IOException {
7089
doThrow(AuthenticationException.class).when(binaryLogClient).connect();
7190

7291
try {
7392
binlogClientWrapper.connect();
7493
} catch (Exception e) {
7594
verify(streamApiInvocations).increment();
7695
verify(stream4xxErrors).increment();
96+
verify(streamAuthErrors).increment();
97+
}
98+
}
99+
100+
@Test
101+
void test_connect_with_4xx_server_not_found_exception() throws IOException {
102+
Exception connectionRefusedException = new IOException(
103+
"Failed to connect to MySQL server",
104+
new Exception(CONNECTION_REFUSED)
105+
);
106+
doThrow(connectionRefusedException).when(binaryLogClient).connect();
107+
108+
try {
109+
binlogClientWrapper.connect();
110+
} catch (Exception e) {
111+
verify(streamApiInvocations).increment();
112+
verify(stream4xxErrors).increment();
113+
verify(streamServerNotFoundErrors).increment();
114+
}
115+
}
116+
117+
@Test
118+
void test_connect_with_4xx_binlog_exception() throws IOException {
119+
final Exception binlogNotEnabledException = new IOException(FAILED_TO_DETERMINE_BINLOG_FILENAME);
120+
doThrow(binlogNotEnabledException).when(binaryLogClient).connect();
121+
122+
try {
123+
binlogClientWrapper.connect();
124+
} catch (Exception e) {
125+
verify(streamApiInvocations).increment();
126+
verify(stream4xxErrors).increment();
127+
verify(streamReplicationNotEnabledErrors).increment();
128+
}
129+
}
130+
131+
@Test
132+
void test_connect_with_4xx_access_exception() throws IOException {
133+
final Exception accessDeniedException = new IOException(ACCESS_DENIED);
134+
doThrow(accessDeniedException).when(binaryLogClient).connect();
135+
136+
try {
137+
binlogClientWrapper.connect();
138+
} catch (Exception e) {
139+
verify(streamApiInvocations).increment();
140+
verify(stream4xxErrors).increment();
141+
verify(streamAccessDeniedErrors).increment();
77142
}
78143
}
79144

‎data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/stream/LogicalReplicationClientTest.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,9 @@
4242
import static org.mockito.Mockito.verifyNoMoreInteractions;
4343
import static org.mockito.Mockito.when;
4444
import static org.opensearch.dataprepper.plugins.source.rds.stream.LogicalReplicationClient.AUTHENTICATION_FAILED;
45+
import static org.opensearch.dataprepper.plugins.source.rds.stream.LogicalReplicationClient.CONNECTION_REFUSED;
46+
import static org.opensearch.dataprepper.plugins.source.rds.stream.LogicalReplicationClient.PERMISSION_DENIED;
47+
import static org.opensearch.dataprepper.plugins.source.rds.stream.LogicalReplicationClient.REPLICATION_SLOT_DOES_NOT_EXIST;
4548

4649
@ExtendWith(MockitoExtension.class)
4750
class LogicalReplicationClientTest {
@@ -64,6 +67,18 @@ class LogicalReplicationClientTest {
6467
@Mock
6568
private Counter stream5xxErrors;
6669

70+
@Mock
71+
private Counter streamAuthErrors;
72+
73+
@Mock
74+
private Counter streamServerNotFoundErrors;
75+
76+
@Mock
77+
private Counter streamReplicationNotEnabledErrors;
78+
79+
@Mock
80+
private Counter streamAccessDeniedErrors;
81+
6782
private String publicationName;
6883
private String replicationSlotName;
6984
private LogicalReplicationClient logicalReplicationClient;
@@ -77,6 +92,10 @@ void setUp() {
7792
lenient().when(rdsSourceAggregateMetrics.getStreamApiInvocations()).thenReturn(streamApiInvocations);
7893
lenient().when(rdsSourceAggregateMetrics.getStream4xxErrors()).thenReturn(stream4xxErrors);
7994
lenient().when(rdsSourceAggregateMetrics.getStream5xxErrors()).thenReturn(stream5xxErrors);
95+
lenient().when(rdsSourceAggregateMetrics.getStreamAuthErrors()).thenReturn(streamAuthErrors);
96+
lenient().when(rdsSourceAggregateMetrics.getStreamServerNotFoundErrors()).thenReturn(streamServerNotFoundErrors);
97+
lenient().when(rdsSourceAggregateMetrics.getStreamReplicationNotEnabledErrors()).thenReturn(streamReplicationNotEnabledErrors);
98+
lenient().when(rdsSourceAggregateMetrics.getStreamAccessDeniedErrors()).thenReturn(streamAccessDeniedErrors);
8099
}
81100

82101
@Test
@@ -117,6 +136,41 @@ void test_connect_auth_exception_should_throw() throws SQLException {
117136
assertThrows(RuntimeException.class, () -> logicalReplicationClient.connect());
118137
verify(streamApiInvocations).increment();
119138
verify(stream4xxErrors).increment();
139+
verify(streamAuthErrors).increment();
140+
}
141+
142+
@Test
143+
void test_connect_server_exception_should_throw() throws SQLException {
144+
Exception connectionRefusedException = new RuntimeException(
145+
"Failed to establish connection",
146+
new Exception(CONNECTION_REFUSED)
147+
);
148+
when(connectionManager.getConnection()).thenThrow(connectionRefusedException);
149+
150+
assertThrows(RuntimeException.class, () -> logicalReplicationClient.connect());
151+
verify(streamApiInvocations).increment();
152+
verify(stream4xxErrors).increment();
153+
verify(streamServerNotFoundErrors).increment();
154+
}
155+
156+
@Test
157+
void test_connect_log_exception_should_throw() throws SQLException {
158+
when(connectionManager.getConnection()).thenThrow(new RuntimeException(REPLICATION_SLOT_DOES_NOT_EXIST));
159+
160+
assertThrows(RuntimeException.class, () -> logicalReplicationClient.connect());
161+
verify(streamApiInvocations).increment();
162+
verify(stream4xxErrors).increment();
163+
verify(streamReplicationNotEnabledErrors).increment();
164+
}
165+
166+
@Test
167+
void test_connect_access_exception_should_throw() throws SQLException {
168+
when(connectionManager.getConnection()).thenThrow(new RuntimeException(PERMISSION_DENIED));
169+
170+
assertThrows(RuntimeException.class, () -> logicalReplicationClient.connect());
171+
verify(streamApiInvocations).increment();
172+
verify(stream4xxErrors).increment();
173+
verify(streamAccessDeniedErrors).increment();
120174
}
121175

122176
@Test

0 commit comments

Comments
 (0)