Skip to content

Commit 7002f06

Browse files
george0stgeorge0st
authored
Merge pull request #84 from george0st/change
Extend unit tests (add validation after sequence write)
2 parents 1101598 + 46d0a9e commit 7002f06

File tree

7 files changed

+311
-50
lines changed

7 files changed

+311
-50
lines changed

‎nifi/cql-processor/docs/todo-list.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
11
TO-DO list
2+
==========
3+
4+
1. Extend checks for Write/Validate
5+
- check count write and read
6+
- generate exception in case of null value for read-validate
27

3-
1. Add to the perf test, check of writed values
48

‎nifi/cql-processor/nifi-cql-processors/src/test/java/org/george0st/processors/cql/PutCQLBase.java‎

Lines changed: 52 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@
1010
import org.george0st.processors.cql.helper.CqlTestSchema;
1111
import org.george0st.processors.cql.helper.ReadableValue;
1212
import org.george0st.processors.cql.helper.TestSetup;
13+
import org.george0st.processors.cql.processor.CsvCqlValidate;
14+
import org.junit.jupiter.api.AfterAll;
15+
import org.junit.jupiter.api.AfterEach;
1316
import org.junit.jupiter.api.BeforeEach;
1417

1518
import java.io.IOException;
@@ -79,11 +82,34 @@ public void init() throws IOException, InterruptedException, InitializationExcep
7982
}
8083
}
8184

82-
protected FlowFile coreTest(TestSetup setup, String content) {
83-
return coreTest(setup, content, null, null);
85+
@AfterEach
86+
public void Close() throws InterruptedException {
87+
for (TestSetup setup: setups) {
88+
setup.setProperty(testRunner, testService);
89+
testRunner.enableControllerService(testService);
90+
91+
// build schema, if needed
92+
try (CqlSession session=testService.getSession()) {
93+
CqlTestSchema schema = new CqlTestSchema(session, setup);
94+
schema.cleanData();
95+
}
96+
testRunner.disableControllerService(testService);
97+
}
8498
}
8599

86-
protected FlowFile coreTest(TestSetup setup, String content, PropertyDescriptor property, String propertyValue){
100+
protected FlowFile runTest(TestSetup setup, String content) {
101+
return runTestWithProperty(setup, content, null, null, false);
102+
}
103+
104+
protected FlowFile runTest(TestSetup setup, String content, boolean validate) {
105+
return runTestWithProperty(setup, content, null, null, validate);
106+
}
107+
108+
protected FlowFile runTestWithProperty(TestSetup setup, String content, PropertyDescriptor property, String propertyValue){
109+
return runTestWithProperty(setup, content, property, propertyValue, false);
110+
}
111+
112+
protected FlowFile runTestWithProperty(TestSetup setup, String content, PropertyDescriptor property, String propertyValue, boolean validate){
87113
HashMap<String, String> attributes = new HashMap<String, String>(Map.of("CQLName",setup.name));
88114
FlowFile result;
89115

@@ -92,12 +118,12 @@ protected FlowFile coreTest(TestSetup setup, String content, PropertyDescriptor
92118
if (property != null)
93119
setup.setProperty(testRunner, property, propertyValue);
94120
testRunner.enableControllerService(testService);
95-
result = coreTest();
121+
result = coreTest(setup, content, validate);
96122
testRunner.disableControllerService(testService);
97123
return result;
98124
}
99125

100-
private FlowFile coreTest(){
126+
private FlowFile coreTest(TestSetup setup, String content, boolean validate){
101127
try {
102128
long finish, start, count;
103129
FlowFile result;
@@ -111,14 +137,34 @@ private FlowFile coreTest(){
111137

112138
if (ok) {
113139
count = Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT));
114-
System.out.printf("SetupName: '%s'; '%s': %s (%d ms); Items: %d; Perf: %.1f [calls/sec]%s",
140+
System.out.printf("Source: '%s'; WRITE; '%s': %s (%d ms); Items: %d; Perf: %.1f [calls/sec]%s",
115141
result.getAttribute("CQLName"),
116142
"FlowFile",
117143
ReadableValue.fromMillisecond(finish - start),
118144
finish - start,
119145
count,
120146
count / ((finish - start) / 1000.0),
121147
System.lineSeparator());
148+
149+
if (validate) {
150+
// delay (before read for synch on CQL side)
151+
Thread.sleep(3000);
152+
153+
// validate (read value from CSV and from CQL and compare content)
154+
try (CqlSession session=testService.getSession()) {
155+
start = System.currentTimeMillis();
156+
count = (new CsvCqlValidate(session, setup, CqlTestSchema.primaryKeys)).executeContent(content);
157+
finish = System.currentTimeMillis();
158+
}
159+
System.out.printf("Source: '%s'; VALIDATE; '%s': %s (%d ms); Items: %d; Perf: %.1f [calls/sec]%s",
160+
result.getAttribute("CQLName"),
161+
"FlowFile",
162+
ReadableValue.fromMillisecond(finish - start),
163+
finish - start,
164+
count,
165+
count / ((finish - start) / 1000.0),
166+
System.lineSeparator());
167+
}
122168
return result;
123169
}
124170
return null;

‎nifi/cql-processor/nifi-cql-processors/src/test/java/org/george0st/processors/cql/PutCQLFunction.java‎

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,13 @@
1818

1919
import org.apache.nifi.flowfile.FlowFile;
2020
import org.apache.nifi.reporting.InitializationException;
21-
import org.apache.nifi.util.TestRunners;
22-
import org.george0st.cql.CQLControllerService;
23-
import org.george0st.processors.cql.helper.ReadableValue;
2421
import org.george0st.processors.cql.helper.TestSetup;
25-
import org.junit.jupiter.api.BeforeEach;
2622
import org.junit.jupiter.api.Test;
2723

2824
import static org.junit.jupiter.api.Assertions.*;
2925
import static org.junit.jupiter.api.Assumptions.assumeTrue;
3026

3127
import java.io.IOException;
32-
import java.util.*;
3328

3429
public class PutCQLFunction extends PutCQLBase {
3530
// Helper
@@ -48,7 +43,7 @@ public void testBasic() {
4843
FlowFile result;
4944

5045
for (TestSetup setup : setups) {
51-
result = coreTest(setup, content);
46+
result = runTest(setup, content);
5247
// check amount of write items
5348
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
5449
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -65,7 +60,7 @@ public void testBatchLoggedTypes() {
6560
FlowFile result;
6661

6762
for (TestSetup setup : setups) {
68-
result = coreTest(setup, content, PutCQL.BATCH_TYPE, PutCQL.BT_LOGGED.getValue());
63+
result = runTestWithProperty(setup, content, PutCQL.BATCH_TYPE, PutCQL.BT_LOGGED.getValue());
6964
// check amount of write items
7065
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
7166
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -82,7 +77,7 @@ public void testBatchUnLoggedTypes() {
8277
FlowFile result;
8378

8479
for (TestSetup setup : setups) {
85-
result = coreTest(setup, content, PutCQL.BATCH_TYPE, PutCQL.BT_LOGGED.getValue());
80+
result = runTestWithProperty(setup, content, PutCQL.BATCH_TYPE, PutCQL.BT_LOGGED.getValue());
8681
// check amount of write items
8782
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
8883
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -107,12 +102,12 @@ public void testMoreItems() {
107102
FlowFile result;
108103

109104
for (TestSetup setup: setups) {
110-
result = coreTest(setup, content);
105+
result = runTest(setup, content);
111106
// check amount of write items
112107
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
113108
assertEquals(5, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
114109

115-
result = coreTest(setup, content2);
110+
result = runTest(setup, content2);
116111
// check amount of write items
117112
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
118113
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -129,7 +124,7 @@ public void testNoQuotas() {
129124
FlowFile result;
130125

131126
for (TestSetup setup: setups) {
132-
result = coreTest(setup, content);
127+
result = runTest(setup, content);
133128
// check amount of write items
134129
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
135130
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -142,7 +137,7 @@ public void testEmptyInput() {
142137
FlowFile result;
143138

144139
for (TestSetup setup: setups) {
145-
result = coreTest(setup, content);
140+
result = runTest(setup, content);
146141
// check amount of write items
147142
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
148143
assertEquals(0, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -155,7 +150,7 @@ public void testOnlyHeader() {
155150
FlowFile result;
156151

157152
for (TestSetup setup: setups) {
158-
result = coreTest(setup, content);
153+
result = runTest(setup, content);
159154
// check amount of write items
160155
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
161156
assertEquals(0, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -168,7 +163,7 @@ public void testOnlyHeaderNoQuotas() {
168163
FlowFile result;
169164

170165
for (TestSetup setup: setups) {
171-
result = coreTest(setup, content);
166+
result = runTest(setup, content);
172167
// check amount of write items
173168
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
174169
assertEquals(0, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -186,7 +181,7 @@ public void testBasicRepeat5() {
186181

187182
for (TestSetup setup: setups) {
188183
for (int i = 0; i < 5; i++) {
189-
result = coreTest(setup, content);
184+
result = runTest(setup, content);
190185
// check amount of write items
191186
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
192187
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));
@@ -204,7 +199,7 @@ public void testNoQuotas2() {
204199
FlowFile result;
205200

206201
for (TestSetup setup: setups) {
207-
result = coreTest(setup, content);
202+
result = runTest(setup, content);
208203
// check amount of write items
209204
assertNotNull(result, String.format("Issue with processing in '%s'", setup.name));
210205
assertEquals(4, Long.parseLong(result.getAttribute(PutCQL.ATTRIBUTE_COUNT)));

0 commit comments

Comments
 (0)